From 8b087bb3e4ab199a2c0eadd00b3a5a042172bcee Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Miquel?= Date: Thu, 20 Aug 2026 14:35:04 +0200 Subject: [PATCH] Standardisation 3 --- Readme.org | 13 + annotating.py | 376 +++++++++---------------- annotating_by_label.py | 495 +++++++++++++++++++-------------- annotating_with_checks.py | 505 +++++++++++++++++++--------------- copienator/__init__.py | 4 + copienator/annotation_data.py | 201 ++++++++++++++ copienator/cli.py | 25 ++ copienator/filesystem.py | 43 +++ tests/test_gui_core.py | 208 +++++++++++++- 9 files changed, 1187 insertions(+), 683 deletions(-) create mode 100644 copienator/annotation_data.py create mode 100644 copienator/filesystem.py diff --git a/Readme.org b/Readme.org index 47b9227..2520499 100644 --- a/Readme.org +++ b/Readme.org @@ -171,6 +171,8 @@ scripts migrés vers cette convention sont actuellement : - =copies_tools.py=, =grouping.py= et =verify_groups.py= ; - =post-correction.py= et =resolve_manual.py= ; +- =annotating.py=, =annotating_with_checks.py= et + =annotating_by_label.py= ; - =export.py=, =import.py= et =giving_names.py=. ** Correction d'un paquet de copies @@ -355,6 +357,11 @@ OU Ajoute les annotations Gemini, et des checkboxes à cocher. Enregistrées dans le dossier =Bnot=, =--overwrite= + + Une seule copie peut être ciblée avec, par exemple, + =python annotating_with_checks.py Interro/Copies/Copie01.pdf=. + Le mode =--refaire= exige un fichier =refaire.json= et écrit dans + =BRnot=. OU 2. =python annotating_by_label.py Interro= dans =BGnot= @@ -364,6 +371,12 @@ OU _Needs_ : label_groups file (made automatically by this function), qui dit quelles questions regrouper. + +Dans ces trois modes, les métadonnées JSON sont écrites atomiquement. +Lors d'une régénération, les nouvelles sorties sont préparées dans un +dossier temporaire voisin. Une erreur de rendu conserve donc la sortie +précédente ; pour =BGnot --overwrite=, le dossier complet n'est remplacé +que si tous les groupes ont été produits. 3. =python export.py Interro= (gestion perso) Cela déplace les groupes dans le dossier configuré par =EXPORT_DIR= (par défaut =Export=). diff --git a/annotating.py b/annotating.py index c823298..63b6ddf 100644 --- a/annotating.py +++ b/annotating.py @@ -1,140 +1,52 @@ -import sys +from __future__ import annotations + +import argparse +import io import os -import json -import glob -from pathlib import Path +import re import subprocess -from PIL import Image +import tempfile +from collections.abc import Sequence +from pathlib import Path + +import matplotlib +import matplotlib.colors as mcolors +import PIL.ImageOps + +matplotlib.use("Agg") + +from matplotlib import pyplot as plt +from pdf2image import convert_from_path +from PIL import Image, ImageDraw, ImageFont + +import utils +from config import LATEX_ANOT_AFTER, LATEX_ANOT_BEFORE +from copienator import ( + EvaluationWorkspace, + ExitCode, + atomic_write_json, + evaluation_parser, + execute, + workspace_from_args, +) +from copienator.annotation_data import load_annotation_data +from copienator.filesystem import staged_directory +from utils import natural_key MARGIN_LEFT = 300 ANNOT_WIDTH = 600 -# Results is : Copie id -> label -> {pdf_path, gemini_result, coordinates} -# Coordinates are the real coordinates (hmin, hmax) of the image in the Group -# The gemini_result coordinates should be un-normalized ! -def make_dictionary(root_dir, refaire=False, refaire_list=[]): - correction_path = os.path.join(root_dir, "correction.json") - # Load correction data - try: - with open(correction_path, 'r', encoding='utf-8') as f: - corrections = json.load(f) - except FileNotFoundError: - print(f"Error: {correction_path} not found.") - sys.exit(1) - # Dictionary: keys are IDs - result_data = {} +def make_dictionary(root_dir, refaire=False, refaire_list=None): + """Compatibility wrapper used by the annotation-reading scripts.""" + workspace = EvaluationWorkspace(Path(root_dir)) + loaded = load_annotation_data( + workspace, + refaire_list=(refaire_list or []) if refaire else None, + ) + return loaded.data - # Iterate through labels and items in correction.json - for label, items in corrections.items(): - items = sum(items, []) # Flatten - for item in items: - # print(item) - student_id = item['id'] - result_obj = item['result'] - - if result_obj.get("suffix") == "_old": - continue - - # Find coordinates - coordinates = None - height,width= None, None - label_dir = Path(root_dir) / "Par label" / label - - # Search all json files in Dir/label - json_files = glob.glob(os.path.join(label_dir, "*.json")) - for jf in json_files: - try: - with open(jf, 'r', encoding='utf-8') as f: - coord_list = json.load(f) - # Format: [["id", x, y, width_r, "label"], ...] - for entry in coord_list: - if entry[0] == student_id: - coordinates = (entry[1], entry[2]) - img_path = os.path.splitext(jf)[0] + ".jpg" - with Image.open(img_path) as img: - width, height = img.size - break - except json.JSONDecodeError: - continue - if coordinates: - break - - suffix = result_obj.get("suffix", "") - if suffix == "_new": - pdf_path = Path(root_dir) / "Copies" / f"Copie{student_id}" / f"{label}_new.pdf" - else: - pdf_path = Path(root_dir) / "Copies" / f"Copie{student_id}" / f"{label}.pdf" - # Initialize dictionary structure for this ID if missing - if student_id not in result_data: - result_data[student_id] = {} - - fb = result_obj.get("feedback", []) - for i in range(len(fb)): - el = fb[i] - if height == None or width == None: - print("?? height or width is None, for ", student_id, label) - if "box_2d" in el and el["box_2d"]: - el["box_2d"][0] = (el["box_2d"][0] * height)//1000 - el["box_2d"][2] = (el["box_2d"][2] * height)//1000 - el["box_2d"][1] = (el["box_2d"][1] * width)//1000 - el["box_2d"][3] = (el["box_2d"][3] * width)//1000 - - # Populate the object - result_data[student_id][label] = { - "pdf_path": pdf_path, - "result": result_obj, - "coordinates": coordinates - } - - if refaire: - for copie_name, labels_to_redo in refaire_list: - sid = copie_name.replace("Copie", "") # Extract "01" from "Copie01" - if sid in result_data: - # Si des labels à refaire ne sont pas présent dans la correction - # On ajoute des dummies - if labels_to_redo: # Si la liste est non vide - for lbl in labels_to_redo: - pdf_path = Path(root_dir) / "Copies" / f"Copie{sid}" / f"{lbl}.pdf" - if not Path(pdf_path).exists(): - pdf_path_new = Path(root_dir) / "Copies" / f"Copie{sid}" / f"{lbl}_new.pdf" - if pdf_path_new.exists(): - pdf_path = pdf_path_new - else: - print("Debug : asked to refaire", sid, lbl, "but pdf absent") - continue - # result_data[sid][lbl] = { - # "pdf_path": pdf_path, - # "result": { - # "score": 0.0, - # "feedback": [], - # "error": "non traité" - # }, - # "coordinates": (0,0) - # } - else: # Ce student id n'a jamais été corrigé - result_data[sid] = {} - for lbl in labels_to_redo: - pdf_path = Path(root_dir) / "Copies" / f"Copie{sid}" / f"{lbl}.pdf" - if not pdf_path.exists(): - pdf_path_new = Path(root_dir) / "Copies" / f"Copie{sid}" / f"{lbl}_new.pdf" - if pdf_path_new.exists(): - pdf_path = pdf_path_new - else: - print("Debug : asked to refaire", sid, lbl, "but pdf absent") - continue - result_data[sid][lbl] = { - "pdf_path": pdf_path, - "result": { - "score": 0.0, - "feedback": [], - "error": "non traité" - }, - "coordinates": (0,0) - } - - return result_data def make_base_image(pdf_path): pages = convert_from_path(pdf_path) @@ -152,20 +64,6 @@ def make_base_image(pdf_path): current_y += page.height return (base_img, total_h, max_w) -import io -import shutil -from pdf2image import convert_from_path -from PIL import Image, ImageDraw, ImageFont -import matplotlib -matplotlib.use('Agg') # Force headless rendering -import matplotlib.pyplot as plt - -# plt.rcParams.update({ "text.usetex": True, - # "text.latex.preamble": r"\usepackage{bbold}"}) - -import re -import textwrap - def normalize_mathtext(text): """ Replaces LaTeX shortcuts not supported by Matplotlib's mathtext parser. @@ -274,16 +172,6 @@ def render_latex_text(text, width_px, bg_color=(255, 255, 255, 255), max_lines=N final_img.alpha_composite(img) return final_img -import os -import tempfile -import subprocess -import PIL.ImageOps - - -from config import LATEX_ANOT_AFTER, LATEX_ANOT_BEFORE -from string import Template - - def render_real_latex_text(text, width_px, bg_color=(255, 255, 255, 255), max_lines=None, fontsize=19): dpi = 100 width_in = width_px / dpi @@ -303,11 +191,12 @@ def render_real_latex_text(text, width_px, bg_color=(255, 255, 255, 255), max_li f.write(latex_template) # Compile to PDF - result = subprocess.run( + subprocess.run( ['pdflatex', '-interaction=nonstopmode', 'text.tex'], cwd=temp_dir, stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL + stderr=subprocess.DEVNULL, + check=False, ) if not os.path.exists(pdf_path): @@ -336,12 +225,6 @@ def render_real_latex_text(text, width_px, bg_color=(255, 255, 255, 255), max_li return final_img -import io -from PIL import Image -import matplotlib.pyplot as plt -import matplotlib.colors as mcolors -from highlight_text import ax_text - def color(score): t = max(0.0, min(1.0, float(score) / 4.0)) t = t*1.5 - 0.25 @@ -350,8 +233,6 @@ def color(score): green = 150 * t return mcolors.to_hex((red/255, green/255, 0)) -from PIL import Image, ImageDraw, ImageFont - def render_score_text(label, score, error, width_px, fontsize=30, bg_color=(255, 255, 255, 255), with_error=True, id=None): @@ -380,7 +261,7 @@ def render_score_text(label, score, error, width_px, fontsize=30, try: font_regular = ImageFont.truetype("DejaVuSans.ttf", fontsize) font_bold = ImageFont.truetype("DejaVuSans-Bold.ttf", fontsize) - except IOError: + except OSError: # Fallback for systems without specific TTFs readily available print("here") try: @@ -429,7 +310,6 @@ def compose_label_image(base_img, label, result, hmin, if base_img.width < TARGET_MIN_WIDTH: total_missing = TARGET_MIN_WIDTH - base_img.width left_pad = min(total_missing, MARGIN_LEFT) - right_pad = total_missing - left_pad new_base = Image.new("RGB", (TARGET_MIN_WIDTH, base_img.height), "white") new_base.paste(base_img, (left_pad, 0)) @@ -549,100 +429,118 @@ def compose_label_image(base_img, label, result, hmin, return final_img, header_height -from utils import natural_key -import concurrent.futures - def process_student(student_id, labels_data, root_dir, all_labels, overwrite): """Helper function to process a single student.""" # Prepare output directory: Dir/Anot_CopieID - output_dir = os.path.join(root_dir, "Anot", f"Copie{student_id}") + output_dir = Path(root_dir) / "Anot" / f"Copie{student_id}" # Check if already processed (Concat.jpg exists) - concat_path = os.path.join(output_dir, "Concat.jpg") - if os.path.exists(concat_path) and not overwrite: + concat_path = output_dir / "Concat.jpg" + if concat_path.exists() and not overwrite: print(f"Skipping Copie {student_id} (Concat.jpg exists)") - return + return "skipped" print("Processing :", student_id) + problems = False + with staged_directory(output_dir) as staging: + d_notes = dict.fromkeys(all_labels, "") + label_images = [] + sorted_labels = sorted(labels_data.items(), key=lambda item: natural_key(item[0])) - # Clean folder if re-processing - if os.path.exists(output_dir): - shutil.rmtree(output_dir) - os.makedirs(output_dir) + for label, content in sorted_labels: + pdf_full_path = content.get('pdf_path') + if not pdf_full_path or not Path(pdf_full_path).exists(): + print(f"File not found: {pdf_full_path}") + problems = True + continue + try: + base_img, _, _ = make_base_image(pdf_full_path) + except Exception as exc: # noqa: BLE001 - PDF/LaTeX backends vary + print(f"Error converting {pdf_full_path}: {exc}") + problems = True + continue - d_notes = dict.fromkeys(all_labels, "") - label_images = [] + result = content.get('result', {}) + coordinates = content.get('coordinates', (0, 0)) + d_notes[label] = str(result.get('score', 0)) + final_img, _ = compose_label_image( + base_img, + label, + result, + coordinates[0], + with_empty=True, + render_fn=render_real_latex_text, + ) + final_img.save(staging / f"{label}.jpg") + if result.get('error', "") != "empty-answer": + label_images.append(final_img) - # !! Trier par l'ordre des labels plutôt - sorted_labels = sorted(list(labels_data.items()), key=natural_key) - - for label, content in sorted_labels: - # 1. Find PDF path - copie_folder = f"Copie{student_id}" - pdf_full_path = content.get('pdf_path') - - if not pdf_full_path or not os.path.exists(pdf_full_path): - print(f"File not found: {pdf_full_path}") - continue - - # 2. Convert PDF to Image - try: - (base_img, _, _) = make_base_image(pdf_full_path) - except Exception as e: - print(f"Error converting {pdf_full_path}: {e}") - continue - - result = content.get('result', {}) - coordinates = content.get('coordinates', (0, 0)) # (hmin, hmax) - score = result.get('score', 0) - d_notes[label] = str(score) - - final_img, _ = compose_label_image(base_img, label, result, coordinates[0], - with_empty=True, - render_fn=render_real_latex_text) - # 7. Save Image - save_path = os.path.join(output_dir, f"{label}.jpg") - final_img.save(save_path) - if result.get('error', "") != "empty-answer": - label_images.append(final_img) - - # Save scores - with open(os.path.join(output_dir, "score.json"), "w") as f: - json.dump(d_notes, f, indent=4) - - # Concatenate - if label_images: - max_w = max(i.width for i in label_images) - total_h = sum(i.height for i in label_images) - canvas = Image.new('RGB', (max_w, total_h)) - cy = 0 - for img in label_images: - canvas.paste(img, (0, cy)) - cy += img.height - canvas.save(concat_path) + atomic_write_json(staging / "score.json", d_notes) + if label_images: + max_w = max(image.width for image in label_images) + total_h = sum(image.height for image in label_images) + canvas = Image.new('RGB', (max_w, total_h)) + current_y = 0 + for image in label_images: + canvas.paste(image, (0, current_y)) + current_y += image.height + canvas.save(staging / "Concat.jpg") + elif labels_data: + problems = True + return "partial" if problems else "success" def process_correction(root_dir, data, all_labels, overwrite=False): # Ne pas thread cette application # 1. Il faut protéger les appels à matplotlib # 2. tu vas perdre les erreurs - for student_id, labels in sorted(data.items()): + statuses = [ process_student(student_id, labels, root_dir, all_labels, overwrite) + for student_id, labels in sorted(data.items()) + ] + return ExitCode.PARTIAL if "partial" in statuses else ExitCode.SUCCESS + + +def build_parser() -> argparse.ArgumentParser: + parser = evaluation_parser("Generate simple annotated copies.") + parser.add_argument( + "--overwrite", + action="store_true", + help="Replace existing student output directories", + ) + return parser + + +def run(workspace: EvaluationWorkspace, *, overwrite: bool = False) -> ExitCode: + workspace.require_files("labels", "correction.json") + workspace.require_directories("Copies", "Par label") + labels = utils.read_all_labels(workspace.root) + loaded = load_annotation_data(workspace) + for warning in loaded.warnings: + print(f"Warning: {warning}") + if not loaded.data: + print("Warning: no annotation data was found.") + return ExitCode.PARTIAL + result = process_correction( + workspace.root, + loaded.data, + labels, + overwrite=overwrite, + ) + if loaded.warnings and result == ExitCode.SUCCESS: + return ExitCode.PARTIAL + return result + + +def main(argv: Sequence[str] | None = None) -> int: + parser = build_parser() + return execute( + parser, + argv, + lambda args: run(workspace_from_args(args), overwrite=args.overwrite), + ) -import argparse -import utils if __name__ == "__main__": - parser = argparse.ArgumentParser(description="Annotate copies") - parser.add_argument("root_dir", help="Directory containing the copies") - parser.add_argument("--overwrite", action="store_true", help="Reprocess even if Concat.jpg exists") - - args = parser.parse_args() - root_dir = args.root_dir - labels = utils.read_all_labels(root_dir) - results = make_dictionary(root_dir) - # Results is : Copie id -> label -> {pdf_path, gemini_result, coordinates} - # Coordinates are the real coordinates (hmin, hmax) of the image in the Group - # print(results,"\n\n\n") - process_correction(root_dir, results, labels,overwrite=args.overwrite) + raise SystemExit(main()) diff --git a/annotating_by_label.py b/annotating_by_label.py index 91647c8..216f686 100644 --- a/annotating_by_label.py +++ b/annotating_by_label.py @@ -1,256 +1,323 @@ -import sys -import os -import json -import utils -import shutil +from __future__ import annotations + import argparse import concurrent.futures +import os +from collections.abc import Sequence +from pathlib import Path +from typing import Any + from PIL import Image, ImageDraw from reportlab.pdfgen import canvas import annotating import annotating_with_checks - +import utils +from copienator import ( + CliError, + EvaluationWorkspace, + ExitCode, + atomic_write_json, + atomic_write_text, + evaluation_parser, + execute, + read_json, + workspace_from_args, +) +from copienator.annotation_data import load_annotation_data +from copienator.filesystem import staged_directory from utils import natural_key -MAX_HEIGHT_PX = 25000 # Can be increased by 10%. +MAX_HEIGHT_PX = 25000 + def render_item(item): student_id, label, content = item - pdf_path = content['pdf_path'] - if not os.path.exists(pdf_path): - print("no pdf path for ", pdf_path) + pdf_path = Path(content["pdf_path"]) + if not pdf_path.exists(): + print(f"Warning: answer PDF not found: {pdf_path}") return None - - base_img, _, _ = annotating.make_base_image(pdf_path) - cb_renderer = annotating_with_checks.CheckboxRenderer(label) - - final_img, header_h = annotating.compose_label_image( - base_img, label, content['result'], content['coordinates'][0], - draw_callback=cb_renderer.callback, + base_image, _, _ = annotating.make_base_image(pdf_path) + checkbox_renderer = annotating_with_checks.CheckboxRenderer(label) + final_image, header_height = annotating.compose_label_image( + base_image, + label, + content["result"], + content["coordinates"][0], + draw_callback=checkbox_renderer.callback, more_right=True, - with_id=student_id + with_id=student_id, ) - if final_img is None: + if final_image is None: return None + return ( + student_id, + label, + final_image, + header_height, + checkbox_renderer.checkboxes, + ) - return (student_id, label, final_img, header_h, cb_renderer.checkboxes) -def save_batch(batch, prefix, group_id, root_dir, overwrite): - output_dir = os.path.join(root_dir, "BGnot", f"{prefix} G{group_id}") - - if os.path.exists(output_dir): - if not overwrite: - print(f"Skipping {output_dir}: Output already exists.") - return - shutil.rmtree(output_dir) - - print(f"Generating Group PDF: {prefix} G{group_id} ({len(batch)} elements)") - os.makedirs(output_dir) - - max_w = max(item[2].width for item in batch) - total_h = sum(item[2].height for item in batch) - concat_img = Image.new("RGB", (max_w, total_h), "white") - draw = ImageDraw.Draw(concat_img) - - final_json_map = [] - bnote_entries = [] +def save_batch(batch, prefix, group_id, output_root: Path) -> None: + output_dir = output_root / f"{prefix} G{group_id}" + print(f"Generating group PDF: {prefix} G{group_id} ({len(batch)} elements)") + max_width = max(item[2].width for item in batch) + total_height = sum(item[2].height for item in batch) + concatenated = Image.new("RGB", (max_width, total_height), "white") + draw = ImageDraw.Draw(concatenated) + checkbox_map: list[dict[str, Any]] = [] + bnote_entries: list[dict[str, Any]] = [] current_y = 0 - last_sid = None + previous_student = None - for sid, label, img, header_h, boxes in batch: - concat_img.paste(img, (0, current_y)) + for student_id, label, image, header_height, checkboxes in batch: + concatenated.paste(image, (0, current_y)) + if student_id != previous_student: + draw.rectangle([0, current_y, max_width, current_y + 4], fill="purple") + previous_student = student_id + bnote_entries.append( + { + "id": student_id, + "label": label, + "header_height": header_height, + "hmin": current_y, + "hmax": current_y + image.height, + } + ) + for item in checkboxes: + box = item.get("final_box") or item.get("rel_box") + item["global_box"] = [ + box[0], + box[1] + current_y, + box[2], + box[3] + current_y, + ] + item["student_id"] = student_id + checkbox_map.append(item) + current_y += image.height - if sid != last_sid: - draw.rectangle([0, current_y, max_w, current_y + 4], fill="purple") - last_sid = sid + with staged_directory(output_dir) as staging: + atomic_write_json( + staging / "bnote.json", + {"width": max_width, "height": total_height, "images": bnote_entries}, + ) + atomic_write_json(staging / "checkboxes.json", checkbox_map) + reference = staging / "Reference.jpg" + concatenated.save(reference, quality=90) + pdf_path = staging / "Concat.pdf" + pdf_canvas = canvas.Canvas(str(pdf_path), pagesize=(max_width, total_height)) + pdf_canvas.drawImage( + str(reference), + 0, + 0, + width=max_width, + height=total_height, + ) + pdf_canvas.save() - bnote_entries.append({ - "id": sid, - "label": label, - "header_height": header_h, - "hmin": current_y, - "hmax": current_y + img.height - }) - for item in boxes: - b = item.get('final_box') or item.get('rel_box') - item['global_box'] = [b[0], b[1] + current_y, b[2], b[3] + current_y] - item['student_id'] = sid # Required to map checkbox to the correct student - final_json_map.append(item) +def _initial_label_groups(labels: list[str]) -> str: + groups: dict[str, list[str]] = {} + for label in labels: + key = label.split(" : ")[0] if " : " in label else label + groups.setdefault(key, []).append(label) + return "".join(",".join(items) + "\n" for items in groups.values()) - current_y += img.height - with open(os.path.join(output_dir, "bnote.json"), "w") as f: - json.dump({"width": max_w, "height": total_h, "images": bnote_entries}, f, indent=2) - - with open(os.path.join(output_dir, "checkboxes.json"), "w") as f: - json.dump(final_json_map, f, indent=2) - - temp_img_path = os.path.join(output_dir, "Reference.jpg") - concat_img.save(temp_img_path, quality=90) - - pdf_path = os.path.join(output_dir, "Concat.pdf") - w, h = concat_img.size - c = canvas.Canvas(pdf_path, pagesize=(w, h)) - c.drawImage(temp_img_path, 0, 0, width=w, height=h) - c.save() - -def main(): - parser = argparse.ArgumentParser(description="Generate annotated PDFs grouped by labels.") - parser.add_argument("input_path", help="Directory containing Bnot structure") - parser.add_argument("--overwrite", action="store_true", help="Overwrite existing output files") - args = parser.parse_args() - - root_dir = args.input_path - validated_labels = utils.read_all_labels(root_dir) - results = annotating.make_dictionary(root_dir) - label_groups = os.path.join(root_dir, "label_groups") - all_labels = os.path.join(root_dir, "labels") - - if not os.path.exists(label_groups): - print(f"Warning: Labels file '{label_groups}' not found, making it out of '{all_labels}'") - if not os.path.exists(all_labels): - print(f"Error: {all_labels} not found.") - sys.exit(1) - - lines = validated_labels - - groups = {} - for line in lines: - # Key is the part before the colon, or the whole line if no colon - key = line.split(' : ')[0] if ' : ' in line else line - groups.setdefault(key, []).append(line) - - with open(label_groups, 'w') as f: - for items in groups.values(): - f.write(",".join(items) + "\n") +def _load_label_groups(workspace: EvaluationWorkspace, labels: list[str]) -> list[list[str]]: + label_groups = workspace.root / "label_groups" + if not label_groups.exists(): + atomic_write_text(label_groups, _initial_label_groups(labels)) + print(f"Created {label_groups}; review the groups before continuing.") utils.edit_file_and_enter(label_groups) + known_labels = set(labels) + groups: list[list[str]] = [] + unknown: set[str] = set() + for line in label_groups.read_text(encoding="utf-8").splitlines(): + group = [label.strip() for label in line.split(",") if label.strip()] + unknown.update(label for label in group if label not in known_labels) + if group: + groups.append(group) + if unknown: + raise CliError( + "label_groups contains unknown labels: " + ", ".join(sorted(unknown)) + ) + return groups - with open(label_groups, "r") as f: - lines = [line.strip() for line in f if line.strip()] - bgnot_dir = os.path.join(root_dir, "BGnot") - if args.overwrite and os.path.exists(bgnot_dir): - shutil.rmtree(bgnot_dir) - os.makedirs(bgnot_dir, exist_ok=True) - - used_prefixes = set() - - previous_prefix = None - for line in lines: - labels = [l.strip() for l in line.split(',') if l.strip()] - safe_labels = [l.replace(":", "").strip() for l in line.split(',') if l.strip()] - if not labels: - continue - - base_prefix = os.path.commonprefix(safe_labels).strip() - - if base_prefix and previous_prefix is not None: - if natural_key(base_prefix) < natural_key(previous_prefix): - base_prefix_maybe = f"{safe_labels[0]}+" - if natural_key(base_prefix_maybe) > natural_key(previous_prefix): - base_prefix = base_prefix_maybe - - if not base_prefix: - base_prefix = "Group" - - unique_prefix = base_prefix - if unique_prefix[-1] == "i": - unique_prefix = unique_prefix[:-1] +def _unique_prefixes(groups: list[list[str]]) -> list[tuple[str, list[str]]]: + used: set[str] = set() + result: list[tuple[str, list[str]]] = [] + previous: str | None = None + for labels in groups: + safe_labels = [label.replace(":", "").strip() for label in labels] + base = os.path.commonprefix(safe_labels).strip() or "Group" + if base and previous is not None and natural_key(base) < natural_key(previous): + candidate = f"{safe_labels[0]}+" + if natural_key(candidate) > natural_key(previous): + base = candidate + prefix = base.removesuffix("i") counter = 2 - while unique_prefix in used_prefixes: - unique_prefix = f"{base_prefix}-{counter}" + while prefix in used: + prefix = f"{base}-{counter}" counter += 1 - if counter == 2 and previous_prefix and previous_prefix in unique_prefix: - unique_prefix = f"{previous_prefix}-{counter}" + if counter == 2 and previous and previous in prefix: + prefix = f"{previous}-{counter}" elif counter == 2: - previous_prefix = unique_prefix - - used_prefixes.add(unique_prefix) - - existing_items = set() - max_existing_group = 0 + previous = prefix + used.add(prefix) + result.append((prefix, labels)) + return result - if not args.overwrite and os.path.exists(bgnot_dir): - for d in os.listdir(bgnot_dir): - if d.startswith(f"{unique_prefix} G"): - try: - g_id = int(d.split(' G')[-1]) - max_existing_group = max(max_existing_group, g_id) - except ValueError: - pass - - bnote_path = os.path.join(bgnot_dir, d, "bnote.json") - if os.path.exists(bnote_path): - with open(bnote_path, "r") as bf: - bdata = json.load(bf) - for img in bdata.get("images", []): - existing_items.add((img["id"], img["label"])) - - items_to_render = [] - for sid, lbls in results.items(): - for lbl in labels: - if lbl in lbls: - # Only add if it hasn't been generated yet - if (sid, lbl) not in existing_items: - items_to_render.append((sid, lbl, lbls[lbl])) - if not items_to_render: +def _existing_group_state( + output_root: Path, + prefix: str, +) -> tuple[set[tuple[str, str]], int]: + existing_items: set[tuple[str, str]] = set() + maximum_group = 0 + if not output_root.is_dir(): + return existing_items, maximum_group + for directory in output_root.iterdir(): + if not directory.is_dir() or not directory.name.startswith(f"{prefix} G"): continue - - # Sort structurally: by student id and label - items_to_render.sort(key=lambda x: (natural_key(x[0]), natural_key(x[1]))) - # Render images in parallel using the pre-existing lock & render function - with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor: - rendered = list(executor.map(render_item, items_to_render)) - - rendered = [r for r in rendered if r is not None] - if not rendered: + try: + maximum_group = max(maximum_group, int(directory.name.split(" G")[-1])) + except ValueError: continue + metadata = directory / "bnote.json" + if not metadata.exists(): + continue + loaded = read_json(metadata) + if not isinstance(loaded, dict): + raise TypeError(f"Expected a JSON object in {metadata}") + for image in loaded.get("images", []): + existing_items.add((str(image["id"]), str(image["label"]))) + return existing_items, maximum_group - # Split into constrained height batches + +def split_batches(rendered): + def split(maximum_height: float): batches = [] - current_batch = [] - current_h = 0 - for r in rendered: - sid = r[0] - img_h = r[2].height - # Split if we exceed max height AND we are on a new student - if current_batch and current_h + img_h > MAX_HEIGHT_PX and sid != last_sid: - batches.append(current_batch) - current_batch = [] - current_h = 0 - current_batch.append(r) - current_h += img_h - last_sid = sid - if current_batch: - batches.append(current_batch) + current = [] + current_height = 0 + previous_student = None + for item in rendered: + student_id = item[0] + image_height = item[2].height + if ( + current + and current_height + image_height > maximum_height + and student_id != previous_student + ): + batches.append(current) + current = [] + current_height = 0 + current.append(item) + current_height += image_height + previous_student = student_id + if current: + batches.append(current) + return batches - batches2 = [] - current_batch2 = [] - current_h2 = 0 - last_sid2 = None - for r in rendered: - sid = r[0] - img_h = r[2].height - # Split if we exceed max height AND we are on a new student - if current_batch2 and current_h2 + img_h > 1.1 *MAX_HEIGHT_PX \ - and sid != last_sid2: - batches2.append(current_batch2) - current_batch2 = [] - current_h2 = 0 - current_batch2.append(r) - current_h2 += img_h - last_sid2 = sid - if current_batch2: - batches2.append(current_batch2) + strict = split(MAX_HEIGHT_PX) + relaxed = split(1.1 * MAX_HEIGHT_PX) + return relaxed if len(relaxed) < len(strict) else strict - if len(batches2) < len(batches): - batches = batches2 - for i, batch in enumerate(batches, 1): - save_batch(batch, unique_prefix, max_existing_group + i, root_dir, args.overwrite) +def _generate_groups( + output_root: Path, + data, + groups: list[list[str]], + *, + resume: bool, +) -> tuple[int, bool]: + generated = 0 + problems = False + for prefix, labels in _unique_prefixes(groups): + existing_items, maximum_group = ( + _existing_group_state(output_root, prefix) if resume else (set(), 0) + ) + items = [ + (student_id, label, student_labels[label]) + for student_id, student_labels in data.items() + for label in labels + if label in student_labels and (student_id, label) not in existing_items + ] + if not items: + continue + items.sort(key=lambda item: (natural_key(item[0]), natural_key(item[1]))) + with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor: + rendered = list(executor.map(render_item, items)) + if any(item is None for item in rendered): + problems = True + successful = [item for item in rendered if item is not None] + for index, batch in enumerate(split_batches(successful), start=1): + save_batch(batch, prefix, maximum_group + index, output_root) + generated += 1 + return generated, problems + + +def run(workspace: EvaluationWorkspace, *, overwrite: 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) + 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: + class IncompleteGroupedOutput(Exception): + pass + + try: + with staged_directory(output_root) as staging: + generated, problems = _generate_groups( + staging, + loaded.data, + groups, + resume=False, + ) + if generated == 0 or problems or loaded.warnings: + raise IncompleteGroupedOutput + except IncompleteGroupedOutput: + print("Warning: grouped overwrite was incomplete; previous BGnot was preserved.") + return ExitCode.PARTIAL + else: + output_root.mkdir(parents=True, exist_ok=True) + generated, problems = _generate_groups( + output_root, + loaded.data, + groups, + resume=True, + ) + if problems: + return ExitCode.PARTIAL + if generated == 0: + print("No new grouped annotations were required.") + return ExitCode.PARTIAL if loaded.warnings else ExitCode.SUCCESS + + +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") + return parser + + +def main(argv: Sequence[str] | None = None) -> int: + parser = build_parser() + return execute( + parser, + argv, + lambda args: run(workspace_from_args(args), overwrite=args.overwrite), + ) + if __name__ == "__main__": - main() + raise SystemExit(main()) diff --git a/annotating_with_checks.py b/annotating_with_checks.py index cbb58f3..e3b6019 100644 --- a/annotating_with_checks.py +++ b/annotating_with_checks.py @@ -1,283 +1,330 @@ -import sys -import os -import json -import shutil +from __future__ import annotations + +import argparse import concurrent.futures -import threading -import img2pdf +import re +from collections.abc import Sequence +from pathlib import Path +from typing import Any + +import matplotlib + +matplotlib.use("Agg") + +from PIL import Image, ImageFont from reportlab.pdfgen import canvas -# Fix for Matplotlib in threads: Set backend to non-interactive 'Agg' -import matplotlib -matplotlib.use('Agg') - -from PIL import Image, ImageDraw, ImageFont import annotating import utils -from annotating import MARGIN_LEFT, ANNOT_WIDTH +from copienator import ( + CliError, + EvaluationWorkspace, + ExitCode, + atomic_write_json, + execute, + read_json, + target_parser, + workspace_from_target, +) +from copienator.annotation_data import load_annotation_data +from copienator.filesystem import staged_directory +from utils import natural_key -# Global lock for Matplotlib/Latex rendering to prevent race conditions -LATEX_LOCK = threading.Lock() -DPI = 100 BOX_SIZE = 30 SCORE_BOX_SIZE = 40 -SCORES = [x * 0.5 for x in range(10)] # 0.0 to 4.5 +SCORES = [value * 0.5 for value in range(10)] +EXPECTED_OUTPUTS = ("bnote.json", "checkboxes.json", "Reference.jpg", "Concat.pdf") try: CHECKBOX_FONT = ImageFont.truetype("DejaVuSans.ttf", 20) -except IOError: +except OSError: try: CHECKBOX_FONT = ImageFont.truetype("arial.ttf", 20) - except IOError: + except OSError: CHECKBOX_FONT = ImageFont.load_default() + 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] -def safe_render_latex(*args, **kwargs): - """Thread-safe wrapper for latex rendering.""" - # with LATEX_LOCK: - # return annotating.render_latex_text(*args, **kwargs) - return annotating.render_real_latex_text(*args, **kwargs) class CheckboxRenderer: def __init__(self, label_name): self.label = label_name - self.checkboxes = [] # List of {type, box, etc.} + self.checkboxes = [] def callback(self, kind, draw, pos, meta): - """ - Called by compose_label_image during rendering. - pos contains {x, y, w, h} or {box}. - meta contains {data, index, etc.} - """ if kind == "header_item": - # meta['data'] is either result object (for score) or feedback object if meta.get("type") == "score": - # Draw score boxes - start_x = pos['w'] + 20 - for val in SCORES: - box = draw_checkbox(draw, start_x, pos['y'] + 25, - SCORE_BOX_SIZE, str(val)) - self.checkboxes.append({ - "type": "score", "label": self.label, "value": val, - "rel_box": box # Will be adjusted for global Y later - }) + start_x = pos["w"] + 20 + for value in SCORES: + box = draw_checkbox( + draw, + start_x, + pos["y"] + 25, + SCORE_BOX_SIZE, + str(value), + ) + self.checkboxes.append( + { + "type": "score", + "label": self.label, + "value": value, + "rel_box": box, + } + ) start_x += SCORE_BOX_SIZE + 45 - start_x += SCORE_BOX_SIZE + 60 - box = draw_checkbox(draw, start_x, pos['y'] + 25, SCORE_BOX_SIZE, "clr") - self.checkboxes.append({ - "type": "clear_all", "label": self.label, - "rel_box": box - }) - + box = draw_checkbox( + draw, + start_x, + pos["y"] + 25, + SCORE_BOX_SIZE, + "clr", + ) + self.checkboxes.append( + {"type": "clear_all", "label": self.label, "rel_box": box} + ) elif meta.get("type") == "global_fb": - # Draw delete box for global feedback - bx = pos['w'] - BOX_SIZE - 5 - by = pos['y'] + 5 - box = draw_checkbox(draw, bx, by, BOX_SIZE) - self.checkboxes.append({ - "type": "del_global", "label": self.label, "index": meta["index"], - "rel_box": box, "text_preview": meta["data"]["text"][:20] - }) - + box = draw_checkbox( + draw, + pos["w"] - BOX_SIZE - 5, + pos["y"] + 5, + BOX_SIZE, + ) + self.checkboxes.append( + { + "type": "del_global", + "label": self.label, + "index": meta["index"], + "rel_box": box, + "text_preview": meta["data"]["text"][:20], + } + ) elif kind == "local_rect": - # Delete rect checkbox - b = pos['box'] # [xmin, ymin, xmax, ymax] - box = draw_checkbox(draw, b[2] - BOX_SIZE, b[1], BOX_SIZE) - self.checkboxes.append({ - "type": "del_local_rect", "label": self.label, "index": meta["index"], - "final_box": box, "text_preview": meta["data"]["text"][:20] - }) - + rectangle = pos["box"] + box = draw_checkbox( + draw, + rectangle[2] - BOX_SIZE, + rectangle[1], + BOX_SIZE, + ) + self.checkboxes.append( + { + "type": "del_local_rect", + "label": self.label, + "index": meta["index"], + "final_box": box, + "text_preview": meta["data"]["text"][:20], + } + ) elif kind == "local_text": - # Delete whole local feedback checkbox - bx = pos['x'] + pos['w'] - BOX_SIZE - by = pos['y'] - box = draw_checkbox(draw, bx, by, BOX_SIZE) - self.checkboxes.append({ - "type": "del_local", "label": self.label, "index": meta["index"], - "final_box": box, "text_preview": meta["data"]["text"][:20] - }) + box = draw_checkbox( + draw, + pos["x"] + pos["w"] - BOX_SIZE, + pos["y"], + BOX_SIZE, + ) + self.checkboxes.append( + { + "type": "del_local", + "label": self.label, + "index": meta["index"], + "final_box": box, + "text_preview": meta["data"]["text"][:20], + } + ) -from utils import natural_key -def process_student(args): - """Thread worker: Processes one student.""" - root_dir, student_id, labels, overwrite, sub_folder = args +def _output_complete(output_dir: Path) -> bool: + return all((output_dir / name).is_file() for name in EXPECTED_OUTPUTS) - output_dir = os.path.join(root_dir, sub_folder, f"Copie{student_id}") - if os.path.exists(output_dir): - if not overwrite: - print(f"Skipping {student_id}: Output already exists.") - return - shutil.rmtree(output_dir) +def _render_student( + workspace: EvaluationWorkspace, + student_id: str, + labels: dict[str, dict[str, Any]], + *, + overwrite: bool, + output_mode: str, +) -> str: + output_dir = 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" - print(f"Generating Checkable PDF for: {student_id}") - os.makedirs(output_dir) + print(f"Generating checkable PDF for: {student_id}") + label_images: list[Image.Image] = [] + checkbox_groups: list[list[dict[str, Any]]] = [] + bnote_entries: list[dict[str, Any]] = [] + problems = False - label_images = [] - # ... (rest of the function remains exactly the same) - all_checkboxes = [] - bnote_entries = [] # For bnote.json - - sorted_labels = sorted(labels.items(), key=lambda x: natural_key(x[0])) - - for label, content in sorted_labels: - pdf_path = content['pdf_path'] - if not os.path.exists(pdf_path): continue - - base_img, _, _ = annotating.make_base_image(pdf_path) - - # Initialize the hook - cb_renderer = CheckboxRenderer(label) - - # Render using the shared engine - final_img, header_h = annotating.compose_label_image( - base_img, label, content['result'], content['coordinates'][0], - draw_callback=cb_renderer.callback + for label, content in sorted(labels.items(), key=lambda item: natural_key(item[0])): + pdf_path = Path(content["pdf_path"]) + if not pdf_path.exists(): + print(f"Warning: answer PDF not found: {pdf_path}") + problems = True + continue + base_image, _, _ = annotating.make_base_image(pdf_path) + checkbox_renderer = CheckboxRenderer(label) + final_image, header_height = annotating.compose_label_image( + base_image, + label, + content["result"], + content["coordinates"][0], + draw_callback=checkbox_renderer.callback, + ) + if final_image is None: + continue + label_images.append(final_image) + checkbox_groups.append(checkbox_renderer.checkboxes) + bnote_entries.append( + { + "id": student_id, + "label": label, + "header_height": header_height, + "img_h": final_image.height, + } ) - if final_img == None: - continue - label_images.append(final_img) - all_checkboxes.append(cb_renderer.checkboxes) - bnote_entries.append({ - "id": student_id, - "label": label, - "header_height": header_h, - # hmin/hmax will be filled during concatenation - "img_h": final_img.height - }) + if not label_images: + print(f"Warning: no annotations could be rendered for Copie{student_id}") + return "partial" - if not label_images: return - - # Concatenate - max_w = max(i.width for i in label_images) - total_h = sum(i.height for i in label_images) - concat_img = Image.new("RGB", (max_w, total_h), "white") - - final_json_map = [] + max_width = max(image.width for image in label_images) + total_height = sum(image.height for image in label_images) + concatenated = Image.new("RGB", (max_width, total_height), "white") + checkbox_map: list[dict[str, Any]] = [] current_y = 0 + for index, (image, checkboxes) in enumerate( + zip(label_images, checkbox_groups, strict=True) + ): + concatenated.paste(image, (0, current_y)) + bnote_entries[index]["hmin"] = current_y + bnote_entries[index]["hmax"] = current_y + image.height + del bnote_entries[index]["img_h"] + for item in checkboxes: + box = item.get("final_box") or item.get("rel_box") + item["global_box"] = [ + box[0], + box[1] + current_y, + box[2], + box[3] + current_y, + ] + checkbox_map.append(item) + current_y += image.height - for idx, (img, boxes) in enumerate(zip(label_images, all_checkboxes)): - concat_img.paste(img, (0, current_y)) + with staged_directory(output_dir) as staging: + atomic_write_json( + staging / "bnote.json", + {"width": max_width, "height": total_height, "images": bnote_entries}, + ) + atomic_write_json(staging / "checkboxes.json", checkbox_map) + reference = staging / "Reference.jpg" + concatenated.save(reference, quality=90) + pdf_path = staging / "Concat.pdf" + pdf_canvas = canvas.Canvas(str(pdf_path), pagesize=(max_width, total_height)) + pdf_canvas.drawImage( + str(reference), + 0, + 0, + width=max_width, + height=total_height, + ) + pdf_canvas.save() + return "partial" if problems else "success" - bnote_entries[idx]["hmin"] = current_y - bnote_entries[idx]["hmax"] = current_y + img.height - del bnote_entries[idx]["img_h"] # Clean up temp data - # Adjust coordinates for concatenated image - for item in boxes: - # item might have 'rel_box' (header) or 'final_box' (local) - # Both were relative to the label image. We just add current_y. - b = item.get('final_box') or item.get('rel_box') - item['global_box'] = [b[0], b[1] + current_y, b[2], b[3] + current_y] - final_json_map.append(item) +def _copy_id_from_target(workspace: EvaluationWorkspace, target: Path) -> str | None: + if target == workspace.root: + return None + match = re.search(r"Copie(\d+)", target.name) + if match is None: + raise CliError(f"Could not extract a copy id from target: {target}") + return match.group(1) - current_y += img.height - bnote_data = { - "width": max_w, - "height": total_h, - "images": bnote_entries - } - with open(os.path.join(output_dir, "bnote.json"), "w") as f: - json.dump(bnote_data, f, indent=2) +def _load_refaire(workspace: EvaluationWorkspace): + workspace.require_files("refaire.json") + loaded = read_json(workspace.refaire_file) + if not isinstance(loaded, list): + raise CliError("refaire.json must contain a JSON array") + return loaded - with open(os.path.join(output_dir, "checkboxes.json"), "w") as f: - json.dump(final_json_map, f, indent=2) - temp_img_path = os.path.join(output_dir, "Reference.jpg") # Can't use png here - concat_img.save(temp_img_path, quality=90) +def run( + workspace: EvaluationWorkspace, + target: Path, + *, + overwrite: bool = False, + refaire: bool = False, +) -> ExitCode: + workspace.require_files("labels", "correction.json") + workspace.require_directories("Copies", "Par label") + utils.read_all_labels(workspace.root) + copy_id = _copy_id_from_target(workspace, target) + refaire_list = _load_refaire(workspace) if refaire else None + loaded = load_annotation_data( + workspace, + refaire_list=refaire_list, + copy_id=None if refaire else copy_id, + ) + for warning in loaded.warnings: + print(f"Warning: {warning}") + if not loaded.data: + print("Warning: no annotation data was found.") + return ExitCode.PARTIAL - pdf_path = os.path.join(output_dir, "Concat.pdf") - w, h = concat_img.size - c = canvas.Canvas(pdf_path, pagesize=(w, h)) - c.drawImage(temp_img_path, 0, 0, width=w, height=h) - c.save() + output_mode = "refaire" if refaire else "checks" + tasks = sorted(loaded.data.items(), key=lambda item: natural_key(item[0])) + statuses: list[str] = [] + with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor: + futures = [ + executor.submit( + _render_student, + workspace, + student_id, + labels, + overwrite=overwrite, + output_mode=output_mode, + ) + for student_id, labels in tasks + ] + for future in futures: + statuses.append(future.result()) + if loaded.warnings or "partial" in statuses: + return ExitCode.PARTIAL + return ExitCode.SUCCESS + + +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( + "--refaire", + action="store_true", + help="Process only entries from refaire.json", + ) + return parser + + +def main(argv: Sequence[str] | None = None) -> int: + parser = build_parser() + + def handler(args: argparse.Namespace) -> ExitCode: + workspace, target = workspace_from_target(args) + return run( + workspace, + target, + overwrite=args.overwrite, + refaire=args.refaire, + ) + + return execute(parser, argv, handler) -import argparse # Added if __name__ == "__main__": - parser = argparse.ArgumentParser(description="Generate annotated PDFs.") - parser.add_argument("input_path", help="Directory or specific file path") - parser.add_argument("--overwrite", action="store_true", help="Overwrite existing output files") - parser.add_argument("--refaire", action="store_true", help="Process only copies/labels defined in refaire.json") # ADD THIS LINE - - - args = parser.parse_args() - - input_path = args.input_path - overwrite = args.overwrite # Capture flag - target_id = None - - # Detect if input is a specific file - if os.path.isfile(input_path): - root_dir = os.path.dirname(input_path) or "." - # Extract ID from filename (e.g., Copie40.pdf -> 40) - match = re.search(r'Copie(\d+)', os.path.basename(input_path)) - if match: - target_id = match.group(1) - else: - print("Error: Could not extract student ID from filename.") - sys.exit(1) - else: - root_dir = input_path - - if os.path.exists(os.path.join(root_dir, "labels")): - utils.read_all_labels(root_dir) - - if not args.refaire: - results = annotating.make_dictionary(root_dir) - - if args.refaire: - refaire_path = os.path.join(root_dir, "refaire.json") - if os.path.exists(refaire_path): - with open(refaire_path, "r", encoding="utf-8") as f: - refaire_list = json.load(f) - results = annotating.make_dictionary(root_dir, - refaire=True,refaire_list=refaire_list) - - filtered_results = {} - for copie_name, labels_to_redo in refaire_list: - sid = copie_name.replace("Copie", "") # Extract "01" from "Copie01" - if sid in results: - if not labels_to_redo: - # Empty list: keep all labels for this Copie - filtered_results[sid] = results[sid] - else: - # Keep only the requested labels - filtered_results[sid] = { - lbl: data for lbl, data in results[sid].items() - if lbl in labels_to_redo - } - results = filtered_results - else: - print(f"Warning: --refaire flag used, but {refaire_path} not found.") - elif target_id: - if target_id in results: - results = {target_id: results[target_id]} - else: - print(f"Student ID {target_id} not found in directory scan.") - results = {} - - sub_folder = "BRnot" if args.refaire else "Bnot" - - tasks = sorted([(root_dir, sid, lbls, overwrite, sub_folder) - for sid, lbls in results.items()]) - - with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor: - results = executor.map(process_student, tasks) - try: - for _ in results: - pass - except Exception: - import traceback - traceback.print_exc() + raise SystemExit(main()) diff --git a/copienator/__init__.py b/copienator/__init__.py index dfff4d9..9701ab9 100644 --- a/copienator/__init__.py +++ b/copienator/__init__.py @@ -7,7 +7,9 @@ from .cli import ( evaluation_workspace, execute, standard_parser, + target_parser, workspace_from_args, + workspace_from_target, ) from .json_io import ( JsonLockTimeout, @@ -37,5 +39,7 @@ __all__ = [ "execute", "read_json", "standard_parser", + "target_parser", "workspace_from_args", + "workspace_from_target", ] diff --git a/copienator/annotation_data.py b/copienator/annotation_data.py new file mode 100644 index 0000000..404e366 --- /dev/null +++ b/copienator/annotation_data.py @@ -0,0 +1,201 @@ +from __future__ import annotations + +import copy +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from PIL import Image + +from .json_io import read_json +from .workspace import EvaluationWorkspace + +AnnotationData = dict[str, dict[str, dict[str, Any]]] +RefaireList = list[list[Any]] + + +@dataclass(frozen=True, slots=True) +class GroupCoordinates: + minimum: int + maximum: int + width: int + height: int + + +@dataclass(slots=True) +class AnnotationLoadResult: + data: AnnotationData + warnings: list[str] + + +def _coordinate_index( + workspace: EvaluationWorkspace, +) -> tuple[dict[tuple[str, str], GroupCoordinates], list[str]]: + index: dict[tuple[str, str], GroupCoordinates] = {} + warnings: list[str] = [] + 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")): + image_path = metadata_path.with_suffix(".jpg") + try: + entries = read_json(metadata_path) + if not isinstance(entries, list): + raise TypeError("expected a JSON array") + with Image.open(image_path) as image: + width, height = image.size + for entry in entries: + copy_id = str(entry[0]) + minimum = int(entry[1]) + maximum = int(entry[2]) + label = str(entry[4]) + index.setdefault( + (label, copy_id), + GroupCoordinates(minimum, maximum, width, height), + ) + except (IndexError, OSError, TypeError, ValueError) as exc: + warnings.append(f"Could not read group metadata {metadata_path}: {exc}") + return index, warnings + + +def _scaled_result(result: dict[str, Any], coordinates: GroupCoordinates | None): + scaled = copy.deepcopy(result) + if coordinates is None: + return scaled + for feedback in scaled.get("feedback", []): + box = feedback.get("box_2d") + if not box or len(box) != 4: + continue + box[0] = int(box[0] * coordinates.height) // 1000 + box[2] = int(box[2] * coordinates.height) // 1000 + box[1] = int(box[1] * coordinates.width) // 1000 + box[3] = int(box[3] * coordinates.width) // 1000 + return scaled + + +def _answer_pdf( + workspace: EvaluationWorkspace, + copy_id: str, + label: str, + suffix: str = "", +) -> Path: + copy_dir = workspace.copies_dir / f"Copie{copy_id}" + preferred = copy_dir / f"{label}{suffix}.pdf" + if preferred.exists(): + return preferred + for candidate in ( + copy_dir / f"{label}.pdf", + copy_dir / f"{label}_new.pdf", + ): + if candidate.exists(): + return candidate + return preferred + + +def _dummy_entry( + workspace: EvaluationWorkspace, + copy_id: str, + label: str, +) -> dict[str, Any]: + return { + "pdf_path": _answer_pdf(workspace, copy_id, label), + "result": { + "score": 0.0, + "feedback": [], + "error": "non traité", + }, + "coordinates": (0, 0), + "issues": ["No correction result was available for this answer."], + } + + +def _apply_refaire_filter( + workspace: EvaluationWorkspace, + data: AnnotationData, + refaire_list: RefaireList, + warnings: list[str], +) -> AnnotationData: + filtered: AnnotationData = {} + for raw_entry in refaire_list: + if not isinstance(raw_entry, list) or len(raw_entry) != 2: + warnings.append(f"Ignoring malformed refaire entry: {raw_entry!r}") + continue + copy_name, requested_labels = raw_entry + copy_id = str(copy_name).removeprefix("Copie") + available = data.get(copy_id, {}) + if not requested_labels: + filtered[copy_id] = dict(available) + continue + selected: dict[str, dict[str, Any]] = {} + for raw_label in requested_labels: + label = str(raw_label) + selected[label] = ( + available[label] + if label in available + else _dummy_entry(workspace, copy_id, label) + ) + filtered[copy_id] = selected + return filtered + + +def load_annotation_data( + workspace: EvaluationWorkspace, + *, + refaire_list: RefaireList | None = None, + copy_id: str | None = None, +) -> AnnotationLoadResult: + workspace.require_files("correction.json") + corrections = read_json(workspace.correction_file) + if not isinstance(corrections, dict): + raise TypeError("correction.json must contain a JSON object") + + coordinate_index, warnings = _coordinate_index(workspace) + data: AnnotationData = {} + for label, raw_batches in corrections.items(): + if not isinstance(raw_batches, list): + warnings.append(f"Ignoring malformed correction batches for {label!r}") + continue + for raw_batch in raw_batches: + if not isinstance(raw_batch, list): + 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): + 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}") + continue + result = item["result"] + suffix = str(result.get("suffix", "")) + if suffix == "_old": + continue + coordinates = coordinate_index.get((str(label), student_id)) + issues: list[str] = [] + if coordinates is None: + issues.append("Group coordinates were not found.") + pdf_path = _answer_pdf(workspace, student_id, str(label), suffix) + if not pdf_path.exists(): + issues.append(f"Answer PDF not found: {pdf_path}") + data.setdefault(student_id, {})[str(label)] = { + "pdf_path": pdf_path, + "result": _scaled_result(result, coordinates), + "coordinates": ( + (coordinates.minimum, coordinates.maximum) + if coordinates is not None + else (0, 0) + ), + "issues": issues, + } + warnings.extend( + f"Copie{student_id} {label}: {issue}" for issue in issues + ) + + if refaire_list is not None: + data = _apply_refaire_filter(workspace, data, refaire_list, warnings) + if copy_id is not None: + data = {copy_id: data[copy_id]} if copy_id in data else {} + if not data: + warnings.append(f"Copy id {copy_id} was not found in correction.json") + return AnnotationLoadResult(data, warnings) diff --git a/copienator/cli.py b/copienator/cli.py index 68ec86b..2b9cbae 100644 --- a/copienator/cli.py +++ b/copienator/cli.py @@ -50,6 +50,16 @@ def evaluation_parser(description: str) -> argparse.ArgumentParser: return parser +def target_parser(description: str) -> argparse.ArgumentParser: + parser = standard_parser(description) + parser.add_argument( + "target", + type=Path, + help="Evaluation directory or nested file to process", + ) + return parser + + def evaluation_workspace( path: str | Path, *, @@ -77,6 +87,21 @@ def workspace_from_args( return evaluation_workspace(args.evaluation, repository=repository) +def workspace_from_target( + args: argparse.Namespace, + *, + repository: str | Path | None = None, +) -> tuple[EvaluationWorkspace, Path]: + target = Path(args.target).expanduser().resolve() + if not target.exists(): + raise CliError(f"Target does not exist: {target}", ExitCode.INVALID_WORKSPACE) + repository_path = Path(repository) if repository is not None else None + if target.is_dir() and EvaluationWorkspace.looks_like_evaluation(target): + return EvaluationWorkspace(target, repository_path), target + workspace = EvaluationWorkspace.discover(target, repository=repository_path) + return workspace, target + + def execute( parser: argparse.ArgumentParser, argv: Sequence[str] | None, diff --git a/copienator/filesystem.py b/copienator/filesystem.py new file mode 100644 index 0000000..e9780ee --- /dev/null +++ b/copienator/filesystem.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +import shutil +import uuid +from contextlib import contextmanager +from pathlib import Path + + +def _remove_path(path: Path) -> None: + if path.is_symlink() or path.is_file(): + path.unlink() + elif path.exists(): + shutil.rmtree(path) + + +@contextmanager +def staged_directory(destination: str | Path): + """Build a directory beside its destination and replace on success.""" + target = Path(destination) + target.parent.mkdir(parents=True, exist_ok=True) + token = uuid.uuid4().hex + staging = target.with_name(f".{target.name}.{token}.tmp") + backup = target.with_name(f".{target.name}.{token}.backup") + staging.mkdir() + committed = False + try: + yield staging + if target.exists(): + target.replace(backup) + try: + staging.replace(target) + committed = True + except Exception: + if backup.exists() and not target.exists(): + backup.replace(target) + raise + if backup.exists(): + _remove_path(backup) + finally: + if staging.exists(): + _remove_path(staging) + if not committed and backup.exists() and not target.exists(): + backup.replace(target) diff --git a/tests/test_gui_core.py b/tests/test_gui_core.py index b24448d..4eb5724 100644 --- a/tests/test_gui_core.py +++ b/tests/test_gui_core.py @@ -13,6 +13,7 @@ from contextlib import redirect_stderr from pathlib import Path from unittest.mock import patch +from PIL import Image from pypdf import PdfReader, PdfWriter from copienator import ( @@ -22,7 +23,10 @@ from copienator import ( atomic_update_json, atomic_write_json, read_json, + workspace_from_target, ) +from copienator.annotation_data import AnnotationLoadResult, load_annotation_data +from copienator.filesystem import staged_directory from copienator_gui.app import process_status from copienator_gui.diagnostics import collect_diagnostics from copienator_gui.runner import ProcessRunner @@ -156,10 +160,95 @@ class AtomicJsonTests(unittest.TestCase): self.assertEqual(persisted["steps"]["labels"]["status"], "success") +class AnnotationDataTests(unittest.TestCase): + def test_loader_indexes_coordinates_without_mutating_correction(self) -> None: + with tempfile.TemporaryDirectory() as directory: + evaluation = Path(directory) / "Exam" + copy_dir = evaluation / "Copies" / "Copie01" + group_dir = evaluation / "Par label" / "Ex 1" + copy_dir.mkdir(parents=True) + group_dir.mkdir(parents=True) + (copy_dir / "Ex 1_new.pdf").write_bytes(b"pdf") + correction = { + "Ex 1": [ + [ + { + "id": "01", + "result": { + "suffix": "_new", + "feedback": [{"text": "x", "box_2d": [100, 200, 300, 400]}], + }, + } + ] + ] + } + atomic_write_json(evaluation / "correction.json", correction) + atomic_write_json( + group_dir / "Group_1.json", + [["01", 10, 90, 1.0, "Ex 1"]], + ) + Image.new("RGB", (100, 200), "white").save(group_dir / "Group_1.jpg") + + loaded = load_annotation_data(EvaluationWorkspace(evaluation)) + item = loaded.data["01"]["Ex 1"] + self.assertEqual(item["pdf_path"], copy_dir / "Ex 1_new.pdf") + self.assertEqual(item["coordinates"], (10, 90)) + self.assertEqual(item["result"]["feedback"][0]["box_2d"], [20, 20, 60, 40]) + self.assertEqual(read_json(evaluation / "correction.json"), correction) + self.assertEqual(loaded.warnings, []) + + def test_refaire_filter_adds_a_missing_correction_entry(self) -> None: + with tempfile.TemporaryDirectory() as directory: + evaluation = Path(directory) / "Exam" + copy_dir = evaluation / "Copies" / "Copie01" + copy_dir.mkdir(parents=True) + (evaluation / "Par label").mkdir() + (copy_dir / "Ex 2.pdf").write_bytes(b"pdf") + atomic_write_json(evaluation / "correction.json", {}) + + loaded = load_annotation_data( + EvaluationWorkspace(evaluation), + refaire_list=[["Copie01", ["Ex 2"]]], + ) + item = loaded.data["01"]["Ex 2"] + self.assertEqual(item["pdf_path"], copy_dir / "Ex 2.pdf") + self.assertEqual(item["result"]["error"], "non traité") + + def test_staged_directory_preserves_then_replaces_destination(self) -> None: + with tempfile.TemporaryDirectory() as directory: + destination = Path(directory) / "output" + destination.mkdir() + (destination / "state.txt").write_text("old", encoding="utf-8") + + with self.assertRaises(RuntimeError), staged_directory( + destination + ) as staging: + (staging / "state.txt").write_text("broken", encoding="utf-8") + raise RuntimeError("rendering failed") + self.assertEqual( + (destination / "state.txt").read_text(encoding="utf-8"), "old" + ) + + with staged_directory(destination) as staging: + (staging / "state.txt").write_text("new", encoding="utf-8") + self.assertEqual( + (destination / "state.txt").read_text(encoding="utf-8"), "new" + ) + leftovers = [path for path in destination.parent.iterdir() if path.name.startswith(".output.")] + self.assertEqual(leftovers, []) + + class StandardCliTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.modules = { + "annotating": load_script_module("annotating.py", "annotating"), + "annotating_with_checks": load_script_module( + "annotating_with_checks.py", "annotating_with_checks" + ), + "annotating_by_label": load_script_module( + "annotating_by_label.py", "annotating_by_label" + ), "copies_tools": load_script_module( "copies_tools.py", "copienator_copies_tools_test" ), @@ -192,6 +281,9 @@ class StandardCliTests(unittest.TestCase): "resolve_manual": [missing], "verify_groups": [missing], "copies_tools": ["rotate", missing], + "annotating": [missing], + "annotating_with_checks": [missing], + "annotating_by_label": [missing], } for name, arguments in invocations.items(): with self.subTest(script=name), redirect_stderr(io.StringIO()): @@ -259,6 +351,21 @@ class StandardCliTests(unittest.TestCase): "default", {"target": evaluation}, ), + "annotating": ( + "annotation", + "simple", + {"target": evaluation, "overwrite": True}, + ), + "annotating_with_checks": ( + "annotation", + "checks", + {"target": evaluation, "overwrite": True, "refaire": True}, + ), + "annotating_by_label": ( + "annotation", + "grouped", + {"target": evaluation, "overwrite": True}, + ), } for module_name, (step_id, variant_id, values) in cases.items(): step = steps[step_id] @@ -266,7 +373,8 @@ class StandardCliTests(unittest.TestCase): command = build_command(REPOSITORY, step, variant, values, evaluation) with self.subTest(script=module_name): parsed = self.modules[module_name].build_parser().parse_args(command[3:]) - self.assertEqual(str(parsed.evaluation), evaluation) + parsed_path = getattr(parsed, "evaluation", None) or parsed.target + self.assertEqual(str(parsed_path), evaluation) for step_id in ("rotate", "rename"): step = steps[step_id] @@ -458,6 +566,104 @@ class StandardCliTests(unittest.TestCase): ) self.assertEqual(module.main([str(evaluation)]), 0) + def test_checked_annotation_discovers_workspace_from_copy_pdf(self) -> None: + module = self.modules["annotating_with_checks"] + with tempfile.TemporaryDirectory() as directory: + evaluation = Path(directory) / "Exam" + copy_pdf = evaluation / "Copies" / "Copie40.pdf" + copy_pdf.parent.mkdir(parents=True) + copy_pdf.write_bytes(b"pdf") + args = module.build_parser().parse_args([str(copy_pdf)]) + workspace, target = workspace_from_target(args) + self.assertEqual(workspace.root, evaluation) + self.assertEqual(target, copy_pdf) + self.assertEqual(module._copy_id_from_target(workspace, target), "40") + + def test_checked_refaire_requires_refaire_file(self) -> None: + module = self.modules["annotating_with_checks"] + with tempfile.TemporaryDirectory() as directory: + evaluation = Path(directory) / "Exam" + (evaluation / "Copies").mkdir(parents=True) + (evaluation / "Par label").mkdir() + (evaluation / "labels").write_text("Ex 1\n", encoding="utf-8") + atomic_write_json(evaluation / "correction.json", {}) + with redirect_stderr(io.StringIO()): + self.assertEqual(module.main([str(evaluation), "--refaire"]), 3) + + def test_checked_render_failure_preserves_previous_student_output(self) -> None: + module = self.modules["annotating_with_checks"] + with tempfile.TemporaryDirectory() as directory: + evaluation = Path(directory) / "Exam" + copy_dir = evaluation / "Copies" / "Copie01" + copy_dir.mkdir(parents=True) + (evaluation / "Par label").mkdir() + (evaluation / "labels").write_text("Ex 1\n", encoding="utf-8") + atomic_write_json(evaluation / "correction.json", {}) + answer = copy_dir / "Ex 1.pdf" + answer.write_bytes(b"pdf") + previous = evaluation / "Bnot" / "Copie01" + previous.mkdir(parents=True) + (previous / "sentinel.txt").write_text("old", encoding="utf-8") + loaded = AnnotationLoadResult( + { + "01": { + "Ex 1": { + "pdf_path": answer, + "result": {"feedback": [], "score": 1}, + "coordinates": (0, 0), + } + } + }, + [], + ) + with patch.object( + module, "load_annotation_data", return_value=loaded + ), patch.object( + self.modules["annotating"], + "make_base_image", + side_effect=RuntimeError("render failed"), + ), redirect_stderr(io.StringIO()): + self.assertEqual(module.main([str(evaluation), "--overwrite"]), 1) + self.assertEqual( + (previous / "sentinel.txt").read_text(encoding="utf-8"), "old" + ) + + def test_grouped_overwrite_preserves_previous_output_when_incomplete(self) -> None: + module = self.modules["annotating_by_label"] + with tempfile.TemporaryDirectory() as directory: + evaluation = Path(directory) / "Exam" + (evaluation / "Copies").mkdir(parents=True) + (evaluation / "Par label").mkdir() + (evaluation / "labels").write_text("Ex 1\n", encoding="utf-8") + (evaluation / "label_groups").write_text("Ex 1\n", encoding="utf-8") + atomic_write_json(evaluation / "correction.json", {}) + previous = evaluation / "BGnot" + previous.mkdir() + (previous / "sentinel.txt").write_text("old", encoding="utf-8") + loaded = AnnotationLoadResult( + {"01": {"Ex 1": {"pdf_path": Path("missing")}}}, + [], + ) + with patch.object( + module, "load_annotation_data", return_value=loaded + ), patch.object(module, "_generate_groups", return_value=(1, True)): + self.assertEqual(module.main([str(evaluation), "--overwrite"]), 4) + self.assertEqual( + (previous / "sentinel.txt").read_text(encoding="utf-8"), "old" + ) + + def test_grouped_batching_does_not_split_one_student(self) -> None: + module = self.modules["annotating_by_label"] + image = Image.new("RGB", (10, 60), "white") + rendered = [ + ("01", "Ex 1", image, 0, []), + ("01", "Ex 2", image, 0, []), + ("02", "Ex 1", image, 0, []), + ] + with patch.object(module, "MAX_HEIGHT_PX", 100): + batches = module.split_batches(rendered) + self.assertEqual([len(batch) for batch in batches], [2, 1]) + class WorkflowTests(unittest.TestCase): def setUp(self) -> None: