survive empty prefs entry

glpm.py

@@ -1,22 +0,0 @@
-def col_all_zeros(ls, i, al=0):
-    return all(x[i] == al for x in ls)
-def row_all_zeros(ls, i, al=0):
-    return all(x == al for x in ls[i])
-def matrix(name, ls, default=0):
-    nonzero_cols = [i+1 for i in range(len(ls[0])) if not col_all_zeros(ls, i, default)]
-    nonzero_rows = [i+1 for i in range(len(ls)) if not row_all_zeros(ls, i, default)]
-    res = ""
-    for r in nonzero_rows:
-        res += "\n{:2d}".format(r)
-        for c in nonzero_cols:
-            res += " {:2d}".format(ls[r-1][c-1])
-    return param(name, res, " : " + " ".join("{:2d}".format(x) for x in nonzero_cols))
-
-def dict(name, thing, default=None):
-    fmt_key = lambda k: " ".join((str(x+1) for x in k)) if type(k) == tuple else k+1
-    return param(name, ", ".join(["{} {}".format(fmt_key(k), v) for k,v in thing.items() if v != default]))
-def param(name, val, middle=""):
-    val = str(val)
-    if "\n" in val:
-        val = val.replace("\n", "\n" + " " * (len(name) + 6))
-    return "param {}{} := {};".format(name, middle, val)

magisch_corvee_script.py

@@ -1,61 +1,83 @@
 #! /usr/bin/env nix-shell
-#!nix-shell -i python3 -p python3 python3Packages.pyyaml glpk -I nixpkgs=flake:nixpkgs
+#!nix-shell -i python3 -p python3 python3Packages.pyyaml -I nixpkgs=flake:nixpkgs
+# pip install pyyaml pyscipopt pyright
 import sys
 import yaml
 import re
-import subprocess
+import argparse
-from collections import OrderedDict
+from collections import OrderedDict, defaultdict
-import glpm
+from pyscipopt import Model, quicksum
+from typing import Any, Tuple, TypeVar
+from dataclasses import dataclass, field
+from tabulate import tabulate
 conf = yaml.safe_load(open('config.yaml', 'r'))
-config = conf['config']
+
+DEFAULT_CONFIG = {
+    "max_load_person": 6,
+    "day_names": [f"dag {str(i)}" for i in range(conf['config']['days'])],
+}
+config = DEFAULT_CONFIG | conf['config']
 config['ignore'].append('')
-tasks = OrderedDict(conf['tasks'])
+assert(len(config['day_names']) == config['days'])
-index = lambda x: {v:k for k,v in enumerate(x)}
+
+QUADRATIC = False
+
+@dataclass
+class TaskConfig:
+    personen: list[str]
+    workload: int
+    req: list[int]
+    name: str
+    hardcode: list[str] | None = None
+    lookup: list[str] = field(default_factory=list)
+
+tasks: dict[str, TaskConfig] = OrderedDict({k: TaskConfig(**({"name": k} | t)) for k, t in conf['tasks'].items()})
+X = TypeVar("X")
+def index(x: dict[X, Any]) -> dict[X, int]:
+    return {v: k for k, v in enumerate(x)}
 daily_workloads = \
-     [sum(task['workload'] * task['req'][d] for task in tasks.values()) for d in range(config['days'])]
+     [sum(task.workload * task.req[d] for task in tasks.values()) for d in range(config['days'])]
-ALL_DAYS = set(range(config['days']))
+ALL_DAYS: set[int] = set(range(config['days']))
 class Person(object):
-    def __init__(self, conf={"dagen":ALL_DAYS}):
+    def __init__(self, name: str, conf={"dagen":ALL_DAYS}):
-        self.can = set()
+        self.name = name
-        self.loves = set()
+        self.can: set[str] = set()
-        self.hates = set()
+        self.loves: set[str] = set()
-        self.does = set() # hardcoded
+        self.hates: set[str] = set()
+        self.does: set[Tuple[int, str]] = set() # hardcoded
         self.has_prefs = False
         self.conf = conf
         self.conf['dagen'] = set(conf['dagen'])
     def vrolijkheid(self):
         res = config['days'] - len(self.does)
