diff --git a/Script.org b/Script.org index 2121d2a..91965a8 100644 --- a/Script.org +++ b/Script.org @@ -346,39 +346,103 @@ le plan sans rien supprimer, et =--yes= pour omettre la confirmation interactive. * Autres -** Recorrection d'une seule copie (peu testé) +** Recorrection d'une copie ou de quelques questions -!! Attention, refaire ne marchera pas si tu fais une annotation non -groupée into refaire !! +Dans le GUI, ouvrir la section =Refaire des copies (facultatif)=, +repliée par défaut. Ajouter les copies et leurs questions depuis les +listes déroulantes, ou choisir =Toute la copie=, puis enregistrer la +sélection. Le dossier du passage principal est présélectionné d'après +le dernier mode utilisé et les dossiers présents ; vérifier ce choix. +Pour reprendre une même question dans toute la classe, choisir +=Toutes les copies=, sélectionner la question, puis =+ Ajouter=. +Seules les copies ayant un PDF de réponse pour cette question (normal +ou =_new=) sont ajoutées ; le nombre de copies sans réponse est affiché. +Une copie déjà sélectionnée entièrement reste sélectionnée entièrement. - 1. Redécoupage - + =python -m copienator review-labels InterroTest/Copie01.pdf= - + =python -m copienator split-answers InterroTest/Copie20.pdf= - 2. Créer =refaire.json=, avec un contenu comme +Les boutons =Corrigés (Sol)= et =Consignes de notation (Persp)= ouvrent +les dossiers des textes utilisés par la correction. Modifier et enregistrer +les fichiers des questions concernées avant de relancer =Refaire la correction=. +Ces boutons sont aussi disponibles dans le parcours principal. + +Le choix =PDF à vérifier= propose =Automatique=, =Par question (groupé)= +ou =Par copie=. En automatique, une question présente dans plusieurs +copies déclenche le regroupement ; sinon les PDF sont produits par copie. +Les groupes sont limités en hauteur : une question pour toute la classe +peut donc produire quelques PDF plutôt qu'un fichier par élève. +Le GUI écrit =refaire.json= et guide ensuite le parcours ci-dessous. +La vérification du découpage, le redécoupage et la recorrection peuvent +être ignorés. Pour plusieurs copies, les vérifications et découpages +s'exécutent successivement ; un échec ou une interruption arrête la suite. +Ce parcours a sa propre progression : les boutons de navigation du +passage principal ne l'ouvrent pas automatiquement. Après la fusion, +les étapes de restitution déjà terminées sont marquées à revalider. + +Ce flux fonctionne après =annotate-grouped= (=BGnot=), +=annotate-checks= (=Bnot=) ou =annotate-simple= (=Anot=). +Terminer d'abord la lecture des annotations du passage principal +(=read-grouped= ou =read-annotations= pour les modes à cases). +Conserver les dossiers d'annotation et leurs fichiers de référence. + + 1. Si nécessaire, reprendre le découpage : + + =python -m copienator review-labels Interro/Copies/Copie01.pdf= + + =python -m copienator split-answers Interro/Copies/Copie01.pdf= + Vérifier les réponses découpées avant de relancer la correction, + notamment les fichiers =_new= et =_old= issus de résolutions manuelles. + 2. Créer =Interro/refaire.json= : : [["Copie02", []], : ["Copie01", ["Ex 1 : 1)"]]] - 3. Appeler =correction= avec --refaire. Il doit créer des groupes - individuels, faire des requêtes, et remplacer les corrections - précédentes (à sauver ailleurs). + Une liste vide sélectionne toute la copie ; sinon donner les labels + exacts des questions (pas seulement le nom de l'exercice). + 3. =python -m copienator correct Interro --refaire= + Crée des groupes individuels et remplace les corrections sélectionnées. + Les anciennes corrections sont conservées dans =overwritten_correction.json=. + Cette étape peut être omise si les corrections sont modifiées à la main. + 4. Générer les PDF de vérification, selon la présentation souhaitée : + + Par question : =python -m copienator annotate-grouped Interro --refaire --overwrite= + + Par copie : =python -m copienator annotate-checks Interro --refaire --overwrite= + Les deux commandes produisent uniquement les réponses sélectionnées + dans =BRnot=, avec des cases à cocher, quel que soit le mode du passage + principal. Le mode groupé garde les identifiants des élèves et regroupe + les réponses par label sans demander de modifier =label_groups=. + Cela ne nécessite pas d'avoir généré =Bnot= ou =BGnot= auparavant. + Attention : =--overwrite= remplace tout le précédent passage dans + =BRnot=, y compris ses annotations manuscrites. Une génération incomplète + conserve l'ancien =BRnot=. Sans =--overwrite=, un =BRnot= existant est refusé. + 5. Vider les dossiers personnels d'export/import des anciens fichiers, + puis =python -m copienator export Interro --refaire=. + Annoter les PDF sur la tablette, puis placer les PDF retournés dans + =IMPORT_DIR= en conservant leur nom exporté (nom de groupe ou =Copie01.pdf=). + Même sans modification manuscrite, retourner le PDF pour valider ce passage. + 6. =python -m copienator import Interro --refaire= + 7. Fusionner dans le dossier du passage principal : + + Groupé : =python -m copienator read-grouped Interro --refaire= + + Cases : =python -m copienator read-grouped Interro --refaire --annotation-dir Bnot= + + Simple : =python -m copienator read-grouped Interro --refaire --annotation-dir Anot= - Ou non, si tu veux le faire à la main. - 4. ?? Si je fais refaire, avant d'avoir créer les annotating with - checks, que se passe-t-il ??? - 5. Appeler =python -m copienator annotate-checks --refaire --overwrite= - 6. =python -m copienator export --refaire Interro24= - 6. =python -m copienator import --refaire Interro24= - 7. =python -m copienator read-grouped --refaire Interro24= + Le lecteur reconnaît les retours par question comme les retours par + copie grâce aux métadonnées et réattribue les cases et notes à chaque + élève. Un groupe manquant laisse intactes les copies qui en dépendent. + Il reconstruit la copie complète, conserve les réponses non + sélectionnées et leurs scores, et remplace les réponses sélectionnées + par celles de =BRnot=. Dans la compilation filtrée, les images déjà + enregistrées des questions non sélectionnées sont conservées par + prudence, même si leur score est parfait, pour ne pas perdre de notes. + Les anciennes cases et notes manuscrites des + questions refaites sont remplacées. Les autres copies restent intactes. + En mode simple, une image =Concat_annotated.jpg= ou =.jpeg= importée + doit conserver les dimensions de l'image exportée ; les parties non + sélectionnées sont conservées. Le fichier =refaire_simple_layout.json= + mémorise le découpage de cette image pour les passages suivants. - Avec =--refaire=, =refaire.json= et le dossier =BRnot= sont des - prérequis obligatoires ; leur absence produit le code de sortie 3. + Les sorties finales (=Concat.jpg=, images par question, =score.json= + et compilation filtrée) sont mises à jour dans =BGnot=, =Bnot= ou + =Anot= ; les PDF et références du passage principal restent ceux de + ce passage. Pour une nouvelle retouche, reprendre ce flux =--refaire=, + sans relire ensuite les anciennes annotations avec le lecteur normal. + Relancer ensuite les étapes habituelles de calcul des notes et de diffusion. -** Exemple de replotting, refaire d'une copie - - 1. replot it. - 2. `python -m copienator split-answers DS09VA/Copies/Copie25.pdf` - this will get rid of old/new. - !! Attention, et si ça dégage un new : bad bad bad. - 3. Make `refaire.json`, avec la copie, et les labels à refaire. - 4. `python -m copienator correct DS09VA --refaire` - 5. `python -m copienator annotate-checks DS09VA --refaire` - 6. `python -m copienator import Interro24 --refaire` + =refaire.json=, =BRnot= et le dossier du passage principal sont + obligatoires (code de sortie 3 s'ils manquent). Une copie dont les + fichiers de retour sont incomplets est laissée intacte (code 4). + Ne pas ajouter =--update-score= sauf pour imposer volontairement les + anciens scores, y compris ceux des questions refaites. diff --git a/copienator/annotation_data.py b/copienator/annotation_data.py index 404e366..b7ce3ff 100644 --- a/copienator/annotation_data.py +++ b/copienator/annotation_data.py @@ -36,7 +36,12 @@ def _coordinate_index( if not workspace.groups_dir.is_dir(): return index, [f"Group directory not found: {workspace.groups_dir}"] - for metadata_path in sorted(workspace.groups_dir.glob("*/Group_*.json")): + # A redo appends a new numbered group; its coordinates supersede the old group. + for metadata_path in sorted( + workspace.groups_dir.glob("*/Group_*.json"), + key=lambda path: int(path.stem.removeprefix("Group_")), + reverse=True, + ): image_path = metadata_path.with_suffix(".jpg") try: entries = read_json(metadata_path) @@ -160,12 +165,16 @@ def load_annotation_data( warnings.append(f"Ignoring malformed correction batch for {label!r}") continue for item in raw_batch: - if not isinstance(item, dict) or not isinstance(item.get("result"), dict): + if not isinstance(item, dict) or not isinstance( + item.get("result"), dict + ): warnings.append(f"Ignoring malformed correction item for {label!r}") continue student_id = str(item.get("id", "")) if not student_id: - warnings.append(f"Ignoring correction item without an id for {label!r}") + warnings.append( + f"Ignoring correction item without an id for {label!r}" + ) continue result = item["result"] suffix = str(result.get("suffix", "")) diff --git a/copienator/commands/annotating_by_label.py b/copienator/commands/annotating_by_label.py index 764afaa..2cdf1d6 100644 --- a/copienator/commands/annotating_by_label.py +++ b/copienator/commands/annotating_by_label.py @@ -10,9 +10,6 @@ from typing import Any from PIL import Image, ImageDraw from reportlab.pdfgen import canvas -from copienator.commands import annotating -from copienator.commands import annotating_with_checks -from copienator import utils from copienator import ( CliError, EvaluationWorkspace, @@ -22,9 +19,11 @@ from copienator import ( evaluation_parser, execute, read_json, + utils, workspace_from_args, ) from copienator.annotation_data import load_annotation_data +from copienator.commands import annotating, annotating_with_checks from copienator.filesystem import staged_directory from copienator.utils import natural_key @@ -184,7 +183,9 @@ def _serialize_label_groups(groups: list[list[str]]) -> str: return "".join(",".join(group) + "\n" for group in groups) -def _load_label_groups(workspace: EvaluationWorkspace, labels: list[str]) -> list[list[str]]: +def _load_label_groups( + workspace: EvaluationWorkspace, labels: list[str] +) -> list[list[str]]: label_groups = workspace.label_groups_file if not label_groups.exists(): gemini_groups = _gemini_label_groups(workspace, labels) @@ -328,20 +329,38 @@ def _generate_groups( return generated, problems -def run(workspace: EvaluationWorkspace, *, overwrite: bool = False) -> ExitCode: +def run( + workspace: EvaluationWorkspace, *, overwrite: bool = False, refaire: bool = False +) -> ExitCode: workspace.require_files("labels", "correction.json") workspace.require_directories("Copies", "Par label") labels = utils.read_all_labels(workspace.root) - groups = _load_label_groups(workspace, labels) - loaded = load_annotation_data(workspace) + refaire_list = annotating_with_checks._load_refaire(workspace) if refaire else None + loaded = load_annotation_data(workspace, refaire_list=refaire_list) + groups = ( + [ + [label] + for label in sorted( + {label for answers in loaded.data.values() for label in answers}, + key=natural_key, + ) + ] + if refaire + else _load_label_groups(workspace, labels) + ) for warning in loaded.warnings: print(f"Warning: {warning}") if not loaded.data: print("Warning: no annotation data was found.") return ExitCode.PARTIAL - output_root = workspace.annotation_dir("grouped") - if overwrite: + output_root = workspace.annotation_dir("refaire" if refaire else "grouped") + if refaire and output_root.exists() and not overwrite: + raise CliError( + "BRnot already exists; use --overwrite to replace the previous redo." + ) + if overwrite or refaire: + class IncompleteGroupedOutput(Exception): pass @@ -356,7 +375,9 @@ def run(workspace: EvaluationWorkspace, *, overwrite: bool = False) -> ExitCode: if generated == 0 or problems or loaded.warnings: raise IncompleteGroupedOutput except IncompleteGroupedOutput: - print("Warning: grouped overwrite was incomplete; previous BGnot was preserved.") + print( + f"Warning: grouped overwrite was incomplete; previous {output_root.name} was preserved." + ) return ExitCode.PARTIAL else: output_root.mkdir(parents=True, exist_ok=True) @@ -375,7 +396,14 @@ def run(workspace: EvaluationWorkspace, *, overwrite: bool = False) -> ExitCode: def build_parser() -> argparse.ArgumentParser: parser = evaluation_parser("Generate annotated PDFs grouped by labels.") - parser.add_argument("--overwrite", action="store_true", help="Replace BGnot safely") + parser.add_argument( + "--overwrite", action="store_true", help="Replace annotation outputs safely" + ) + parser.add_argument( + "--refaire", + action="store_true", + help="Group only the answers in refaire.json, writing to BRnot", + ) return parser @@ -384,7 +412,9 @@ def main(argv: Sequence[str] | None = None) -> int: return execute( parser, argv, - lambda args: run(workspace_from_args(args), overwrite=args.overwrite), + lambda args: run( + workspace_from_args(args), overwrite=args.overwrite, refaire=args.refaire + ), ) diff --git a/copienator/commands/annotating_with_checks.py b/copienator/commands/annotating_with_checks.py index 43b7436..b13a37c 100644 --- a/copienator/commands/annotating_with_checks.py +++ b/copienator/commands/annotating_with_checks.py @@ -14,8 +14,6 @@ matplotlib.use("Agg") from PIL import Image, ImageFont from reportlab.pdfgen import canvas -from copienator.commands import annotating -from copienator import utils from copienator import ( CliError, EvaluationWorkspace, @@ -24,9 +22,11 @@ from copienator import ( execute, read_json, target_parser, + utils, workspace_from_target, ) from copienator.annotation_data import load_annotation_data +from copienator.commands import annotating from copienator.filesystem import staged_directory from copienator.utils import natural_key @@ -46,7 +46,9 @@ except OSError: def draw_checkbox(draw, x, y, size=BOX_SIZE, label=None, fill="white"): if label: - draw.text((x - BOX_SIZE - 5, y + 2), str(label), fill="black", font=CHECKBOX_FONT) + draw.text( + (x - BOX_SIZE - 5, y + 2), str(label), fill="black", font=CHECKBOX_FONT + ) draw.rectangle([x, y, x + size, y + size], fill=fill, outline="black", width=2) return [x, y, x + size, y + size] @@ -150,8 +152,13 @@ def _render_student( *, overwrite: bool, output_mode: str, + output_root: Path | None = None, ) -> str: - output_dir = workspace.annotation_dir(output_mode) / f"Copie{student_id}" + output_dir = ( + output_root + if output_root is not None + else workspace.annotation_dir(output_mode) + ) / f"Copie{student_id}" if _output_complete(output_dir) and not overwrite: print(f"Skipping {student_id}: output is complete.") return "skipped" @@ -178,6 +185,7 @@ def _render_student( draw_callback=checkbox_renderer.callback, ) if final_image is None: + problems = True continue label_images.append(final_image) checkbox_groups.append(checkbox_renderer.checkboxes) @@ -280,6 +288,39 @@ def run( output_mode = "refaire" if refaire else "checks" tasks = sorted(loaded.data.items(), key=lambda item: natural_key(item[0])) + if refaire: + output_root = workspace.annotation_dir("refaire") + if output_root.exists() and not overwrite: + raise CliError( + "BRnot already exists; use --overwrite to replace the previous redo." + ) + + class IncompleteRedo(Exception): + pass + + try: + with staged_directory(output_root) as staging: + with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor: + futures = [ + executor.submit( + _render_student, + workspace, + student_id, + labels, + overwrite=True, + output_mode="refaire", + output_root=staging, + ) + for student_id, labels in tasks + ] + statuses = [future.result() for future in futures] + if loaded.warnings or any(status != "success" for status in statuses): + raise IncompleteRedo + except IncompleteRedo: + print("Warning: incomplete redo generation; previous BRnot was preserved.") + return ExitCode.PARTIAL + return ExitCode.SUCCESS + statuses: list[str] = [] with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor: futures = [ @@ -302,7 +343,9 @@ def run( def build_parser() -> argparse.ArgumentParser: parser = target_parser("Generate annotated PDFs with checkboxes.") - parser.add_argument("--overwrite", action="store_true", help="Replace existing outputs") + parser.add_argument( + "--overwrite", action="store_true", help="Replace existing outputs" + ) parser.add_argument( "--refaire", action="store_true", @@ -328,4 +371,3 @@ def main(argv: Sequence[str] | None = None) -> int: if __name__ == "__main__": raise SystemExit(main()) - diff --git a/copienator/commands/reading_grouped_annotations.py b/copienator/commands/reading_grouped_annotations.py index f22c572..a66ffd2 100644 --- a/copienator/commands/reading_grouped_annotations.py +++ b/copienator/commands/reading_grouped_annotations.py @@ -9,8 +9,6 @@ from typing import Any from PIL import Image, ImageDraw -from copienator.commands import annotating -from copienator import utils from copienator import ( EvaluationWorkspace, ExitCode, @@ -18,16 +16,18 @@ from copienator import ( evaluation_parser, execute, read_json, + utils, workspace_from_args, ) from copienator.annotation_actions import apply_checkbox_actions, apply_score_overrides from copienator.annotation_data import AnnotationData, RefaireList, load_annotation_data -from copienator.filesystem import staged_files +from copienator.commands import annotating from copienator.commands.reading_annotations import ( concatenate, detect_checks_and_notes, has_significant_notes, ) +from copienator.filesystem import staged_files LabelNotes = dict[str, dict[str, Any]] ScanResult = tuple[dict[str, list[dict[str, Any]]], dict[str, LabelNotes]] @@ -54,7 +54,9 @@ def get_extra_pdfs_as_images( return images -def save_paginated_pdf(image_groups: list[list[Image.Image]], output_path: Path) -> None: +def save_paginated_pdf( + image_groups: list[list[Image.Image]], output_path: Path +) -> None: """Paginate vertically concatenated image groups and save them as a PDF.""" non_empty = [group for group in image_groups if group] if not non_empty: @@ -118,6 +120,8 @@ def _scan_annotation_directory( directory: Path, only_ids: set[str] | None = None, default_student_id: str | None = None, + *, + required: bool = False, ) -> ScanResult: bnote_path = directory / "bnote.json" if not bnote_path.is_file(): @@ -133,6 +137,8 @@ def _scan_annotation_directory( actions, notes_image = detect_checks_and_notes(directory) if notes_image is None: + if required: + raise ValueError(f"Could not read annotations in {directory}") return {}, {} actions_by_student: dict[str, list[dict[str, Any]]] = defaultdict(list) notes_by_student: dict[str, LabelNotes] = defaultdict(dict) @@ -176,10 +182,12 @@ def apply_actions_and_regenerate_grouped( all_labels: list[str], *, update_score: bool = False, + annotation_dir: str = "BGnot", + selected_labels: set[str] | None = None, ) -> tuple[ExitCode, str]: - """Apply grouped annotations and atomically merge regenerated student files.""" + """Regenerate a copy, preserving reviewed images outside the redo selection.""" logs = [f"\nProcessing compilation for: Copie{student_id}"] - output_dir = workspace.annotation_dir("grouped") / f"Copie{student_id}" + output_dir = workspace.root / annotation_dir / f"Copie{student_id}" labels_data = data.get(student_id, {}) dirty_labels = apply_checkbox_actions(labels_data, actions, logs.append) if update_score: @@ -187,15 +195,91 @@ def apply_actions_and_regenerate_grouped( labels_data, output_dir / "score.json", logs.append ) + selected_labels = selected_labels if selected_labels is not None else set() + dirty_labels |= selected_labels + simple_layout = None + simple_annotated = None + if selected_labels and annotation_dir == "Anot": + imported = next( + ( + output_dir / name + for name in ("Concat_annotated.jpg", "Concat_annotated.jpeg") + if (output_dir / name).is_file() + ), + None, + ) + if imported is not None: + layout_path = output_dir / "refaire_simple_layout.json" + if layout_path.is_file(): + simple_layout = read_json(layout_path) + else: + simple_layout = {"images": {}, "replaced": []} + y = 0 + for label in sorted(labels_data, key=utils.natural_key): + path = output_dir / f"{label}.jpg" + if ( + path.is_file() + and labels_data[label]["result"].get("error") != "empty-answer" + ): + with Image.open(path) as saved: + simple_layout["images"][label] = [y, y + saved.height] + y += saved.height + with Image.open(imported) as saved: + simple_annotated = saved.convert("RGB").copy() + expected_height = max( + (bounds[1] for bounds in simple_layout["images"].values()), default=0 + ) + if simple_annotated.height != expected_height: + raise ValueError( + "Imported simple image height does not match the original copy layout" + ) + old_scores = ( + read_json(output_dir / "score.json") + if selected_labels and (output_dir / "score.json").is_file() + else {} + ) scores = dict.fromkeys(all_labels, "") dirty_images: dict[str, Image.Image] = {} concat_images: list[Image.Image] = [] filtered_groups: list[list[Image.Image]] = [] incomplete = False - for label, content in sorted(labels_data.items(), key=lambda item: utils.natural_key(item[0])): + for label, content in sorted( + labels_data.items(), key=lambda item: utils.natural_key(item[0]) + ): result = content["result"] + if ( + selected_labels + and label not in selected_labels + and old_scores.get(label, "") != "" + ): + result["score"] = old_scores[label] scores[label] = str(result.get("score", 0)) + saved_image = output_dir / f"{label}.jpg" + if selected_labels and label not in selected_labels and saved_image.is_file(): + with Image.open(saved_image) as saved: + final_image = saved.convert("RGB").copy() + if ( + simple_annotated is not None + and label in simple_layout["images"] + and label not in simple_layout["replaced"] + ): + hmin, hmax = simple_layout["images"][label] + final_image = simple_annotated.crop( + (0, hmin, simple_annotated.width, hmax) + ) + dirty_images[label] = final_image + scores[label] = str(old_scores.get(label, scores[label])) + concat_images.append(final_image) + # Keep previously reviewed content, including handwriting. + if annotation_dir == "BGnot": + extras = get_extra_pdfs_as_images( + workspace.root, label, annotating, all_labels + ) + filtered_groups.append([*extras, final_image]) + else: + filtered_groups.append([final_image]) + continue pdf_path = Path(content["pdf_path"]) if not pdf_path.is_file(): logs.append(f" Missing answer PDF: {pdf_path}") @@ -221,13 +305,15 @@ def apply_actions_and_regenerate_grouped( if has_notes: width, height = sub_note.size if old_header_height > 0: - header = sub_note.crop((0, 0, width, min(height, old_header_height))) + header = sub_note.crop( + (0, 0, width, min(height, old_header_height)) + ) final_image.paste(header, (0, 0), mask=header) if height > old_header_height: body = sub_note.crop((0, old_header_height, width, height)) final_image.paste(body, (0, new_header_height), mask=body) - if label in dirty_labels or has_notes: + if label in dirty_labels or has_notes or selected_labels: dirty_images[label] = final_image logs.append(f" Saved dirty image: {label}.jpg") concat_images.append(final_image) @@ -237,33 +323,54 @@ def apply_actions_and_regenerate_grouped( feedback.get("to_delete", False) for feedback in feedbacks ) if not perfect or has_notes: - extras = get_extra_pdfs_as_images( - workspace.root, label, annotating, all_labels + extras = ( + get_extra_pdfs_as_images(workspace.root, label, annotating, all_labels) + if annotation_dir == "BGnot" + else [] ) filtered_groups.append([*extras, final_image]) concat_image = concatenate(concat_images) - with staged_files(output_dir) as staging: + if incomplete: + return ExitCode.PARTIAL, "\n".join(logs) + with staged_files(output_dir, remove=("Concat_F.pdf", "Concat_F.jpg")) as staging: + if simple_layout is not None: + simple_layout["replaced"] = sorted( + set(simple_layout["replaced"]) | selected_labels + ) + atomic_write_json(staging / "refaire_simple_layout.json", simple_layout) for label, image in dirty_images.items(): image.save(staging / f"{label}.jpg") atomic_write_json(staging / "score.json", scores) if concat_image is not None: concat_image.save(staging / "Concat.jpg") if filtered_groups: - save_paginated_pdf(filtered_groups, staging / "Concat_F.pdf") + if annotation_dir == "BGnot": + save_paginated_pdf(filtered_groups, staging / "Concat_F.pdf") + else: + filtered_image = concatenate( + [image for group in filtered_groups for image in group] + ) + filtered_image.save(staging / "Concat_F.jpg") logs.append(f" Saved regenerated files in {output_dir}") status = ExitCode.PARTIAL if incomplete else ExitCode.SUCCESS return status, "\n".join(logs) -def _read_refaire(workspace: EvaluationWorkspace) -> tuple[RefaireList, dict[str, list[str]]]: +def _read_refaire( + workspace: EvaluationWorkspace, +) -> tuple[RefaireList, dict[str, list[str]]]: loaded = read_json(workspace.refaire_file) if not isinstance(loaded, list): raise TypeError("refaire.json must contain a JSON array") entries: RefaireList = [] by_student: dict[str, list[str]] = {} for entry in loaded: - if not isinstance(entry, list) or len(entry) != 2 or not isinstance(entry[1], list): + if ( + not isinstance(entry, list) + or len(entry) != 2 + or not isinstance(entry[1], list) + ): raise TypeError(f"Malformed refaire entry: {entry!r}") copy_name, labels = entry student_id = str(copy_name).removeprefix("Copie") @@ -273,14 +380,78 @@ def _read_refaire(workspace: EvaluationWorkspace) -> tuple[RefaireList, dict[str return entries, by_student +def _scan_redo_annotations( + directory: Path, + expected: dict[str, set[str]], +) -> tuple[dict[str, list[dict[str, Any]]], dict[str, LabelNotes], set[str]]: + """Read either grouped or per-copy redo PDFs using their student/label metadata.""" + actions: dict[str, list[dict[str, Any]]] = defaultdict(list) + notes: dict[str, LabelNotes] = defaultdict(dict) + seen: dict[str, set[str]] = defaultdict(set) + incomplete: set[str] = set() + plans = [] + required = ("checkboxes.json", "Reference.jpg", "Concat_annotated.pdf") + for path in sorted(directory.iterdir()): + if not path.is_dir(): + continue + default_id = ( + path.name.removeprefix("Copie") if path.name.startswith("Copie") else None + ) + try: + metadata = read_json(path / "bnote.json") + pairs = [ + (str(item.get("id", default_id)), str(item["label"])) + for item in metadata["images"] + ] + except (OSError, ValueError, TypeError, KeyError) as exc: + print(f"Warning: unreadable redo metadata in {path}: {exc}") + incomplete.update(expected) + continue + students = { + student_id for student_id, _label in pairs if student_id in expected + } + if not students: + continue + for student_id, label in pairs: + if student_id not in expected: + continue + if label in seen[student_id]: + incomplete.add(student_id) + seen[student_id].add(label) + if any(not (path / name).is_file() for name in required): + print(f"Warning: missing returned redo inputs in {path}") + incomplete.update(students) + else: + plans.append((path, default_id, students)) + for student_id, labels in expected.items(): + if seen[student_id] != labels: + print( + f"Warning: redo labels do not match refaire.json for Copie{student_id}; regenerate BRnot" + ) + incomplete.add(student_id) + for path, default_id, students in plans: + if students <= incomplete: + continue + try: + result = _scan_annotation_directory( + path, default_student_id=default_id, required=True + ) + _merge_scan_result(actions, notes, result) + except (OSError, ValueError, TypeError) as exc: + print(f"Warning: could not read redo annotations in {path}: {exc}") + incomplete.update(students) + return dict(actions), dict(notes), incomplete + + def run( workspace: EvaluationWorkspace, *, refaire: bool = False, update_score: bool = False, + annotation_dir: str = "BGnot", ) -> ExitCode: workspace.require_files("labels", "correction.json") - workspace.require_directories("Copies", "Par label", "BGnot") + workspace.require_directories("Copies", "Par label", annotation_dir) refaire_list: RefaireList | None = None refaire_by_student: dict[str, list[str]] = {} if refaire: @@ -289,7 +460,12 @@ def run( refaire_list, refaire_by_student = _read_refaire(workspace) all_labels = utils.read_all_labels(workspace.root) - loaded = load_annotation_data(workspace, refaire_list=refaire_list) + loaded = load_annotation_data(workspace) + if refaire_list: + # Add explicitly requested answers without filtering out the rest of a copy. + selected_data = load_annotation_data(workspace, refaire_list=refaire_list) + for student_id, labels in selected_data.data.items(): + loaded.data.setdefault(student_id, {}).update(labels) for warning in loaded.warnings: print(f"Warning: {warning}") if not loaded.data: @@ -301,8 +477,10 @@ def run( only_ids = set(refaire_by_student) or None group_dirs = [ path - for path in workspace.annotation_dir("grouped").iterdir() - if path.is_dir() and not path.name.startswith("Copie") + for path in (workspace.root / annotation_dir).iterdir() + if annotation_dir == "BGnot" + and path.is_dir() + and not path.name.startswith("Copie") ] with concurrent.futures.ThreadPoolExecutor(max_workers=6) as executor: futures = [ @@ -312,39 +490,58 @@ def run( for future in concurrent.futures.as_completed(futures): _merge_scan_result(actions_by_student, notes_by_student, future.result()) - refaire_incomplete = False + if annotation_dir == "Bnot": + for student_id in refaire_by_student: + directory = workspace.root / annotation_dir / f"Copie{student_id}" + if directory.is_dir(): + _merge_scan_result( + actions_by_student, + notes_by_student, + _scan_annotation_directory( + directory, default_student_id=student_id + ), + ) + + skipped_students: set[str] = set() if refaire: - for student_id, requested_labels in refaire_by_student.items(): - selected = requested_labels or list(loaded.data.get(student_id, {})) - selected_set = set(selected) - directory = workspace.annotation_dir("refaire") / f"Copie{student_id}" - if not directory.is_dir(): - print(f"Warning: missing refaire annotation directory {directory}") - refaire_incomplete = True + expected = { + student_id: set(labels or loaded.data.get(student_id, {})) + for student_id, labels in refaire_by_student.items() + } + redo_actions, redo_notes, skipped_students = _scan_redo_annotations( + workspace.annotation_dir("refaire"), expected + ) + for student_id, selected in expected.items(): + if student_id in skipped_students: continue actions_by_student[student_id] = [ action for action in actions_by_student[student_id] - if str(action.get("label")) not in selected_set + if str(action.get("label")) not in selected ] for label in selected: notes_by_student[student_id].pop(label, None) - refaire_actions, refaire_notes = _scan_annotation_directory( - directory, default_student_id=student_id + actions_by_student[student_id].extend( + action + for action in redo_actions.get(student_id, []) + if str(action.get("label")) in selected + ) + notes_by_student[student_id].update( + { + label: note + for label, note in redo_notes.get(student_id, {}).items() + if label in selected + } ) - for action in refaire_actions.get(student_id, []): - if str(action.get("label")) in selected_set: - actions_by_student[student_id].append(action) - for label, note in refaire_notes.get(student_id, {}).items(): - if label in selected_set: - notes_by_student[student_id][label] = note status = ( - ExitCode.PARTIAL - if loaded.warnings or refaire_incomplete - else ExitCode.SUCCESS + ExitCode.PARTIAL if loaded.warnings or skipped_students else ExitCode.SUCCESS + ) + student_ids = ( + list(refaire_by_student) + if refaire + else sorted(loaded.data, key=utils.natural_key) ) - student_ids = list(refaire_by_student) if refaire else sorted(loaded.data, key=utils.natural_key) with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor: futures = { executor.submit( @@ -356,9 +553,15 @@ def run( notes_by_student[student_id], all_labels, update_score=update_score, + annotation_dir=annotation_dir, + selected_labels=( + set(refaire_by_student[student_id] or loaded.data[student_id]) + if refaire + else None + ), ): student_id for student_id in student_ids - if student_id in loaded.data + if student_id in loaded.data and student_id not in skipped_students } for future in concurrent.futures.as_completed(futures): result, output = future.result() @@ -370,6 +573,12 @@ def run( def build_parser() -> argparse.ArgumentParser: parser = evaluation_parser("Read grouped annotations and regenerate copies") + parser.add_argument( + "--annotation-dir", + choices=("BGnot", "Bnot", "Anot"), + default="BGnot", + help="Original annotation directory for --refaire (default: BGnot)", + ) parser.add_argument( "--refaire", action="store_true", @@ -387,10 +596,13 @@ def main(argv: Sequence[str] | None = None) -> int: parser = build_parser() def handle(args: argparse.Namespace) -> ExitCode: + if args.annotation_dir != "BGnot" and not args.refaire: + parser.error("--annotation-dir requires --refaire") return run( workspace_from_args(args), refaire=args.refaire, update_score=args.update_score, + annotation_dir=args.annotation_dir, ) return execute(parser, argv, handle) @@ -398,4 +610,3 @@ def main(argv: Sequence[str] | None = None) -> int: if __name__ == "__main__": raise SystemExit(main()) - diff --git a/copienator_gui/app.py b/copienator_gui/app.py index f0fd8fc..6351473 100644 --- a/copienator_gui/app.py +++ b/copienator_gui/app.py @@ -3,11 +3,12 @@ from __future__ import annotations import os import queue import tkinter as tk +from dataclasses import replace from pathlib import Path from tkinter import filedialog, messagebox, ttk from typing import Any -from copienator import ExitCode +from copienator import EvaluationWorkspace, ExitCode, atomic_write_json from copienator.platform import ( WindowsLabelError, add_platform_executable_paths, @@ -16,6 +17,14 @@ from copienator.platform import ( ) from .diagnostics import collect_diagnostics +from .refaire import SECTION as REFAIRE_SECTION +from .refaire import ( + RefaireSelection, + available_copies, + load_selection, + resolve_layout, + validate_selection, +) from .runner import ProcessRunner from .state import StateStore from .workflow import ( @@ -154,6 +163,9 @@ class CopienatorApp(tk.Tk): self.arg_vars: dict[str, tk.Variable] = {} self.copy_paths: dict[str, Path] = {} self._rendering = False + self.refaire_panel: RefaireSelection | None = None + self.pending_refaire_commands: list[list[str]] = [] + self.refaire_command_index = 0 self.title("Copienator — assistant de correction") self.geometry("1180x820") @@ -245,8 +257,19 @@ class CopienatorApp(tk.Tk): self.title_label.grid(row=0, column=0, sticky="w") self.description_label = ttk.Label(self.detail, text="", wraplength=680, justify="left") self.description_label.grid(row=1, column=0, sticky="ew", pady=(5, 8)) - self.form = ttk.Frame(self.detail) - self.form.grid(row=2, column=0, sticky="nsew") + form_container = ttk.Frame(self.detail) + form_container.grid(row=2, column=0, sticky="nsew") + form_container.columnconfigure(0, weight=1) + form_container.rowconfigure(0, weight=1) + self.form_canvas = tk.Canvas(form_container, highlightthickness=0, width=1, height=1) + self.form_canvas.grid(row=0, column=0, sticky="nsew") + form_scroll = ttk.Scrollbar(form_container, orient="vertical", command=self.form_canvas.yview) + form_scroll.grid(row=0, column=1, sticky="ns") + self.form_canvas.configure(yscrollcommand=form_scroll.set) + self.form = ttk.Frame(self.form_canvas) + form_window = self.form_canvas.create_window((0, 0), window=self.form, anchor="nw") + self.form.bind("", lambda _event: self.form_canvas.configure(scrollregion=self.form_canvas.bbox("all"))) + self.form_canvas.bind("", lambda event: self.form_canvas.itemconfigure(form_window, width=event.width)) self.form.columnconfigure(1, weight=1) command_box = ttk.LabelFrame(self.detail, text="Commande", padding=6) @@ -362,6 +385,9 @@ class CopienatorApp(tk.Tk): self.proxy_entry.configure(state="normal" if self.use_proxy_var.get() else "disabled") def _load_evaluation(self) -> None: + if self.active_step_id or self.runner.running: + messagebox.showinfo("Traitement en cours", "Attendez la fin du traitement avant de changer d’évaluation.") + return evaluation = self.evaluation if not evaluation or not evaluation.is_dir(): messagebox.showerror("Dossier invalide", "Choisissez un dossier d’évaluation existant.") @@ -392,13 +418,14 @@ class CopienatorApp(tk.Tk): def _populate_tree(self) -> None: selected = self.current_step.id if self.current_step else None + opened = {self.tree.item(item, "text"): self.tree.item(item, "open") for item in self.tree.get_children()} self.tree.delete(*self.tree.get_children()) section_items: dict[str, str] = {} for step in self.steps: if step.section not in section_items: section_id = f"section:{len(section_items)}" section_items[step.section] = section_id - self.tree.insert("", "end", iid=section_id, text=step.section, values=("",), open=True) + self.tree.insert("", "end", iid=section_id, text=step.section, values=("",), open=opened.get(step.section, step.section != REFAIRE_SECTION)) status = self._step_status(step) suffix = " (facultative)" if step.optional else "" self.tree.insert( @@ -513,9 +540,16 @@ class CopienatorApp(tk.Tk): if not step: return self._rendering = True + redo = step.section == REFAIRE_SECTION + self.console.configure(height=8 if redo else 14) + self.rowconfigure(1, weight=4 if redo else 3) + self.rowconfigure(2, weight=1 if redo else 2) + self.form_canvas.configure(height=(340 if step.id == "refaire_selection" else 260) if redo else 200) + self.form_canvas.yview_moveto(0) for child in self.form.winfo_children(): child.destroy() self.arg_vars.clear() + self.refaire_panel = None entry = self.state_store.step(step.id) if self.state_store.evaluation else {} saved_variant = entry.get("variant", step.variants[0].id) if saved_variant not in {variant.id for variant in step.variants}: @@ -544,6 +578,8 @@ class CopienatorApp(tk.Tk): variant = self._current_variant() evaluation_arg = self._evaluation_arg() for spec in step.arguments: + if step.section == REFAIRE_SECTION: + continue if spec.variants and variant.id not in spec.variants: continue value = values.get(spec.name, value_for_default(spec.default, evaluation_arg)) @@ -582,27 +618,44 @@ class CopienatorApp(tk.Tk): row += 1 row += 1 - if not step.is_manual: + if not step.is_manual and step.section != REFAIRE_SECTION: ttk.Label(self.form, text="Arguments supplémentaires").grid( row=row, column=0, sticky="w", pady=(10, 4), padx=(0, 8) ) ttk.Entry(self.form, textvariable=self.extra_var).grid(row=row, column=1, sticky="ew", pady=(10, 4)) - self.run_button.configure(text="Marquer terminée" if step.is_manual else "Exécuter") + self.run_button.configure(text="Enregistrer la sélection" if step.id == "refaire_selection" else ("Marquer terminée" if step.is_manual else "Exécuter")) self.skip_button.configure(state="normal" if step.optional else "disabled") self._rendering = False self._update_command_preview() self._update_controls() def _render_context_controls(self, step: StepDefinition, row: int) -> int: - if step.id == "review_persp": - ttk.Button( - self.form, - text="Ouvrir le dossier Persp", - command=self._open_persp, - ).grid(row=row, column=0, columnspan=2, sticky="w", pady=(0, 8)) - row += 1 - + if step.section == REFAIRE_SECTION: + evaluation = self.evaluation + if step.id in {"refaire_selection", "refaire_correct"}: + row = self._correction_folder_buttons(row) + if step.id == "refaire_selection" and evaluation: + choices, preferred = self._annotation_directory_choices("export") + draft = self.state_store.step(step.id).get("values", {}) + self.refaire_panel = RefaireSelection(self.form, evaluation, draft, choices, preferred) + self.refaire_panel.grid(row=row, column=0, columnspan=2, sticky="nsew") + else: + try: + entries = self._refaire_selection() + summary = f"{len(entries)} copie(s) sélectionnée(s) :\n" + " ; ".join(f"{name} : {', '.join(labels) or 'toute la copie'}" for name, labels in entries[:5]) + if len(entries) > 5: + summary += f" ; et {len(entries) - 5} autres (voir la sélection)." + source = self.state_store.step("refaire_selection").get("values", {}).get("annotation_dir", "") + choice = self.state_store.step("refaire_selection").get("values", {}).get("layout", "auto") + layout = resolve_layout(entries, EvaluationWorkspace(evaluation).read_labels(), choice) + summary += f"\nPassage principal : {source} — PDF à vérifier : {'par question' if layout == 'grouped' else 'par copie'}" + except (OSError, ValueError, TypeError) as exc: + summary = str(exc) + ttk.Label(self.form, text=summary, wraplength=600, justify="left").grid(row=row, column=0, columnspan=2, sticky="w") + return row + 1 + if step.id in {"review_persp", "correction"}: + row = self._correction_folder_buttons(row) if step.section == "Prétraitement des copies": evaluation = self.evaluation paths = copy_pdf_paths(evaluation) if evaluation else [] @@ -646,6 +699,14 @@ class CopienatorApp(tk.Tk): row += 1 return row + def _correction_folder_buttons(self, row: int) -> int: + buttons = ttk.Frame(self.form) + buttons.grid(row=row, column=0, columnspan=2, sticky="w", pady=(0, 6)) + for directory, label in (("Sol", "Corrigés (Sol)"), ("Persp", "Consignes de notation (Persp)")): + ttk.Button(buttons, text=label, + command=lambda name=directory: self._open_desktop_path(self.evaluation / name if self.evaluation else None, f"dossier {name}")).pack(side="left", padx=(0, 6)) + return row + 1 + def _annotation_directory_choices(self, step_id: str) -> tuple[tuple[str, ...], str]: evaluation = self.evaluation detected = detected_annotation_directories(evaluation) if evaluation else () @@ -696,6 +757,10 @@ class CopienatorApp(tk.Tk): return next((variant for variant in self.current_step.variants if variant.id == selected), self.current_step.variants[0]) def _values(self) -> dict[str, object]: + if self.current_step and self.current_step.section == REFAIRE_SECTION: + if self.current_step.id == "refaire_selection" and self.refaire_panel: + return self.refaire_panel.values() + return {"target": self._evaluation_arg(), "annotation_dir": self.state_store.step("refaire_selection").get("values", {}).get("annotation_dir", "BGnot")} return {name: variable.get() for name, variable in self.arg_vars.items()} def _save_current_form(self) -> None: @@ -715,9 +780,62 @@ class CopienatorApp(tk.Tk): evaluation = self.evaluation return evaluation_argument(self.repository, evaluation) if evaluation else "<évaluation>" + def _progression_ids(self, step_id: str) -> list[str]: + redo = step_id.startswith("refaire_") + return [step.id for step in self.steps if step.id.startswith("refaire_") == redo] + + def _refaire_selection(self) -> list[list]: + evaluation = self.evaluation + if not evaluation or not self.state_store.evaluation: + raise ValueError("Chargez d’abord une évaluation.") + entry = self.state_store.step("refaire_selection") + if entry.get("status") != "success": + raise ValueError("Enregistrez d’abord les copies et les questions à refaire.") + values = entry.get("values", {}) + selection = load_selection(evaluation) + if selection != validate_selection(values.get("selection"), available_copies(evaluation), EvaluationWorkspace(evaluation).read_labels()): + raise ValueError("La sélection a changé. Enregistrez-la avant de continuer.") + source = values.get("annotation_dir") + if source not in ANNOTATION_DIRECTORIES or not (evaluation / source).is_dir(): + raise ValueError("Choisissez un dossier existant pour le passage principal.") + return selection + + def _save_refaire_selection(self) -> None: + if not self.refaire_panel or not self.evaluation: + return + try: + values = self.refaire_panel.values() + entries = validate_selection(values["selection"], available_copies(self.evaluation), EvaluationWorkspace(self.evaluation).read_labels()) + source = values["annotation_dir"] + if source not in detected_annotation_directories(self.evaluation): + raise ValueError("Choisissez un dossier existant pour le passage principal.") + atomic_write_json(self.evaluation / "refaire.json", entries) + self._mark_step("success") + self._move_selection_from("refaire_selection", 1) + except (OSError, ValueError, TypeError) as exc: + messagebox.showerror("Sélection à vérifier", str(exc)) + + def _refaire_commands(self) -> list[list[str]]: + assert self.current_step is not None + selection = self._refaire_selection() + values = self._values() + targets = [self._evaluation_arg()] + if self.current_step.id in {"refaire_review", "refaire_split"}: + copies = available_copies(self.evaluation) + targets = [str(copies[name]) for name, _labels in selection] + variant = self._current_variant() + if self.current_step.id == "refaire_annotate": + choice = self.state_store.step("refaire_selection").get("values", {}).get("layout", "auto") + layout = resolve_layout(selection, EvaluationWorkspace(self.evaluation).read_labels(), choice) + variant = replace(variant, program="annotate-grouped" if layout == "grouped" else "annotate-checks") + return [build_command(self.repository, self.current_step, variant, + {**values, "target": target}, self._evaluation_arg()) for target in targets] + def _make_command(self) -> list[str]: if not self.current_step: return [] + if self.current_step.section == REFAIRE_SECTION: + return self._refaire_commands()[0] return build_command( self.repository, self.current_step, @@ -731,11 +849,15 @@ class CopienatorApp(tk.Tk): if self._rendering or not self.current_step: return if self.current_step.is_manual: - self.command_var.set("Étape manuelle — aucune commande ne sera exécutée.") + self.command_var.set("La sélection sera utilisée pour toutes les étapes de ce parcours." if self.current_step.id == "refaire_selection" else "Étape manuelle — aucune commande ne sera exécutée.") return try: - self.command_var.set(command_display(self._make_command())) - except ValueError as exc: + commands = self._refaire_commands() if self.current_step.section == REFAIRE_SECTION else [self._make_command()] + preview = "\n".join(command_display(command) for command in commands[:2]) + if len(commands) > 2: + preview += f"\nPuis {len(commands) - 2} autres copies, successivement." + self.command_var.set(preview) + except (OSError, ValueError, TypeError) as exc: self.command_var.set(f"Arguments invalides : {exc}") def _browse_target_file(self, variable: tk.Variable) -> None: @@ -764,7 +886,7 @@ class CopienatorApp(tk.Tk): return False try: self._make_command() - except ValueError as exc: + except (OSError, ValueError, TypeError) as exc: messagebox.showerror("Arguments invalides", str(exc)) return False return True @@ -775,11 +897,22 @@ class CopienatorApp(tk.Tk): if not step or not evaluation or not self.state_store.evaluation: messagebox.showerror("Évaluation absente", "Chargez d’abord un dossier d’évaluation.") return - if self.runner.running: + if self.active_step_id or self.runner.running: messagebox.showwarning("Traitement en cours", "Interrompez le traitement actuel avant d’en lancer un autre.") return + if step.id == "refaire_selection": + self._save_refaire_selection() + return + if step.section == REFAIRE_SECTION: + try: + self._refaire_selection() + except (OSError, ValueError, TypeError) as exc: + messagebox.showerror("Sélection à vérifier", str(exc)) + return if step.is_manual: self._mark_step("success") + if step.section == REFAIRE_SECTION: + self._move_selection_from(step.id, 1) return if os.name == "nt" and step.id != "statement": labels_path = evaluation / "labels" @@ -816,7 +949,7 @@ class CopienatorApp(tk.Tk): self._save_current_form() run_values = self._values() - ordered_ids = [item.id for item in self.steps] + ordered_ids = self._progression_ids(step.id) self.state_store.invalidate_after(ordered_ids, step.id) self.state_store.update_step( step.id, @@ -837,6 +970,8 @@ class CopienatorApp(tk.Tk): self.use_proxy_var.get(), ) + self.pending_refaire_commands = self._refaire_commands()[1:] if step.section == REFAIRE_SECTION else [] + self.refaire_command_index = 1 self._append_console(f"\n$ {command_display(command)}\n") try: self.runner.start(command, self.repository, environment, log_path) @@ -846,6 +981,7 @@ class CopienatorApp(tk.Tk): {"step": step.id, "command": command_display(command), "status": "failed", "error": str(exc)} ) self.active_step_id = None + self.pending_refaire_commands = [] self._append_console(f"Impossible de lancer la commande : {exc}\n") messagebox.showerror("Échec du lancement", str(exc)) self._populate_tree() @@ -854,10 +990,10 @@ class CopienatorApp(tk.Tk): def _mark_step( self, status: str, *, automatic: bool = False, reason: str | None = None ) -> None: - if not self.current_step or not self.state_store.evaluation: + if not self.current_step or not self.state_store.evaluation or self.active_step_id: return self._save_current_form() - self.state_store.invalidate_after([item.id for item in self.steps], self.current_step.id) + self.state_store.invalidate_after(self._progression_ids(self.current_step.id), self.current_step.id) self.state_store.update_step(self.current_step.id, status=status) history: dict[str, object] = { "step": self.current_step.id, @@ -875,7 +1011,10 @@ class CopienatorApp(tk.Tk): def _skip_step(self) -> None: if self.current_step and self.current_step.optional: + step_id = self.current_step.id self._mark_step("skipped") + if step_id.startswith("refaire_"): + self._move_selection_from(step_id, 1) def _poll_runner(self) -> None: while True: @@ -899,6 +1038,26 @@ class CopienatorApp(tk.Tk): if not step_id: return status = process_status(return_code, interrupted) + if status == "success" and self.pending_refaire_commands: + command = self.pending_refaire_commands.pop(0) + self.refaire_command_index += 1 + self._append_console(f"\n$ {command_display(command)}\n") + self.state_store.update_step(step_id, command=command_display(command)) + try: + self.runner.start(command, self.repository, + build_runner_environment(os.environ, self.api_key_var.get(), self.proxy_var.get(), self.use_proxy_var.get()), + self.state_store.workspace.log_path(f"{step_id}_{self.refaire_command_index}")) + self._update_controls() + return + except (OSError, RuntimeError) as exc: + self._append_console(f"Impossible de lancer la copie suivante : {exc}\n") + status, return_code = "failed", 1 + self.pending_refaire_commands = [] + if step_id == "refaire_merge" and status == "success": + for downstream in ("giving_names", "update_ods", "final_score", "personal_deploy", "personal_sent"): + entry = self.state_store.step(downstream) + if entry.get("status") in {"success", "detected", "skipped"}: + self.state_store.update_step(downstream, status="stale") if step_id == "clean" and status == "success": self._append_console( f"\n[Terminé — code {return_code} — {STATUS_LABELS[status]}]\n" @@ -977,8 +1136,9 @@ class CopienatorApp(tk.Tk): self.console.configure(state="disabled") def _update_controls(self) -> None: - running = self.runner.running + running = bool(self.active_step_id) or self.runner.running self.run_button.configure(state="disabled" if running or not self.current_step else "normal") + self.skip_button.configure(state="normal" if not running and self.current_step and self.current_step.optional else "disabled") self.interrupt_button.configure(state="normal" if running else "disabled") self.force_button.configure(state="normal" if running else "disabled") self.send_button.configure(state="normal" if running else "disabled") @@ -990,7 +1150,7 @@ class CopienatorApp(tk.Tk): self._move_selection_from(self.current_step.id, delta) def _move_selection_from(self, step_id: str, delta: int) -> None: - ids = [step.id for step in self.steps] + ids = self._progression_ids(step_id) try: index = ids.index(step_id) except ValueError: diff --git a/copienator_gui/refaire.py b/copienator_gui/refaire.py new file mode 100644 index 0000000..397b0ba --- /dev/null +++ b/copienator_gui/refaire.py @@ -0,0 +1,244 @@ +"""Selection and command planning for the optional redo workflow.""" + +from __future__ import annotations + +import re +import tkinter as tk +from pathlib import Path +from tkinter import ttk + +from copienator import EvaluationWorkspace, read_json +from copienator.utils import natural_key + +SECTION = "Refaire des copies (facultatif)" +ALL_LABELS = "Toute la copie" +ALL_COPIES = "Toutes les copies" +LAYOUTS = { + "Automatique": "auto", + "Par question (groupé)": "grouped", + "Par copie": "copies", +} + + +def resolve_layout( + selection: list[list], labels: list[str], choice: str = "auto" +) -> str: + if choice in {"grouped", "copies"}: + return choice + seen = set() + for _name, selected in selection: + current = set(selected or labels) + if seen & current: + return "grouped" + seen.update(current) + return "copies" + + +def copies_with_answer(copies: dict[str, Path], label: str) -> list[str]: + return [ + name + for name, path in copies.items() + if any( + (path.with_suffix("") / f"{label}{suffix}.pdf").is_file() + for suffix in ("", "_new") + ) + ] + + +def available_copies(evaluation: Path) -> dict[str, Path]: + return { + path.stem: path + for path in sorted((evaluation / "Copies").glob("Copie*.pdf"), key=natural_key) + if re.fullmatch(r"Copie\d+", path.stem) + } + + +def validate_selection( + entries: object, copies: dict[str, Path], labels: list[str] +) -> list[list]: + if not isinstance(entries, list) or not entries: + raise ValueError("Ajoutez au moins une copie à refaire.") + result = {} + for entry in entries: + if not isinstance(entry, list) or len(entry) != 2: + raise ValueError("Sélection de copies invalide.") + name, selected = entry + if not isinstance(name, str) or name not in copies: + raise ValueError(f"Copie introuvable : {name}") + if not isinstance(selected, list) or any( + not isinstance(label, str) or label not in labels for label in selected + ): + raise ValueError(f"Question inconnue pour {name}. Reprenez la sélection.") + if name in result: + raise ValueError(f"Copie sélectionnée plusieurs fois : {name}") + result[name] = sorted(set(selected), key=natural_key) + return [[name, result[name]] for name in sorted(result, key=natural_key)] + + +def load_selection(evaluation: Path) -> list[list]: + return validate_selection( + read_json(evaluation / "refaire.json"), + available_copies(evaluation), + EvaluationWorkspace(evaluation).read_labels(), + ) + + +class RefaireSelection(ttk.Frame): + def __init__( + self, + parent, + evaluation: Path, + draft: dict, + directories: tuple[str, ...], + preferred: str, + ): + super().__init__(parent) + self.columnconfigure(1, weight=1) + self.copies = available_copies(evaluation) + self.labels = ( + EvaluationWorkspace(evaluation).read_labels() + if (evaluation / "labels").is_file() + else [] + ) + self.entries = {} + self.error = "" + try: + entries = draft.get("selection") + if entries is None: + entries = read_json(evaluation / "refaire.json", default=[]) + if entries: + self.entries = dict( + validate_selection(entries, self.copies, self.labels) + ) + except (OSError, ValueError, TypeError) as exc: + self.error = str(exc) + self.copy_var = tk.StringVar(value=next(iter(self.copies), "")) + self.label_var = tk.StringVar(value=ALL_LABELS) + source = draft.get("annotation_dir", preferred) + self.source_var = tk.StringVar( + value=source if source in directories else preferred + ) + ttk.Label(self, text="Copie").grid(row=0, column=0, sticky="w", padx=(0, 8)) + ttk.Combobox( + self, + textvariable=self.copy_var, + values=(ALL_COPIES, *self.copies), + state="readonly", + width=12, + ).grid(row=0, column=1, sticky="ew") + ttk.Label(self, text="Question").grid(row=1, column=0, sticky="w", pady=5) + ttk.Combobox( + self, + textvariable=self.label_var, + values=(ALL_LABELS, *self.labels), + state="readonly", + width=12, + ).grid(row=1, column=1, sticky="ew", pady=5) + ttk.Button(self, text="+ Ajouter", command=self.add).grid( + row=1, column=2, padx=(8, 0) + ) + self.table = ttk.Treeview( + self, columns=("labels",), height=4, selectmode="browse" + ) + self.table.heading("#0", text="Copie") + self.table.heading("labels", text="Questions à refaire") + self.table.column("#0", width=100, stretch=False) + self.table.column("labels", width=330) + self.table.grid(row=4, column=0, columnspan=3, sticky="nsew") + scrollbar = ttk.Scrollbar(self, orient="vertical", command=self.table.yview) + scrollbar.grid(row=4, column=3, sticky="ns") + self.table.configure(yscrollcommand=scrollbar.set) + ttk.Button( + self, text="Retirer la copie sélectionnée", command=self.remove + ).grid(row=5, column=0, columnspan=3, sticky="w", pady=5) + ttk.Label(self, text="Passage principal").grid( + row=3, column=0, sticky="w", padx=(0, 8) + ) + ttk.Combobox( + self, + textvariable=self.source_var, + values=directories, + state="readonly", + width=12, + ).grid(row=3, column=1, sticky="ew") + layout = draft.get("layout", "auto") + self.layout_var = tk.StringVar( + value=next( + (name for name, code in LAYOUTS.items() if code == layout), + "Automatique", + ) + ) + ttk.Label(self, text="PDF à vérifier").grid(row=2, column=0, sticky="w") + ttk.Combobox( + self, + textvariable=self.layout_var, + values=tuple(LAYOUTS), + state="readonly", + width=12, + ).grid(row=2, column=1, sticky="ew") + self.message_var = tk.StringVar( + value=self.error + or "Choisissez « Toutes les copies » pour refaire une question dans toute la classe." + ) + help_label = ttk.Label( + self, + textvariable=self.message_var, + wraplength=500, + ) + help_label.grid(row=6, column=0, columnspan=3, sticky="w", pady=5) + self.bind( + "", + lambda event: help_label.configure(wraplength=max(200, event.width - 10)), + ) + self.refresh() + + def refresh(self): + self.table.delete(*self.table.get_children()) + for name in sorted(self.entries, key=natural_key): + self.table.insert( + "", + "end", + iid=name, + text=name, + values=(", ".join(self.entries[name]) or ALL_LABELS,), + ) + + def add(self): + name, label = self.copy_var.get(), self.label_var.get() + if name not in (ALL_COPIES, *self.copies) or label not in ( + ALL_LABELS, + *self.labels, + ): + return + names = ( + [name] + if name != ALL_COPIES + else list(self.copies) + if label == ALL_LABELS + else copies_with_answer(self.copies, label) + ) + for copy_name in names: + # Adding a question must not narrow a copy already selected in full. + if label == ALL_LABELS: + self.entries[copy_name] = [] + elif copy_name not in self.entries or self.entries[copy_name]: + self.entries[copy_name] = sorted( + set(self.entries.get(copy_name, [])) | {label}, key=natural_key + ) + message = f"{len(names)} copie(s) ajoutée(s)." + if name == ALL_COPIES and len(names) < len(self.copies): + message += f" {len(self.copies) - len(names)} sans réponse découpée pour cette question." + self.message_var.set(message) + self.refresh() + + def remove(self): + for name in self.table.selection(): + self.entries.pop(name, None) + self.refresh() + + def values(self): + return { + "selection": [[name, labels] for name, labels in self.entries.items()], + "annotation_dir": self.source_var.get(), + "layout": LAYOUTS[self.layout_var.get()], + } diff --git a/copienator_gui/workflow.py b/copienator_gui/workflow.py index 90b51af..2271fc6 100644 --- a/copienator_gui/workflow.py +++ b/copienator_gui/workflow.py @@ -505,9 +505,46 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]: requires=("Copies", "correction.json", "A Rendre"), ), ] + steps[-1:-1] = build_refaire_workflow() return [step for step in steps if show_personal_steps or not step.personal] +def build_refaire_workflow() -> list[StepDefinition]: + from .refaire import SECTION + + selection = StepDefinition( + "refaire_selection", SECTION, "Choisir les copies et les questions", + "Sélectionnez les copies et les questions à refaire, puis enregistrez la sélection. " + "Le passage principal doit être terminé ; conservez ses annotations.", + (CommandVariant("default", "Sélection", None, "manual"),), + requires=("Copies", "labels", "correction.json"), + ) + definitions = [ + ("review", "Reprendre le découpage", "Vérifiez et ajustez les labels des copies sélectionnées. Chaque copie s’ouvre à son tour. Fermez la fenêtre pour passer à la suivante.", "review-labels", (), True), + ("split", "Redécouper les réponses", "À exécuter après une modification du découpage. Traite toutes les copies sélectionnées ; vérifiez les fichiers _new et _old en cas de résolution manuelle.", "split-answers", (), True), + ("correct", "Refaire la correction", "Relance la correction des seules questions sélectionnées. Peut être ignorée pour corriger manuellement les résultats.", "correct", ("--refaire",), True), + ("annotate", "Préparer les copies à vérifier", "Génère les questions sélectionnées avec des cases dans BRnot, par question ou par copie selon la sélection. Remplace le précédent passage dans BRnot.", "annotate-checks", ("--refaire", "--overwrite"), False), + ("export", "Exporter vers la tablette", "Exporte BRnot vers EXPORT_DIR. Retirez les anciens fichiers d’export avant le transfert.", "export", ("--refaire",), True), + ("tablet", "Vérifier sur la tablette", "Annotez les PDF exportés, puis placez les retours dans IMPORT_DIR sans changer leur nom (nom de groupe ou Copie01.pdf…). Retournez aussi les PDF sans modification. Retirez les anciens fichiers d’import.", None, (), False), + ("import", "Importer les copies vérifiées", "Importe les PDF retournés dans BRnot. Vous pouvez ignorer cette étape si les fichiers Concat_annotated.pdf y sont déjà en place.", "import", ("--refaire",), True), + ("merge", "Mettre à jour les copies finales", "Fusionne les questions refaites avec le reste de chaque copie dans le dossier du passage principal. Relancez ensuite la préparation de A Rendre, le calcul des notes et la diffusion.", "read-grouped", ("--refaire",), False), + ] + steps = [selection] + for suffix, title, description, program, flags, optional in definitions: + arguments = (arg_target(),) if program else () + if suffix == "merge": + arguments += (ArgumentSpec("annotation_dir", "Passage principal", "choice", "--annotation-dir", default="BGnot", choices=("BGnot", "Bnot", "Anot")),) + requirements = ("refaire.json", "Copies", "labels", "correction.json") + if suffix in {"export", "tablet", "import", "merge"}: + requirements += ("BRnot",) + steps.append(StepDefinition( + f"refaire_{suffix}", SECTION, title, description, + (CommandVariant("default", title, program, "python" if program else "manual", flags),), + arguments=arguments, optional=optional, requires=requirements, + )) + return steps + + def value_for_default(value: object, evaluation_arg: str) -> object: return evaluation_arg if value == EVALUATION else value diff --git a/tests/test_grouped_refaire.py b/tests/test_grouped_refaire.py new file mode 100644 index 0000000..84d5a2b --- /dev/null +++ b/tests/test_grouped_refaire.py @@ -0,0 +1,206 @@ +from __future__ import annotations + +import shutil +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from PIL import Image, ImageDraw + +from copienator import EvaluationWorkspace, ExitCode, atomic_write_json, read_json +from copienator.annotation_data import AnnotationLoadResult +from copienator.commands import annotating_by_label as grouped +from copienator.commands import annotating_with_checks as checks +from copienator.commands import export, import_annotations +from copienator.commands import reading_grouped_annotations as reader + + +class GroupedRedoTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.root = Path(self.temp.name) / "Exam" + for directory in ("Copies", "Par label", "BGnot", "BRnot"): + (self.root / directory).mkdir(parents=True) + (self.root / "labels").write_text("Ex 1\nEx 2\n") + atomic_write_json(self.root / "correction.json", {}) + atomic_write_json( + self.root / "refaire.json", [["Copie01", ["Ex 1"]], ["Copie02", ["Ex 1"]]] + ) + (self.root / "BRnot/previous.txt").write_text("previous redo") + (self.root / "BGnot/main.txt").write_text("main run") + self.workspace = EvaluationWorkspace(self.root) + self.data = {"01": {"Ex 1": {}}, "02": {"Ex 1": {}}} + + def tearDown(self): + self.temp.cleanup() + + @staticmethod + def render(item): + student_id, label, _content = item + image = Image.new("RGB", (100, 100), "white") + ImageDraw.Draw(image).rectangle((10, 10, 50, 50), outline="black", width=2) + return ( + student_id, + label, + image, + 0, + [ + { + "type": "score", + "label": label, + "value": 3, + "final_box": [10, 10, 50, 50], + } + ], + ) + + def generate(self): + with ( + patch.object( + grouped, + "load_annotation_data", + return_value=AnnotationLoadResult(self.data, []), + ) as load, + patch.object(grouped, "render_item", side_effect=self.render), + patch.object(grouped, "_load_label_groups") as label_groups, + ): + self.assertEqual( + grouped.run(self.workspace, refaire=True, overwrite=True), + ExitCode.SUCCESS, + ) + self.assertEqual( + load.call_args.kwargs["refaire_list"], + read_json(self.root / "refaire.json"), + ) + label_groups.assert_not_called() + directories = list((self.root / "BRnot").iterdir()) + self.assertEqual(len(directories), 1) + self.assertTrue(directories[0].is_dir()) + self.assertEqual((self.root / "BGnot/main.txt").read_text(), "main run") + return directories[0] + + def test_grouped_redo_export_import_and_actual_annotation_detection(self): + directory = self.generate() + metadata = read_json(directory / "bnote.json")["images"] + self.assertEqual( + [(item["id"], item["label"]) for item in metadata], + [("01", "Ex 1"), ("02", "Ex 1")], + ) + export_root = Path(self.temp.name) / "Export" + with patch.object(export, "EXPORT_DIR", export_root): + self.assertEqual(export.run(self.workspace, refaire=True), ExitCode.SUCCESS) + self.assertEqual(len(list((export_root / "Exam").glob("*.pdf"))), 1) + imported_root = Path(self.temp.name) / "Import" + imported_root.mkdir() + with Image.open(directory / "Reference.jpg") as reference: + annotated = reference.convert("RGB") + draw = ImageDraw.Draw(annotated) + draw.rectangle((17, 17, 43, 43), fill="black") + draw.rectangle((70, 170, 90, 190), fill="black") + annotated.save(imported_root / f"{directory.name}.pdf", "PDF", resolution=72) + with patch.object(import_annotations, "IMPORT_DIR", imported_root): + self.assertEqual( + import_annotations.run(self.workspace, refaire=True), ExitCode.SUCCESS + ) + actions, notes, incomplete = reader._scan_redo_annotations( + self.root / "BRnot", {"01": {"Ex 1"}, "02": {"Ex 1"}} + ) + self.assertFalse(incomplete) + self.assertEqual(actions["01"][0]["value"], 3) + self.assertFalse(actions.get("02")) + self.assertIn("Ex 1", notes["02"]) + full_data = {student: {"Ex 1": {}, "Ex 2": {}} for student in ("01", "02")} + for mode in ("BGnot", "Bnot", "Anot"): + (self.root / mode).mkdir(exist_ok=True) + with ( + self.subTest(mode=mode), + patch.object( + reader, + "load_annotation_data", + return_value=AnnotationLoadResult(full_data, []), + ), + patch.object( + reader, + "apply_actions_and_regenerate_grouped", + return_value=(ExitCode.SUCCESS, ""), + ) as regenerate, + ): + self.assertEqual( + reader.run(self.workspace, refaire=True, annotation_dir=mode), + ExitCode.SUCCESS, + ) + calls = {call.args[2]: call for call in regenerate.call_args_list} + self.assertEqual(set(calls), {"01", "02"}) + self.assertEqual(set(calls["01"].args[1]["01"]), {"Ex 1", "Ex 2"}) + self.assertEqual(calls["01"].args[3][0]["value"], 3) + self.assertIn("Ex 1", calls["02"].args[4]) + + def test_missing_group_leaves_affected_copy_incomplete(self): + directory = self.generate() + shutil.copy2(directory / "Concat.pdf", directory / "Concat_annotated.pdf") + extra = self.root / "BRnot/Ex 2 G1" + extra.mkdir() + atomic_write_json( + extra / "bnote.json", {"images": [{"id": "01", "label": "Ex 2"}]} + ) + _actions, _notes, incomplete = reader._scan_redo_annotations( + self.root / "BRnot", {"01": {"Ex 1", "Ex 2"}, "02": {"Ex 1"}} + ) + self.assertEqual(incomplete, {"01"}) + + def test_failed_generation_preserves_previous_redo_for_both_layouts(self): + for module, worker in ((grouped, "render_item"), (checks, "_render_student")): + with ( + self.subTest(module=module), + patch.object( + module, + "load_annotation_data", + return_value=AnnotationLoadResult(self.data, []), + ), + patch.object( + module, + worker, + return_value=None if module is grouped else "partial", + ), + ): + if module is grouped: + status = module.run(self.workspace, refaire=True, overwrite=True) + else: + status = module.run( + self.workspace, self.root, refaire=True, overwrite=True + ) + self.assertEqual(status, ExitCode.PARTIAL) + self.assertEqual( + (self.root / "BRnot/previous.txt").read_text(), "previous redo" + ) + + def test_switching_to_per_copy_removes_previous_group_layout(self): + self.generate() + + def render(_workspace, student_id, _labels, **kwargs): + output = kwargs["output_root"] / f"Copie{student_id}" + output.mkdir() + (output / "Concat.pdf").touch() + return "success" + + with ( + patch.object( + checks, + "load_annotation_data", + return_value=AnnotationLoadResult(self.data, []), + ), + patch.object(checks, "_render_student", side_effect=render), + ): + self.assertEqual( + checks.run(self.workspace, self.root, refaire=True, overwrite=True), + ExitCode.SUCCESS, + ) + self.assertEqual( + {path.name for path in (self.root / "BRnot").iterdir()}, + {"Copie01", "Copie02"}, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_gui_refaire.py b/tests/test_gui_refaire.py new file mode 100644 index 0000000..6a2b59c --- /dev/null +++ b/tests/test_gui_refaire.py @@ -0,0 +1,265 @@ +from __future__ import annotations + +import os +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from copienator import atomic_write_json, read_json +from copienator_gui.app import CopienatorApp +from copienator_gui.refaire import ( + ALL_COPIES, + SECTION, + available_copies, + resolve_layout, + validate_selection, +) +from copienator_gui.workflow import build_command, build_workflow + + +class SelectionTests(unittest.TestCase): + def test_selection_validates_and_deduplicates_labels(self): + copies = {"Copie01": Path("/tmp/Copie01.pdf")} + self.assertEqual( + validate_selection( + [["Copie01", ["Ex 2", "Ex 1", "Ex 2"]]], copies, ["Ex 1", "Ex 2"] + ), + [["Copie01", ["Ex 1", "Ex 2"]]], + ) + for invalid in ( + [], + [["missing", []]], + [["Copie01", ["unknown"]]], + [["../Copie01", []]], + ): + with self.subTest(invalid=invalid), self.assertRaises(ValueError): + validate_selection(invalid, copies, ["Ex 1"]) + + def test_automatic_layout_groups_shared_labels_and_honors_explicit_choice(self): + selection = [["Copie01", ["Ex 1"]], ["Copie02", ["Ex 1"]]] + self.assertEqual(resolve_layout(selection, ["Ex 1", "Ex 2"]), "grouped") + self.assertEqual(resolve_layout(selection, ["Ex 1"], "copies"), "copies") + self.assertEqual(resolve_layout([["Copie01", ["Ex 1"]]], ["Ex 1"]), "copies") + self.assertEqual( + resolve_layout([["Copie01", ["Ex 1"]], ["Copie02", []]], ["Ex 1"]), + "grouped", + ) + + def test_redo_steps_are_in_both_profiles_and_never_autostart(self): + for personal in (False, True): + steps = [ + step for step in build_workflow(personal) if step.section == SECTION + ] + self.assertEqual(len(steps), 9) + self.assertEqual(steps[0].id, "refaire_selection") + self.assertTrue(all(not step.auto_start_first_visit for step in steps)) + merge = steps[-1] + for mode in ("BGnot", "Bnot", "Anot"): + command = build_command( + Path.cwd(), + merge, + merge.variants[0], + {"target": "/tmp/Exam", "annotation_dir": mode}, + "/tmp/Exam", + ) + self.assertEqual( + command[4:], + [ + "read-grouped", + "/tmp/Exam", + "--refaire", + "--annotation-dir", + mode, + ], + ) + + +@unittest.skipUnless( + os.environ.get("DISPLAY"), "Tk tests require a display (use xvfb-run)" +) +class RefaireGuiTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.root = Path(self.temp.name) + (self.root / "Copies").mkdir() + (self.root / "Anot").mkdir() + (self.root / "Par label").mkdir() + (self.root / "BRnot").mkdir() + for name in ("Copie01", "Copie02"): + (self.root / "Copies" / f"{name}.pdf").touch() + (self.root / "labels").write_text("Ex 1\nEx 2\n", encoding="utf-8") + atomic_write_json(self.root / "correction.json", {}) + self.app = CopienatorApp(Path.cwd(), False, self.root) + self.app.update() + + def tearDown(self): + for callback in self.app.tk.splitlist(self.app.tk.call("after", "info")): + self.app.after_cancel(callback) + self.app.destroy() + self.temp.cleanup() + + def select(self, ident): + self.app.tree.selection_set(ident) + self.app.tree.see(ident) + self.app.update() + + def save_selection(self): + self.select("refaire_selection") + panel = self.app.refaire_panel + panel.label_var.set("Ex 1") + panel.add() + panel.label_var.set("Ex 2") + panel.add() + panel.copy_var.set("Copie02") + panel.label_var.set("Toute la copie") + panel.add() + self.app._run_current_step() + self.app.update() + + def test_collapsed_branch_is_separate_and_stays_open_when_refreshed(self): + section = self.app.tree.parent("refaire_selection") + self.assertFalse(self.app.tree.item(section, "open")) + self.assertNotIn("refaire_selection", self.app._progression_ids("giving_names")) + self.assertNotIn("clean", self.app._progression_ids("refaire_merge")) + self.select("refaire_selection") + self.app._populate_tree() + self.assertTrue(self.app.tree.item(section, "open")) + + def test_dropdown_selection_saves_json_and_drives_all_commands(self): + self.save_selection() + self.assertEqual( + read_json(self.root / "refaire.json"), + [["Copie01", ["Ex 1", "Ex 2"]], ["Copie02", []]], + ) + self.assertEqual(self.app.current_step.id, "refaire_review") + commands = self.app._refaire_commands() + self.assertEqual( + [command[5] for command in commands], + [str(path) for path in available_copies(self.root).values()], + ) + self.select("refaire_annotate") + self.assertEqual( + self.app._make_command()[4:], + ["annotate-grouped", str(self.root), "--refaire", "--overwrite"], + ) + self.select("refaire_merge") + self.assertEqual( + self.app._make_command()[4:], + ["read-grouped", str(self.root), "--refaire", "--annotation-dir", "Anot"], + ) + self.assertEqual(self.app.state_store.step("annotation").get("status"), None) + + def test_one_label_for_all_copies_and_layout_override(self): + for name in ("Copie01", "Copie02"): + directory = self.root / "Copies" / name + directory.mkdir() + (directory / "Ex 1.pdf").touch() + self.select("refaire_selection") + panel = self.app.refaire_panel + panel.copy_var.set(ALL_COPIES) + panel.label_var.set("Ex 1") + panel.add() + self.assertEqual(panel.entries, {"Copie01": ["Ex 1"], "Copie02": ["Ex 1"]}) + panel.layout_var.set("Par copie") + self.app._run_current_step() + self.app.update() + self.select("refaire_annotate") + self.assertEqual(self.app._make_command()[4], "annotate-checks") + + def test_bulk_add_skips_absent_answers_and_preserves_whole_copy_selection(self): + directory = self.root / "Copies/Copie01" + directory.mkdir() + (directory / "Ex 1_new.pdf").touch() + self.select("refaire_selection") + panel = self.app.refaire_panel + panel.copy_var.set("Copie01") + panel.add() + panel.copy_var.set(ALL_COPIES) + panel.label_var.set("Ex 1") + panel.add() + self.assertEqual(panel.entries, {"Copie01": []}) + self.assertIn("1 sans réponse", panel.message_var.get()) + + def test_correction_folder_buttons_open_expected_folders(self): + for name in ("Sol", "Persp"): + (self.root / name).mkdir() + self.select("refaire_selection") + buttons = self.app.form.winfo_children()[0].winfo_children() + with patch("copienator_gui.app.open_path") as opened: + for button in buttons: + button.invoke() + self.assertEqual( + [call.args[0] for call in opened.call_args_list], + [self.root / "Sol", self.root / "Persp"], + ) + + def test_small_window_keeps_selection_accessible_by_scrolling(self): + self.select("refaire_selection") + self.app.geometry("900x640") + self.app.update() + self.app.form_canvas.yview_moveto(1) + self.app.update() + self.assertAlmostEqual(self.app.form_canvas.yview()[1], 1.0) + self.assertLess( + self.app.run_button.winfo_rooty() + self.app.run_button.winfo_height(), + self.app.winfo_rooty() + self.app.winfo_height(), + ) + + def test_queue_runs_copies_in_order_and_stops_on_failure(self): + self.save_selection() + self.select("refaire_split") + with patch.object(self.app.runner, "start") as start: + self.app._run_current_step() + self.assertEqual(start.call_count, 1) + self.assertEqual(len(self.app.pending_refaire_commands), 1) + self.app._finish_process(0, False) + self.assertEqual(start.call_count, 2) + self.assertEqual(self.app.active_step_id, "refaire_split") + self.app._finish_process(1, False) + self.assertEqual( + self.app.state_store.step("refaire_split")["status"], "failed" + ) + self.assertIsNone(self.app.active_step_id) + self.assertFalse(self.app.pending_refaire_commands) + + def test_interruption_does_not_launch_next_copy(self): + self.save_selection() + self.select("refaire_split") + with patch.object(self.app.runner, "start") as start: + self.app._run_current_step() + self.app._finish_process(130, True) + self.assertEqual(start.call_count, 1) + self.assertFalse(self.app.pending_refaire_commands) + self.assertEqual( + self.app.state_store.step("refaire_split")["status"], "interrupted" + ) + + def test_unsaved_or_external_changes_block_commands(self): + self.save_selection() + atomic_write_json(self.root / "refaire.json", [["Copie02", []]]) + with self.assertRaisesRegex(ValueError, "sélection a changé"): + self.app._refaire_commands() + + def test_selection_reload_and_finalization_invalidation(self): + self.save_selection() + self.app.state_store.update_step("giving_names", status="success") + self.app.state_store.update_step( + "annotation", status="success", last_run_variant="simple" + ) + self.select("refaire_merge") + with patch.object(self.app.runner, "start"): + self.app._run_current_step() + self.app._finish_process(0, False) + self.app.update() + self.assertEqual(self.app.state_store.step("giving_names")["status"], "stale") + self.assertEqual(self.app.state_store.step("annotation")["status"], "success") + self.assertEqual(self.app.current_step.id, "refaire_merge") + self.select("refaire_selection") + self.assertEqual( + self.app.refaire_panel.entries, {"Copie01": ["Ex 1", "Ex 2"], "Copie02": []} + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_refaire.py b/tests/test_refaire.py new file mode 100644 index 0000000..dc5548b --- /dev/null +++ b/tests/test_refaire.py @@ -0,0 +1,280 @@ +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from PIL import Image + +from copienator import EvaluationWorkspace, ExitCode, atomic_write_json, read_json +from copienator.annotation_data import AnnotationLoadResult, _coordinate_index +from copienator.commands import reading_grouped_annotations as reader + + +class RefaireTests(unittest.TestCase): + def workspace(self, root, mode): + for name in ("Copies", "Par label", mode, "BRnot/Copie01"): + (root / name).mkdir(parents=True) + (root / "labels").write_text("Ex 1\nEx 2\n") + atomic_write_json(root / "correction.json", {}) + atomic_write_json(root / "refaire.json", [["Copie01", ["Ex 2"]]]) + for name in ( + "bnote.json", + "checkboxes.json", + "Reference.jpg", + "Concat_annotated.pdf", + ): + (root / "BRnot/Copie01" / name).touch() + atomic_write_json( + root / "BRnot/Copie01/bnote.json", {"images": [{"label": "Ex 2"}]} + ) + return EvaluationWorkspace(root) + + def test_reader_merges_full_copy_for_every_original_mode(self): + for mode in ("BGnot", "Bnot", "Anot"): + with self.subTest(mode=mode), tempfile.TemporaryDirectory() as tmp: + workspace = self.workspace(Path(tmp), mode) + data = {"01": {"Ex 1": {}, "Ex 2": {}}, "02": {"Ex 1": {}}} + + def load(_workspace, data=data, **kwargs): + return AnnotationLoadResult( + {"01": {"Ex 2": {}}} if kwargs else data, [] + ) + + with ( + patch.object(reader, "load_annotation_data", side_effect=load), + patch.object( + reader, + "_scan_annotation_directory", + return_value=( + {"01": [{"label": "Ex 2", "type": "score", "value": 3}]}, + {}, + ), + ), + patch.object( + reader, + "apply_actions_and_regenerate_grouped", + return_value=(ExitCode.SUCCESS, ""), + ) as render, + ): + self.assertEqual( + reader.run(workspace, refaire=True, annotation_dir=mode), + ExitCode.SUCCESS, + ) + render.assert_called_once() + self.assertEqual(set(render.call_args.args[1]["01"]), {"Ex 1", "Ex 2"}) + self.assertEqual(render.call_args.args[2], "01") + self.assertEqual(render.call_args.kwargs["selected_labels"], {"Ex 2"}) + self.assertEqual(render.call_args.kwargs["annotation_dir"], mode) + + def test_whole_copy_selection_replaces_all_old_actions(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + workspace = self.workspace(root, "BGnot") + (root / "BGnot/Ex 1").mkdir() + atomic_write_json(root / "refaire.json", [["Copie01", []]]) + atomic_write_json( + root / "BRnot/Copie01/bnote.json", + {"images": [{"label": label} for label in ("Ex 1", "Ex 2")]}, + ) + loaded = AnnotationLoadResult({"01": {"Ex 1": {}, "Ex 2": {}}}, []) + + def scan(directory, *args, **kwargs): + return ( + ({"01": [{"label": "Ex 1", "type": "score", "value": 1}]}, {}) + if directory.parent.name == "BGnot" + else ({"01": [{"label": "Ex 2", "type": "score", "value": 4}]}, {}) + ) + + with ( + patch.object(reader, "load_annotation_data", return_value=loaded), + patch.object(reader, "_scan_annotation_directory", side_effect=scan), + patch.object( + reader, + "apply_actions_and_regenerate_grouped", + return_value=(ExitCode.SUCCESS, ""), + ) as render, + ): + self.assertEqual(reader.run(workspace, refaire=True), ExitCode.SUCCESS) + self.assertEqual( + render.call_args.kwargs["selected_labels"], {"Ex 1", "Ex 2"} + ) + self.assertEqual( + render.call_args.args[3], + [{"label": "Ex 2", "type": "score", "value": 4}], + ) + + def test_stale_redo_selection_is_rejected(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + workspace = self.workspace(root, "Anot") + atomic_write_json( + root / "BRnot/Copie01/bnote.json", {"images": [{"label": "Ex 1"}]} + ) + with ( + patch.object( + reader, + "load_annotation_data", + return_value=AnnotationLoadResult({"01": {"Ex 2": {}}}, []), + ), + patch.object(reader, "apply_actions_and_regenerate_grouped") as render, + ): + self.assertEqual( + reader.run(workspace, refaire=True, annotation_dir="Anot"), + ExitCode.PARTIAL, + ) + render.assert_not_called() + + def test_missing_redo_return_leaves_copy_untouched(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + workspace = self.workspace(root, "BGnot") + (root / "BRnot/Copie01/Concat_annotated.pdf").unlink() + with ( + patch.object( + reader, + "load_annotation_data", + return_value=AnnotationLoadResult({"01": {"Ex 2": {}}}, []), + ), + patch.object(reader, "apply_actions_and_regenerate_grouped") as render, + ): + self.assertEqual(reader.run(workspace, refaire=True), ExitCode.PARTIAL) + render.assert_not_called() + + def test_regeneration_preserves_untouched_image_and_score_and_saves_redo(self): + for mode in ("BGnot", "Bnot", "Anot"): + with self.subTest(mode=mode), tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + workspace = self.workspace(root, mode) + output = root / mode / "Copie01" + output.mkdir() + Image.new("RGB", (40, 30), "red").save(output / "Ex 1.jpg") + before = (output / "Ex 1.jpg").read_bytes() + atomic_write_json(output / "score.json", {"Ex 1": "3.5", "Ex 2": "0"}) + answer = root / "answer.pdf" + answer.touch() + data = { + "01": { + label: { + "result": {"score": 2, "feedback": []}, + "pdf_path": answer, + "coordinates": (0, 0), + } + for label in ("Ex 1", "Ex 2") + } + } + with ( + patch.object( + reader.annotating, + "make_base_image", + return_value=(Image.new("RGB", (40, 20)), 0, 0), + ), + patch.object( + reader.annotating, + "compose_label_image", + return_value=(Image.new("RGB", (40, 20), "blue"), 0), + ), + patch.object(reader, "get_extra_pdfs_as_images", return_value=[]), + ): + status, _ = reader.apply_actions_and_regenerate_grouped( + workspace, + data, + "01", + [], + {}, + ["Ex 1", "Ex 2"], + annotation_dir=mode, + selected_labels={"Ex 2"}, + ) + self.assertEqual(status, ExitCode.SUCCESS) + self.assertEqual( + read_json(output / "score.json"), {"Ex 1": "3.5", "Ex 2": "2"} + ) + self.assertEqual((output / "Ex 1.jpg").read_bytes(), before) + self.assertTrue((output / "Ex 2.jpg").is_file()) + with Image.open(output / "Concat.jpg") as concat: + self.assertEqual(concat.size, (40, 50)) + + def test_simple_import_survives_successive_redos(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + workspace = self.workspace(root, "Anot") + output = root / "Anot/Copie01" + output.mkdir() + for label in ("Ex 1", "Ex 2"): + Image.new("RGB", (40, 30), "white").save(output / f"{label}.jpg") + Image.new("RGB", (40, 60), "red").save(output / "Concat_annotated.jpg") + atomic_write_json(output / "score.json", {"Ex 1": "1", "Ex 2": "1"}) + answer = root / "answer.pdf" + answer.touch() + data = { + "01": { + label: { + "result": {"score": 2, "feedback": []}, + "pdf_path": answer, + "coordinates": (0, 0), + } + for label in ("Ex 1", "Ex 2") + } + } + with ( + patch.object( + reader.annotating, + "make_base_image", + return_value=(Image.new("RGB", (40, 20)), 0, 0), + ), + patch.object( + reader.annotating, + "compose_label_image", + return_value=(Image.new("RGB", (40, 20), "blue"), 0), + ), + ): + for selected in ({"Ex 2"}, {"Ex 1"}): + status, _ = reader.apply_actions_and_regenerate_grouped( + workspace, + data, + "01", + [], + {}, + ["Ex 1", "Ex 2"], + annotation_dir="Anot", + selected_labels=selected, + ) + self.assertEqual(status, ExitCode.SUCCESS) + if selected == {"Ex 2"}: + with Image.open(output / "Ex 1.jpg") as untouched: + self.assertGreater(untouched.getpixel((10, 10))[0], 240) + with Image.open(output / "Concat.jpg") as concat: + self.assertEqual(concat.size, (40, 40)) + self.assertGreater(concat.getpixel((10, 30))[2], 240) + + def test_latest_group_coordinates_win_numerically(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + group = root / "Par label" / "Ex 1" + group.mkdir(parents=True) + for number, height in ((2, 100), (10, 40)): + atomic_write_json( + group / f"Group_{number}.json", [["01", 0, height, "", "Ex 1"]] + ) + Image.new("RGB", (20, height)).save(group / f"Group_{number}.jpg") + index, warnings = _coordinate_index(EvaluationWorkspace(root)) + self.assertFalse(warnings) + self.assertEqual(index[("Ex 1", "01")].height, 40) + + def test_unreadable_redo_is_rejected(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + atomic_write_json(root / "bnote.json", {"images": []}) + with ( + patch.object( + reader, "detect_checks_and_notes", return_value=([], None) + ), + self.assertRaises(ValueError), + ): + reader._scan_annotation_directory(root, required=True) + + +if __name__ == "__main__": + unittest.main()