NPC Pairing Solver

pairing_solver.py

#!/usr/bin/env python3
"""
CP-SAT solver (relaxed):
- 4 rounds
- groups of 4 players (inferred from N players)
- each player occupies each position (seat 1..4) exactly once across rounds
- no same-team players in same group
- each player must meet players from at least 7 of the other teams (of the other T-1 teams)
- allow up to one repeated pairing per player (i.e., a player may have at most one other player they meet >=2 times),
  and any pair is capped at meeting at most 2 times
Outputs CSV with columns: Round,Group,Position,Player,Team,Seed
"""
import argparse
import csv
import sys
from collections import defaultdict
from ortools.sat.python import cp_model

def read_players(csv_path):
    rows = []
    with open(csv_path, newline='', encoding='utf-8-sig') as f:
        reader = csv.DictReader(f)
        if reader.fieldnames is None:
            raise RuntimeError("CSV missing headers")
        # detect columns
        name_key = None
        team_key = None
        seed_key = None
        for h in reader.fieldnames:
            lh = h.lower()
            if any(x in lh for x in ('player','name')):
                name_key = h
            if 'team' in lh:
                team_key = h
            if 'seed' in lh:
                seed_key = h
        for r in reader:
            if name_key is None:
                name = r[reader.fieldnames[0]]
            else:
                name = r[name_key]
            team = r[team_key] if team_key is not None else r[reader.fieldnames[1]]
            seed = r[seed_key] if seed_key is not None else ''
            rows.append({'name': name.strip(), 'team': team.strip(), 'seed': seed})
    return rows

