WE-repartir-taches/repartir_taches/intermed_file.py

168 lines
5.7 KiB
Python

""" Fichier intermédiaire aisément éditable à la main """
from pathlib import Path
import typing as t
from dataclasses import dataclass
import dataclasses
import ruamel
from . import config
Assignment: t.TypeAlias = dict[str, list[int]]
class BadIntermedFile(Exception):
"""Raised when an intermediary file doesn't contain what is expected"""
@dataclass
class _IntermedFileData:
assignation: Assignment
nb_groupes: int
groupes: list[list[str]]
class IntermedFile:
"""Fichier intermédiaire aisément éditable à la main"""
choristes: list[str]
tasks: config.Category
data: _IntermedFileData
def __init__(
self,
choristes: list[str],
tasks: config.Category,
assignment: dict[str, list[int]],
nb_groups: int,
groups: list[list[str]],
):
self.choristes = choristes
self.tasks = tasks
self.data = _IntermedFileData(
assignation=assignment,
nb_groupes=nb_groups,
groupes=groups,
)
def sanity_check(self):
"""Check that the data is consistent -- does not check that the assignment is
fair"""
def check_taches(task: config.Category | config.Task):
if isinstance(task, config.Task):
if task.qualified_name not in self.data.assignation:
raise BadIntermedFile(
f"Aucun groupe assigné pour {task.qualified_name}"
)
nb_assigned = len(self.data.assignation[task.qualified_name])
if nb_assigned != task.nb_groups:
raise BadIntermedFile(
f"{nb_assigned} groupes assignés pour {task.qualified_name}, "
f"il en faut {task.nb_groups}"
)
else:
for child in task.tasks:
check_taches(child)
if len(self.data.groupes) != self.data.nb_groupes:
raise BadIntermedFile("Nombre de groupes incohérent")
choristes_seen: dict[str, int] = {}
for grp_id, group in enumerate(self.data.groupes):
for choriste in group:
if choriste in choristes_seen:
raise BadIntermedFile(
f"{choriste} est dans plusieurs groupes : "
f"{choristes_seen[choriste]} et {grp_id}."
)
if choriste not in self.choristes:
raise BadIntermedFile(
f"{choriste} est dans le groupe {grp_id}, mais ne fait pas "
"partie de la liste des choristes"
)
choristes_seen[choriste] = grp_id
missing_choristes = set(self.choristes) - set(choristes_seen.keys())
if missing_choristes:
raise BadIntermedFile(
"Ces choristes ne sont dans aucun groupe : "
+ ", ".join(missing_choristes)
)
check_taches(self.tasks)
def write(self, to_path: Path):
"""Write the intermediary file to this path"""
yaml = ruamel.yaml.YAML()
yaml.dump(dataclasses.asdict(self.data), to_path)
def to_assignment(self, conf: config.Config) -> list[list[str]]:
"""Use this intermediary file to make an assignment. Assigns groups to tasks
in the config, and returns the groups"""
def assign_tasks(task: config.Category | config.Task):
if isinstance(task, config.Task):
task.assigned = self.data.assignation[task.qualified_name]
task.assigned.sort()
else:
for child in task.tasks:
assign_tasks(child)
assign_tasks(conf.taches)
return self.data.groupes
@classmethod
def from_assignment(
cls, conf: config.Config, groups: list[list[str]]
) -> "IntermedFile":
def make_assignment(task: config.Category | config.Task, out: Assignment):
if isinstance(task, config.Task):
assert task.qualified_name not in out
assert task.assigned is not None
# Internal groups are 0-indexed
out[task.qualified_name] = list(map(lambda x: x + 1, task.assigned))
else:
for child in task.tasks:
make_assignment(child, out)
assignment: Assignment = {}
make_assignment(conf.taches, assignment)
return cls(
choristes=conf.choristes,
tasks=conf.taches,
assignment=assignment,
nb_groups=len(groups),
groups=groups,
)
@classmethod
def from_file(
cls, choristes: list[str], tasks: config.Category, file: Path
) -> "IntermedFile":
yaml = ruamel.yaml.YAML()
with file.open("r") as handle:
raw_data = yaml.load(handle)
for expected_field in dataclasses.fields(_IntermedFileData):
if expected_field.name not in raw_data:
raise BadIntermedFile(
f"Le champ '{expected_field.name}' est absent du fichier !"
)
try:
data = _IntermedFileData(**raw_data)
except TypeError as exn:
raise BadIntermedFile(
"Impossible d'interpréter ce fichier comme un fichier intermédiaire."
) from exn
# Internal groups are 0-indexed
for grp in data.assignation.values():
for pos in range(len(grp)):
grp[pos] -= 1
return cls(
choristes=choristes,
tasks=tasks,
assignment=data.assignation,
nb_groups=data.nb_groupes,
groups=data.groupes,
)