""" 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, )