def main():
    parser = argparse.ArgumentParser(description="Relaxed Groupings CP-SAT solver")
    parser.add_argument('--input', '-i', required=True, help='Input CSV file path')
    parser.add_argument('--output', '-o', default='schedule.csv', help='Output CSV path')
    parser.add_argument('--time_limit', '-t', type=int, default=300, help='Time limit seconds for solver')
    parser.add_argument('--workers', '-w', type=int, default=8, help='Solver worker threads')
    args = parser.parse_args()

    players = read_players(args.input)
    n = len(players)
    if n % 4 != 0:
        print(f"ERROR: number of players ({n}) is not divisible by 4.", file=sys.stderr)
        sys.exit(1)
    R = 4
    K = 4
    G = n // K

    # indexes
    player_ids = list(range(n))
    teams = [p['team'] for p in players]
    team_to_players = defaultdict(list)
    for i,p in enumerate(players):
        team_to_players[p['team']].append(i)
    unique_teams = list(team_to_players.keys())
    num_teams = len(unique_teams)

    print(f"Parsed {n} players, {num_teams} unique teams, {G} groups per round.")
    for t in sorted(team_to_players.keys()):
        print(f"- Team {t!r}: {len(team_to_players[t])} players")

    model = cp_model.CpModel()

    # y[p,g,k,r] = 1 if player p is assigned to group g, position k (seat 0..3) in round r
    y = {}
    for p in player_ids:
        for r in range(R):
            for g in range(G):
                for k in range(K):
                    y[(p,g,k,r)] = model.NewBoolVar(f"y_p{p}_g{g}_k{k}_r{r}")

    # 1) each player exactly one position per round
    for p in player_ids:
        for r in range(R):
            model.Add(sum(y[(p,g,k,r)] for g in range(G) for k in range(K)) == 1)

    # 2) each (g,k,r) exactly one player
    for r in range(R):
        for g in range(G):
            for k in range(K):
                model.Add(sum(y[(p,g,k,r)] for p in player_ids) == 1)

    # 3) each player occupies each seat exactly once across rounds
    for p in player_ids:
        for k in range(K):
            model.Add(sum(y[(p,g,k,r)] for g in range(G) for r in range(R)) == 1)

    # 4) no same-team players in same group (for each group g and round r, at most one player from each team)
    for r in range(R):
        for g in range(G):
            for team in unique_teams:
                members = team_to_players[team]
                if len(members) <= 1:
                    continue
                model.Add(sum(y[(p,g,k,r)] for p in members for k in range(K)) <= 1)

    # 5) pair variables: b[p,q,g,r] = 1 iff p and q both in same group g in round r
    # only create for p<q
    b = {}
    for p in player_ids:
        for q in range(p+1, n):
            for r in range(R):
                for g in range(G):
                    b[(p,q,g,r)] = model.NewBoolVar(f"b_p{p}_q{q}_g{g}_r{r}")
                    # link: b <= presence of p in g,r and b <= presence of q in g,r
                    model.Add(b[(p,q,g,r)] <= sum(y[(p,g,k,r)] for k in range(K)))
                    model.Add(b[(p,q,g,r)] <= sum(y[(q,g,k,r)] for k in range(K)))
                    model.Add(b[(p,q,g,r)] >= sum(y[(p,g,k,r)] for k in range(K)) + sum(y[(q,g,k,r)] for k in range(K)) - 1)

    # RELAXATION: allow up to one repeated pairing per player.
    # For each unordered pair (p,q) we create meetings_pq = sum(b[p,q,g,r]) (0..R)
    # - cap meetings_pq <= 2 (don't allow triples)
    # - create rrep_pq boolean = 1 iff meetings_pq >= 2
    # - For each player p, sum of rrep_pq over q != p <= 1
    meetings = {}
    rrep = {}
    for p in player_ids:
        for q in range(p+1, n):
            meetings[(p,q)] = model.NewIntVar(0, R, f"meetings_p{p}_q{q}")
            model.Add(meetings[(p,q)] == sum(b[(p,q,g,r)] for r in range(R) for g in range(G)))
            # cap a pair meeting at most twice
            model.Add(meetings[(p,q)] <= 2)
            rrep[(p,q)] = model.NewBoolVar(f"rrep_p{p}_q{q}")
            # rrep == 1 <=> meetings >= 2
            model.Add(meetings[(p,q)] >= 2).OnlyEnforceIf(rrep[(p,q)])
            model.Add(meetings[(p,q)] <= 1).OnlyEnforceIf(rrep[(p,q)].Not())

    # For each player p, allow at most one repeated partner
    for p in player_ids:
        reps_for_p = []
        for q in range(0, p):
            reps_for_p.append(rrep[(q,p)])
        for q in range(p+1, n):
            reps_for_p.append(rrep[(p,q)])
        if reps_for_p:
            model.Add(sum(reps_for_p) <= 1)

    # RELAXATION: must meet at least 7 of the other teams (instead of all other teams)
    # For each player p and other team T', create team_met[p,T'] boolean if p meets that team at least once
    team_met = {}
    for p in player_ids:
        team_p = players[p]['team']
        for team_other in unique_teams:
            if team_other == team_p:
                continue
            members = team_to_players[team_other]
            if not members:
                continue
            team_met[(p,team_other)] = model.NewBoolVar(f"teammet_p{p}_team_{team_other}")
            # sum of b variables between p and members across rounds/groups
            sum_b = []
            for q in members:
                # ensure using ordered keys into b
                a,bidx = (q,p) if q < p else (p,q)
                sum_b.extend([b[(a,bidx,g,r)] for r in range(R) for g in range(G)])
            model.Add(sum(sum_b) >= 1).OnlyEnforceIf(team_met[(p,team_other)])
            model.Add(sum(sum_b) == 0).OnlyEnforceIf(team_met[(p,team_other)].Not())

        # Now require p meets at least 7 distinct other teams
        other_team_bools = [team_met[(p,team_other)] for team_other in unique_teams if team_other != players[p]['team']]
        # if number of other teams is less than 7, this is automatically impossible; we keep the inequality
        model.Add(sum(other_team_bools) >= max(0, min(len(other_team_bools), 7)))

    # Note: we no longer enforce global "no pair appears together more than once" constraint.

    # Solver
    solver = cp_model.CpSolver()
    solver.parameters.max_time_in_seconds = args.time_limit
    solver.parameters.num_search_workers = args.workers
    solver.parameters.random_seed = 1
    solver.parameters.log_search_progress = False

    print(f"Solving: players={n}, teams={num_teams}, groups per round={G}, time_limit={args.time_limit}s ...")
    res = solver.Solve(model)

    status = solver.StatusName(res)
    print("Solver status:", status)

    if res == cp_model.OPTIMAL or res == cp_model.FEASIBLE:
        print("Solution found; writing output to", args.output)
        rows = []
        for r in range(R):
            for g in range(G):
                for k in range(K):
                    assigned = None
                    for p in player_ids:
                        if solver.Value(y[(p,g,k,r)]) == 1:
                            assigned = p
                            break
                    if assigned is None:
                        print(f"Warning: empty slot r{r} g{g} k{k}")
                        continue
                    rows.append({
                        'Round': r+1,
                        'Group': g+1,
                        'Position': k+1,
                        'Player': players[assigned]['name'],
                        'Team': players[assigned]['team'],
                        'Seed': players[assigned]['seed']
                    })
        rows.sort(key=lambda x: (x['Round'], x['Group'], x['Position']))
        with open(args.output, 'w', newline='', encoding='utf-8') as f:
            writer = csv.DictWriter(f, fieldnames=['Round','Group','Position','Player','Team','Seed'])
            writer.writeheader()
            for r in rows:
                writer.writerow(r)
        # Optionally print summary per player: number of distinct teams met and repeated partners
        print("\nSummary (per player):")
        for p in player_ids:
            met_teams = []
            for team_other in unique_teams:
                if team_other == players[p]['team']:
                    continue
                if solver.Value(team_met[(p,team_other)]):
                    met_teams.append(team_other)
            repeated_partners = []
            for q in range(0, p):
                if solver.Value(rrep[(q,p)]):
                    repeated_partners.append(q)
            for q in range(p+1, n):
                if solver.Value(rrep[(p,q)]):
                    repeated_partners.append(q)
            print(f"- {players[p]['name']} (team {players[p]['team']}): met {len(met_teams)} other teams; repeated partners: {len(repeated_partners)} -> { [players[q]['name'] for q in repeated_partners] }")
        print("Finished.")
    else:
        print("No feasible solution found within time limit or problem infeasible. Solver status:", status)

if __name__ == '__main__':
    main()