-        for (d,t) in self.does:
+        for (_,t) in self.does:
             if t in self.loves:
                 res += config['weights']['likes']
             if t in self.hates:
                 res -= config['weights']['hates']
         return res
     def workload(self, tasks):
-        return sum(tasks[t]['workload'] for (d,t) in self.does)
+        return sum(tasks[t].workload for (_,t) in self.does)
     def cost(self, num_people):
         return round(sum((daily_workloads[d] for d in self.conf['dagen'])) / num_people)
         # probabilistic round: int(math.floor(x + random.random()))
-def read_people(conf_ppl):
+def read_people(conf_ppl) -> dict[str, Person]:
-    def isdict(x):
-        return type(x) == dict
     people = OrderedDict()
     for x in conf_ppl:
         val = {"dagen": ALL_DAYS}
-        if type(x) == dict:
+        if isinstance(x, dict):
             x,val = x.popitem()
-        people[x.lower()] = Person(val)
+        people[x.lower()] = Person(x, val)
     return people
 # deal with loves/hates
-def make_task_lut(tasks):
+def make_task_lut(tasks: dict[str, TaskConfig]):
-    task_lut = {}
+    task_lut = defaultdict(set)
     for t, taskconf in tasks.items():
-        if 'lookup' in taskconf:
+        for lookup in taskconf.lookup:
-            for lookup in taskconf['lookup']:
+            task_lut[lookup] |= {t.lower()}
-                task_lut[lookup] = t
     task_re = re.compile(config['task_re'])
     def lookup_tasks(tasks):
-        return (task_lut[x] for x in task_re.split(tasks) if not x in config['ignore'])
+        return set.union(set(), *(task_lut[x.strip()] for x in task_re.split(tasks) if x not in config['ignore']))
     return lookup_tasks
 def read_prefs(pref_file, tasks, people):
     lookup_tasks = make_task_lut(tasks)
@@ -69,103 +91,137 @@ def read_prefs(pref_file, tasks, people):
         if not p.has_prefs:
             print("warning: no preferences for", name, file=sys.stderr)
 # deal with capability and hardcode
-def set_capabilities(tasks, people):
+def set_capabilities(tasks: dict[str, TaskConfig], people: dict[str, Person]):
-    for ti,(task,conf) in enumerate(tasks.items()):
+    for (task,conf) in tasks.items():
-        if conf['personen'] == 'iedereen':
+        if conf.personen == 'iedereen':
             for p in people.values():
                 p.can.add(task)
-        elif conf['personen'] == 'liefhebbers':
+        elif conf.personen == 'liefhebbers':
             for p in people.values():
                 if task in p.loves:
                     p.can.add(task)
         else:
-            for p in conf['personen']:
+            for p in conf.personen:
                 people[p.lower()].can.add(task)
-        if 'hardcode' in conf:
+        if conf.hardcode is not None:
-            for day, pers in enumerate(conf['hardcode']):
+            for day, pers in enumerate(conf.hardcode):
-                people[pers.lower()].does.add((day, task))
+                if pers:
-# format as matrices
+                    people[pers.lower()].does.add((day, task))
-def matrices(people, tasks):
+
-    mat = lambda a,b: [[0 for j in b] for i in a]
+def write_tasks(people: dict[str, Person], tasks: dict[str, TaskConfig], file=sys.stdout):
-    loves = mat(people, tasks)
+    headers = ["wie"] + config['day_names']
-    hates = mat(people, tasks)
+    if not args.simple:
-    capab = mat(people, tasks)
+        headers += ["workload", "vrolijkheid"]
-    tsk_idx = index(tasks)
+    tabl = []
-    hardcode = {}
+    for p in people.values():
-    max_loads = {}
+        days = [[] for _ in range(config['days'])]
-    costs = {}
+        for (d,t) in p.does:
-    for i,(person, p) in enumerate(people.items()):
+            days[d].append((t, t in p.loves, t in p.hates))
-        for t in p.loves: loves[i][tsk_idx[t]] = 1
+        if not args.simple:
-        for t in p.hates: hates[i][tsk_idx[t]] = 1
+            def q(w):
-        for t in p.can: capab[i][tsk_idx[t]] = 1
+                return ",".join([tasks[t].name + (" :)" if love else "") + (" :(" if hate else "") for (t,love,hate) in w])
-        for (d,t) in p.does: hardcode[(i,tsk_idx[t],d)] = 1
+        else:
+            def q(w):
+                return ",".join([tasks[t].name for (t,_,_) in w])
+        row = [p.name, *map(q, days)]
+        if not args.simple:
+            row += [p.workload(tasks), p.vrolijkheid()]
+        tabl.append(row)
+    print(tabulate(tabl, headers=headers, tablefmt=args.output_format), file=file)
+
+def scipsol(people: dict[str, Person], tasks: dict[str, TaskConfig]):
+    max_loads: dict[Tuple[str, int], int] = {}
+    for i,p in people.items():
         if 'max_load' in p.conf: # max_load override for Pol
-            for d,l in enumerate(p.conf['max_load']):
+            for d,load in enumerate(p.conf['max_load']):
-                max_loads[(i,d)] = l
+                max_loads[(i,d)] = load
         # filter days that the person does not exist
         for d in ALL_DAYS - p.conf['dagen']:
             max_loads[(i,d)] = 0
-        costs[i] = p.cost(len(people))
-    req = mat(range(config['days']), tasks)
-    for di in range(config['days']):
-        for ti,t in enumerate(tasks.values()):
-            req[di][ti] = t['req'][di]
-    workload = {tsk_idx[t]: tasks[t]['workload'] for t in tasks}
-    return [loves, hates, capab, hardcode, max_loads, req, workload, costs]
-def read_assignment(file, people, tasks):
-    def between_the_lines(f, a=">>>>\n", b="<<<<\n"):
-        for l in f:
-            if l == a: break
-        for l in f:
-            if l == b: break
-            yield map(int, l.strip().split())
 
-    for p in people.values():
+    m = Model()
-        p.does = set()
+    does = {}
-    person_vl = list(people.values())
+    happiness = []
-    task_nm = list(tasks.keys())
+    stdevs = []
-    for [p,d,j,W,l] in between_the_lines(file):
+    errors = []
-        person_vl[p-1].does.add((d-1, task_nm[j-1]))
+    for pname, person in people.items():
-def write_data(people, tasks, file=sys.stdout):
+        workloads = []
-    [loves, hates, capab, hardcode, max_loads, req, workload, costs] = matrices(people, tasks)
+        p_error = m.addVar(vtype="I", name=f"{pname}_error", lb=0, ub=None)
-    print(glpm.matrix("L", loves), file=file)
+        for d in ALL_DAYS:
-    print(glpm.matrix("H", hates), file=file)
+            pdt = []
-    print(glpm.matrix("C", capab, 1), file=file)
+            for task in tasks:
-    print(glpm.matrix("R", req, None), file=file)
+                var = m.addVar(vtype="B", name=f"{pname}_does_{task}@{d}")
-    print(glpm.dict("Q", hardcode), file=file)
+                pdt.append(var)
-    print(glpm.dict("Wl", workload), file=file)
+                does[(pname, d, task)] = var
-    print(glpm.dict("max_load", max_loads), file=file)
+                # a person only does what (s)he is capable of
-    print(glpm.dict("Costs", costs), file=file)
+                if task not in person.can:
-    print(glpm.param("D_count", config['days']), file=file)
+                    m.addCons(var == 0)
-    print(glpm.param("P_count", len(people)), file=file)
+                workloads.append(var * tasks[task].workload)
-    print(glpm.param("J_count", len(tasks)), file=file)
+            for task in person.loves:
-    print(glpm.param("ML", 6), file=file) # CHANGE THIS
+                happiness.append(does[(pname, d, task)] * config['weights']['likes'])
-    print(glpm.param("WL", config['weights']['likes']), file=file)
+            for task in person.hates:
-    print(glpm.param("WH", config['weights']['hates']), file=file)
+                happiness.append(does[(pname, d, task)] * (config['weights']['hates'] * -1))
-def write_tasks(people, tasks, file=sys.stdout):
+
-    for name, p in people.items():
+            # max_load_person: a person only has one task per day at most
-        days = [[] for i in range(config['days'])]
+            m.addCons(quicksum(pdt) <= max_loads.get((pname, d), 1))
-        for (d,t) in p.does:
+
-            days[d].append((t, t in p.loves, t in p.hates))
+        m.addCons(p_error >= person.cost(len(people)) - quicksum(workloads))
-        q = lambda w: ",".join([t + (" <3" if l else "") + (" :(" if h else "") for (t,l,h) in w])
+        m.addCons(p_error >= quicksum(workloads) - person.cost(len(people)))
-        days_fmt = " {} ||" * len(days)
+        errors.append(p_error)
-        days_filled = days_fmt.format(*map(q, days))
+
-        print("| {} ||{} {} || {}".format(name, days_filled, p.vrolijkheid(), p.workload(tasks)), file=file)
+        stdevs.append((person.cost(len(people)) - quicksum(workloads)) ** 2)
-        print("|-")
+
+        # min_load_person: person has at least 1 task
+        m.addCons(quicksum([does[(pname, d, task)] for d in ALL_DAYS for task in tasks]) >= 1)
+
+        # duplicate_jobs: a person does not perform the same job on all days
+        for task in tasks:
+            m.addCons(quicksum(does[(pname, d, task)] for d in ALL_DAYS) <= len(ALL_DAYS) - 1)
+
+        # max_load_person_total
+        m.addCons(quicksum([does[(pname, d, task)] * tasks[task].workload for d in ALL_DAYS for task in tasks]) <= config['max_load_person'])
+
+        # hardcode constraint
+        for d, task in person.does:
+            m.addCons(does[(pname, d, task)] == 1)
+
+
+    # all_allocated: each task in allocated
+    for j in tasks:
+        for d in ALL_DAYS:
+            m.addCons(quicksum(does[(p, d, j)] for p in people) == tasks[j].req[d])
+
+    objective = m.addVar(name="objvar", vtype="C", lb=None, ub=None)
+    m.setObjective(objective, "maximize")
+    if QUADRATIC:
+        m.addCons(objective <= quicksum(happiness) - quicksum(stdevs) / 2)
+    else:
+        if args.max_total_error is not None:
+            m.addCons(quicksum(errors) <= args.max_total_error)
+        m.addCons(objective <= quicksum(happiness) - quicksum(errors))
+        m.setObjective(objective, "maximize")
+    m.solveConcurrent()
+    for pname, person in people.items():
+        for d in ALL_DAYS:
+            for task in tasks:
+                if m.getVal(does[(pname, d, task)]):
+                    person.does.add((d, task))
+    write_tasks(people, tasks)
+
+    print("Totale vrolijkheid:", sum(p.vrolijkheid() for p in people.values()))
+    print("workload deviation:", m.getVal(quicksum(stdevs)))
+
+parser = argparse.ArgumentParser()
+parser.add_argument("-q", "--quadratic", action="store_true")
+parser.add_argument("--simple", action="store_true", help="hide workload and happiness")
+parser.add_argument("--max_total_error", type=int, default=None)
+parser.add_argument("--output-format", default="mediawiki", help="`tabulate` output format")
+
+args = parser.parse_args()
+QUADRATIC = args.quadratic
+
 people = read_people(conf['people'])
 with open('prefs_table', 'r') as pref_file:
     read_prefs(pref_file, tasks, people)
 set_capabilities(tasks, people)
-if len(sys.argv)>1 and sys.argv[1] == 'in':
+scipsol(people, tasks)
-    write_data(people, tasks)
-elif len(sys.argv)>1 and sys.argv[1] == 'out':
-    with open('output', 'r') as out_file:
-        read_assignment(out_file, people, tasks)
-    write_tasks(people, tasks)
-else:
-    with open('data', 'w') as out:
-        write_data(people, tasks, file=out)
-    subprocess.call(["glpsol", "--model", "model.glpm", "-d", "data", "--tmlim", "15", "--log", "output"], stdout=sys.stderr, stderr=sys.stdout)
-    with open('output', 'r') as file:
-        read_assignment(file, people, tasks)
-        write_tasks(people, tasks)

model.glpm

@@ -1,84 +0,0 @@
-/* Number of people */
-param P_count, integer, > 0;
-
-/* Number of jobs */
-param J_count, integer, > 0;
-
-/* Number of days */
-param D_count, integer, > 0;
-
-param WL, integer, > 0;
-param WH, integer, > 0;
-param ML, integer, > 0;
-
-set P := 1..P_count;
-set J := 1..J_count;
-set D := 1..D_count;
-
-/* aanwezigheid x workload for that day */
-param Costs{p in P}, integer, >= 0;
-
-/* Person p likes to solve jobs j */
-param L{p in P, j in J} default 0, binary;
-
-/* Person p hates to solve jobs j */
-param H{p in P, j in J} default 0, binary;
-
-/* Person p is capable to perform job j */
-param C{p in P, j in J} default 1, binary;
-
-/* How many jobs need to be done on what day */
-param R{d in D, j in J}, integer, >= 0;
-
-/* hardcoded */
-param Q{p in P, j in J, d in D}, default 0, binary;
-
-/* workload */
-param Wl{j in J}, integer, >= 0;
-
-param max_load{p in P, d in D}, default 1, integer;
-
-/* Person p is allocated to do job j on day d */
-var A{p in P, j in J, d in D}, binary;
-
-var error{p in P}, integer, >= 0;
-
-s.t. hardcode{p in P, j in J, d in D}: A[p,j,d] >= Q[p,j,d];
-
-/* A person only has one task per day, at most */
-s.t. max_load_person{p in P, d in D}: sum{j in J} A[p,j,d] <= max_load[p,d];
-
-/* A person has at least 1 task */
-s.t. min_load_person{p in P}: sum{j in J, d in D} A[p,j,d] >= 1;
-
-/* A person does not perform the same job on all days */
-s.t. duplicate_jobs{p in P, j in J}: sum{d in D} A[p,j,d] <= D_count-1;
-
-s.t. max_load_person_total{p in P}: (sum{d in D, j in J} A[p,j,d] * Wl[j]) <= ML;
-
-/* Each task is allocated */
-s.t. all_allocated{j in J, d in D}: sum{p in P} A[p,j,d] == R[d, j];
-
-/* A person only performs what (s)he is capable of */
-s.t. capability_person{p in P, j in J, d in D}: A[p,j,d] <= C[p,j];
-
-s.t. error_lt{p in P}: error[p] >= ((sum{j in J, d in D} A[p,j,d] * Wl[j]) - Costs[p]);
-s.t. error_gt{p in P}: error[p] >= Costs[p] - (sum{j in J, d in D} A[p,j,d] * Wl[j]);
-
-/* Maximize enjoyment */
-# minimize error_diff: sum{p in P} error[p];
-maximize enjoyment: (sum{p in P, d in D, j in J} A[p,j,d] * (L[p, j] * WL - H[p, j] * WH)) - sum{p in P} error[p];
-solve;
-
-printf "Sum %d\n", (sum{p in P, d in D, j in J} A[p,j,d] * (L[p, j] * WL - H[p, j] * WH));
-printf "p d j W l\n";
-printf ">>>>\n";
-printf{p in P, d in D, j in J : A[p,j,d] > 0} "%d %d %d %d %d\n", p, d, j, A[p,j,d] * (L[p, j] * WL - H[p, j] * WH), Wl[j];
-printf "<<<<\n";
-printf "workloads\n";
-printf "p l\n";
-printf{p in P} "%d %d\n", p, abs((sum{j in J, d in D : A[p,j,d] > 0} Wl[j]) - Costs[p]);
-printf "workload_dev: %d\n", sum{p in P} abs((sum{j in J, d in D : A[p,j,d] > 0} Wl[j]) - Costs[p])^2;
-
-
-end;

requirements.txt

@@ -0,0 +1,3 @@
+tabulate~=0.9.0
+PySCIPOpt~=4.4.0
+PyYAML~=6.0.1