diff --git a/Readme.org b/Readme.org index 2520499..d16016b 100644 --- a/Readme.org +++ b/Readme.org @@ -173,6 +173,7 @@ scripts migrés vers cette convention sont actuellement : - =post-correction.py= et =resolve_manual.py= ; - =annotating.py=, =annotating_with_checks.py= et =annotating_by_label.py= ; +- =reading_annotations.py= et =reading_grouped_annotations.py= ; - =export.py=, =import.py= et =giving_names.py=. ** Correction d'un paquet de copies @@ -398,11 +399,15 @@ _Before_ : vider le dossier configuré par =IMPORT_DIR= (par défaut 2. =python reading_annotations.py Interro= Lit les =Concat_annotated= dans =Bnot=, regénère les copies avec - les modifications. + les modifications. Les fichiers générés (=score.json=, + =Concat.jpg=, etc.) sont préparés séparément puis installés + ensemble. Les entrées du dossier =Bnot= restent en place et une + erreur de génération conserve les anciennes sorties. OU 2. =python reading_grouped_annotations.py Interro= - Idem, mais pour =BGnot=. + Idem, mais pour =BGnot=. Les tâches parallèles remontent leurs + erreurs au processus principal au lieu de les ignorer. 3. =python giving_names.py Interro BGnot= @@ -455,6 +460,9 @@ groupée into refaire !! 6. =python import.py --refaire Interro24= 7. =python reading_grouped_annotations.py --refaire Interro24= + Avec =--refaire=, =refaire.json= et le dossier =BRnot= sont des + prérequis obligatoires ; leur absence produit le code de sortie 3. + ** Exemple de replotting, refaire d'une copie 1. replot it. diff --git a/copienator/annotation_actions.py b/copienator/annotation_actions.py new file mode 100644 index 0000000..1fc8575 --- /dev/null +++ b/copienator/annotation_actions.py @@ -0,0 +1,84 @@ +from __future__ import annotations + +from collections import defaultdict +from collections.abc import Callable +from pathlib import Path +from typing import Any + +from .json_io import read_json + +Log = Callable[[str], None] + + +def apply_checkbox_actions( + labels_data: dict[str, dict[str, Any]], + actions: list[dict[str, Any]], + log: Log, +) -> set[str]: + actions_by_label: defaultdict[str, list[dict[str, Any]]] = defaultdict(list) + for action in actions: + actions_by_label[str(action.get("label", ""))].append(action) + + dirty_labels: set[str] = set() + for label, label_actions in actions_by_label.items(): + if label not in labels_data: + continue + result = labels_data[label]["result"] + feedbacks = result.get("feedback", []) + global_feedbacks = [item for item in feedbacks if not item.get("box_2d")] + local_feedbacks = [item for item in feedbacks if item.get("box_2d")] + local_feedbacks.sort(key=lambda item: item["box_2d"][0]) + + for action in label_actions: + action_type = action.get("type") + if action_type == "score": + result["score"] = action.get("value") + dirty_labels.add(label) + log(f" > Updated score for {label} to {action.get('value')}") + elif action_type == "clear_all": + for feedback in feedbacks: + feedback["to_delete"] = True + if feedback.get("box_2d"): + feedback["norectangle"] = True + dirty_labels.add(label) + log(f" > Cleared all feedbacks in {label}") + elif action_type == "del_global": + index = int(action.get("index", -1)) + if 0 <= index < len(global_feedbacks): + global_feedbacks[index]["to_delete"] = True + dirty_labels.add(label) + log(f" > Deleted global feedback in {label}") + elif action_type in {"del_local", "del_local_rect"}: + index = int(action.get("index", -1)) + if 0 <= index < len(local_feedbacks): + target = local_feedbacks[index] + if action_type == "del_local": + target["to_delete"] = True + log(f" > Deleted local feedback in {label}") + else: + target["norectangle"] = True + log(f" > Deleted rectangle in {label}") + dirty_labels.add(label) + return dirty_labels + + +def apply_score_overrides( + labels_data: dict[str, dict[str, Any]], + score_path: Path, + log: Log, +) -> set[str]: + if not score_path.exists(): + return set() + loaded = read_json(score_path) + if not isinstance(loaded, dict): + raise TypeError(f"Expected a JSON object in {score_path}") + dirty: set[str] = set() + for label, score in loaded.items(): + if label not in labels_data: + continue + current = str(labels_data[label]["result"].get("score", 0)) + if current != str(score): + labels_data[label]["result"]["score"] = score + dirty.add(label) + log(f" > Overrode score for {label} to {score} from score.json") + return dirty diff --git a/copienator/filesystem.py b/copienator/filesystem.py index e9780ee..3bdb6d8 100644 --- a/copienator/filesystem.py +++ b/copienator/filesystem.py @@ -41,3 +41,43 @@ def staged_directory(destination: str | Path): _remove_path(staging) if not committed and backup.exists() and not target.exists(): backup.replace(target) + + +@contextmanager +def staged_files(destination: str | Path): + """Stage a set of files and merge them into a directory with rollback.""" + target = Path(destination) + target.parent.mkdir(parents=True, exist_ok=True) + token = uuid.uuid4().hex + staging = target.parent / f".{target.name}.{token}.files.tmp" + backup = target.parent / f".{target.name}.{token}.files.backup" + staging.mkdir() + committed: list[Path] = [] + try: + yield staging + staged = sorted(path for path in staging.iterdir() if path.is_file()) + target.mkdir(parents=True, exist_ok=True) + backup.mkdir() + try: + for source in staged: + destination_path = target / source.name + if destination_path.exists() or destination_path.is_symlink(): + destination_path.replace(backup / source.name) + source.replace(destination_path) + committed.append(destination_path) + except Exception: + for destination_path in reversed(committed): + _remove_path(destination_path) + for saved in backup.iterdir(): + saved.replace(target / saved.name) + raise + _remove_path(backup) + finally: + if staging.exists(): + _remove_path(staging) + if backup.exists(): + for saved in backup.iterdir(): + destination_path = target / saved.name + if not destination_path.exists(): + saved.replace(destination_path) + _remove_path(backup) diff --git a/reading_annotations.py b/reading_annotations.py index 5b9a20f..c848338 100644 --- a/reading_annotations.py +++ b/reading_annotations.py @@ -1,380 +1,286 @@ -import sys -import os -import json -import numpy as np -import shutil +from __future__ import annotations + +import argparse +from collections.abc import Sequence from pathlib import Path -from PIL import Image, ImageChops, ImageFilter -Image.MAX_IMAGE_PIXELS = None +from typing import Any + +import numpy as np from pdf2image import convert_from_path -import annotating # Reuse rendering logic +from PIL import Image, ImageChops, ImageDraw, ImageFilter -DPI = 100 +import annotating +import utils +from copienator import ( + EvaluationWorkspace, + ExitCode, + atomic_write_json, + evaluation_parser, + execute, + read_json, + workspace_from_args, +) +from copienator.annotation_actions import apply_checkbox_actions, apply_score_overrides +from copienator.annotation_data import AnnotationData, load_annotation_data +from copienator.filesystem import staged_files -def detect_checks_and_notes(output_dir): - """ - Returns: - actions: List of dicts {type, label, ...} for checked boxes - notes_img: RGBA image of manual notes (checks masked out) - """ +Image.MAX_IMAGE_PIXELS = None - names = ["Concat_annotated.pdf"] - for name in names: - pdf_path = os.path.join(output_dir, name) - if os.path.exists(pdf_path): - break - # ref_path = os.path.join(output_dir, "Reference.png") - ref_path = os.path.join(output_dir, "Reference.jpg") - json_path = os.path.join(output_dir, "checkboxes.json") - if not (os.path.exists(pdf_path) and os.path.exists(ref_path)): - print(f"\tMissing annotated file in {output_dir}") +def detect_checks_and_notes( + output_dir: str | Path, +) -> tuple[list[dict[str, Any]], Image.Image | None]: + """Detect checked boxes and extract handwritten notes from an annotated PDF.""" + directory = Path(output_dir) + pdf_path = directory / "Concat_annotated.pdf" + reference_path = directory / "Reference.jpg" + boxes_path = directory / "checkboxes.json" + missing = [ + path.name + for path in (pdf_path, reference_path, boxes_path) + if not path.is_file() + ] + if missing: + print(f"\tMissing annotation input in {directory}: {', '.join(missing)}") return [], None - # Load Coordinates - with open(json_path, 'r') as f: - boxes = json.load(f) + boxes = read_json(boxes_path) + if not isinstance(boxes, list): + raise TypeError(f"Expected a JSON array in {boxes_path}") + with Image.open(reference_path) as opened_reference: + reference = opened_reference.convert("RGB").copy() - # Load Reference - ref_img = Image.open(ref_path).convert("RGB") - - # Load User PDF (First page only, assuming it's one long strip) - # Warning: If the PDF is huge, pdf2image might split pages or OOM. - # Assuming user didn't change page dimensions/order. try: - # user_pages = convert_from_path(pdf_path, dpi=DPI) - # La version suivante évite les size mismatch - # Mais donne plus de bruit - user_pages = convert_from_path(pdf_path, dpi=72) - except Exception as e: - print(f"Error reading PDF: {e}") + pages = convert_from_path(pdf_path, dpi=72) + except Exception as exc: # noqa: BLE001 - PDF backends expose many errors + print(f"Error reading PDF {pdf_path}: {exc}") + return [], None + if not pages: + print(f"Error reading PDF {pdf_path}: no page found") return [], None - # Concatenate PDF pages back to one image if user saved as multiple pages - total_h = sum(p.height for p in user_pages) - user_img = Image.new("RGB", (user_pages[0].width, total_h)) - y = 0 - for p in user_pages: - user_img.paste(p, (0, y)) - y += p.height - # Resize user_img to match ref_img if slight mismatch (DPI export diffs) - if user_img.size != ref_img.size: - print("Debug : size mismatch : ", user_img.size, ref_img.size) - user_img = user_img.resize(ref_img.size, Image.Resampling.LANCZOS) + user_image = Image.new("RGB", (pages[0].width, sum(page.height for page in pages))) + current_y = 0 + for page in pages: + user_image.paste(page.convert("RGB"), (0, current_y)) + current_y += page.height + if user_image.size != reference.size: + print(f" Resizing annotated PDF from {user_image.size} to {reference.size}") + user_image = user_image.resize(reference.size, Image.Resampling.LANCZOS) - # --- Detection Phase --- - actions = [] + difference = np.abs( + np.array(reference).astype(int) - np.array(user_image).astype(int) + ).astype(np.uint8) + difference_gray = np.mean(difference, axis=2) + keep_mask = Image.new("L", reference.size, 255) + mask_draw = ImageDraw.Draw(keep_mask) + actions: list[dict[str, Any]] = [] - # Convert to numpy for analysis - ref_arr = np.array(ref_img) - user_arr = np.array(user_img) - - # Diff for analysis - # Simple absolute difference - diff = np.abs(ref_arr.astype(int) - user_arr.astype(int)).astype(np.uint8) - # Convert to grayscale for thresholding - diff_gray = np.mean(diff, axis=2) - - # Threshold for "Checked" - CHECK_THRESHOLD = 30 # intensity diff - DENSITY_THRESHOLD = 0.05 # 5% of pixels darkened - - # Mask to hide checkmarks from the "Notes" extraction - mask_img = Image.new("L", ref_img.size, 255) # White (255) = keep, Black (0) = hide - mask_draw = ImageDraw.Draw(mask_img) - - for box in boxes: - # global_box: [x1, y1, x2, y2] - b = box['global_box'] - x1, y1, x2, y2 = map(int, b) - - # Ensure bounds + for raw_box in boxes: + if not isinstance(raw_box, dict) or "global_box" not in raw_box: + continue + x1, y1, x2, y2 = map(int, raw_box["global_box"]) x1, y1 = max(0, x1), max(0, y1) - x2, y2 = min(ref_img.width, x2), min(ref_img.height, y2) - - # Analyze ROI - roi = diff_gray[y1+5:y2-5, x1+5:x2-5] - if roi.size == 0: continue - - changed_pixels = np.sum(roi > CHECK_THRESHOLD) - density = changed_pixels / roi.size - - if density > DENSITY_THRESHOLD: - # print("A checked box !", density, b) - actions.append(box) - # It's checked, so we mask this area out for manual notes - # Expand mask slightly to catch sloppy ticks - mask_draw.rectangle([x1-15, y1-15, x2+15, y2+15], fill=0) + x2, y2 = min(reference.width, x2), min(reference.height, y2) + region = difference_gray[y1 + 5 : y2 - 5, x1 + 5 : x2 - 5] + if region.size == 0: + continue + density = np.sum(region > 30) / region.size + if density > 0.05: + actions.append(raw_box) + mask_draw.rectangle([x1 - 15, y1 - 15, x2 + 15, y2 + 15], fill=0) else: - mask_draw.rectangle([x1-2, y1-2, x2+2, y2+2], fill=0) - - if box["type"] == "score" and box["value"] == 0.0: - # Mask the whole line - mask_draw.rectangle([0, y1-10, ref_img.width, y2+10], fill=0) - - # --- Extraction Phase --- - - # 150 + no blur is alright, with some lines at the end - # 100 + 2 px blur is too clean : tes annotations sont morcelées - # 50 + 2 px blur seems good - - ref_blur = ref_img.filter(ImageFilter.GaussianBlur(2)) - user_blur = user_img.filter(ImageFilter.GaussianBlur(2)) - - # 1. Get difference image - # diff_img = ImageChops.difference(ref_img, user_img).convert("L") - diff_img = ImageChops.difference(ref_blur, user_blur).convert("L") - - diff_data = np.array(diff_img) - alpha = np.where(diff_data > 50, 255, 0).astype(np.uint8) - - notes = user_img.convert("RGBA") - r, g, b, a = notes.split() - - # Combine the diff-based alpha with the box-mask - mask_arr = np.array(mask_img) - final_alpha = np.minimum(alpha, mask_arr) + mask_draw.rectangle([x1 - 2, y1 - 2, x2 + 2, y2 + 2], fill=0) + if raw_box.get("type") == "score" and raw_box.get("value") == 0.0: + mask_draw.rectangle([0, y1 - 10, reference.width, y2 + 10], fill=0) + reference_blur = reference.filter(ImageFilter.GaussianBlur(2)) + user_blur = user_image.filter(ImageFilter.GaussianBlur(2)) + diff_image = ImageChops.difference(reference_blur, user_blur).convert("L") + alpha = np.where(np.array(diff_image) > 50, 255, 0).astype(np.uint8) + final_alpha = np.minimum(alpha, np.array(keep_mask)) + notes = user_image.convert("RGBA") notes.putalpha(Image.fromarray(final_alpha)) - - # notes.show() - return actions, notes -from PIL import ImageDraw -from utils import natural_key -from annotating import MARGIN_LEFT, ANNOT_WIDTH - -def has_significant_notes(note_img, threshold=20): - """Checks if the note layer has visible content (non-transparent pixels).""" - # Assuming note_img is RGBA. - # We check alpha channel for non-zero values (or low transparency) - # Since we generated notes with variable alpha based on diff, checking alpha sum is good. - if note_img.mode != 'RGBA': +def has_significant_notes(note_img: Image.Image | None, threshold: int = 20) -> bool: + """Return whether an RGBA note layer contains enough visible pixels.""" + if note_img is None or note_img.mode != "RGBA": return False alpha = np.array(note_img)[:, :, 3] - # Count pixels with significant opacity - visible_pixels = np.sum(alpha > 50) - # visible_pixels_bis = np.sum(alpha > 200) - # if visible_pixels > 0: - # print(f"Debug : visible pixels is {visible_pixels}") - return visible_pixels > threshold + return bool(np.sum(alpha > 50) > threshold) -def apply_actions_and_regenerate(root_dir, data, student_id, actions, notes_layer, - all_labels, update_score=False): - """ - Modifies data based on actions, reads bnote.json, cuts notes, - regenerates all label images for consistency, saves dirty ones, - and generates Concat.jpg. - """ - output_dir = os.path.join(root_dir, "Bnot", f"Copie{student_id}") - bnote_path = os.path.join(output_dir, "bnote.json") - score_path = os.path.join(output_dir, "score.json") - if not os.path.exists(bnote_path): - print(f"Error: bnote.json not found in {output_dir}") - return +def concatenate(images: list[Image.Image]) -> Image.Image | None: + if not images: + return None + result = Image.new( + "RGB", + (max(image.width for image in images), sum(image.height for image in images)), + "white", + ) + current_y = 0 + for image in images: + result.paste(image, (0, current_y)) + current_y += image.height + return result - with open(bnote_path, 'r') as f: - bnote_data = json.load(f) + +def apply_actions_and_regenerate( + workspace: EvaluationWorkspace, + data: AnnotationData, + student_id: str, + actions: list[dict[str, Any]], + notes_layer: Image.Image | None, + all_labels: list[str], + *, + update_score: bool = False, +) -> ExitCode: + """Apply annotations and atomically merge the regenerated student files.""" + output_dir = workspace.annotation_dir("checks") / f"Copie{student_id}" + bnote_path = output_dir / "bnote.json" + if not bnote_path.is_file(): + print(f" Missing {bnote_path}") + return ExitCode.PARTIAL + bnote_data = read_json(bnote_path) + if not isinstance(bnote_data, dict): + raise TypeError(f"Expected a JSON object in {bnote_path}") labels_data = data[student_id] + dirty_labels = apply_checkbox_actions(labels_data, actions, print) + if update_score: + dirty_labels |= apply_score_overrides(labels_data, output_dir / "score.json", print) - # --- 1. Apply Actions to Data (Update scores / Flags for deletion) --- - actions_by_label = {} - for a in actions: - actions_by_label.setdefault(a['label'], []).append(a) - - dirty_labels = set() # Labels that logic says changed - - for label, acts in actions_by_label.items(): - if label not in labels_data: continue + scores = dict.fromkeys(all_labels, "") + dirty_images: dict[str, Image.Image] = {} + concatenated: list[Image.Image] = [] + filtered: list[Image.Image] = [] + incomplete = False + for image_info in bnote_data.get("images", []): + if not isinstance(image_info, dict): + incomplete = True + continue + label = str(image_info.get("label", "")) + if label not in labels_data: + incomplete = True + continue content = labels_data[label] - result = content['result'] - feedbacks = result.get('feedback', []) + result = content["result"] + scores[label] = str(result.get("score", 0)) - # Helpers to find objects by index (references match those in feedbacks list) - global_fb = [f for f in feedbacks if not f.get('box_2d')] - local_fb = [f for f in feedbacks if f.get('box_2d')] - local_fb.sort(key=lambda x: x['box_2d'][0]) - - for act in acts: - if act['type'] == 'score': - result['score'] = act['value'] - dirty_labels.add(label) - print(f" > Updated score for {label} to {act['value']}") - - elif act['type'] == 'clear_all': - for fb in feedbacks: - fb["to_delete"] = True - if fb.get("box_2d"): - fb["norectangle"] = True - dirty_labels.add(label) - print(f" > Cleared all feedbacks in {label}") - - elif act['type'] == 'del_global': - if act['index'] < len(global_fb): - global_fb[act['index']]["to_delete"] = True - dirty_labels.add(label) - print(f" > Deleted global feedback in {label}") - - elif act['type'] in ('del_local', 'del_local_rect'): - if act['index'] < len(local_fb): - target = local_fb[act['index']] - if act['type'] == 'del_local': - target["to_delete"] = True - print(f" > Deleted local feedback in {label}") - else: - target["norectangle"] = True - print(f" > Deleted rect in {label}") - dirty_labels.add(label) - - # --- 1.5 Override with existing score.json if requested --- - if update_score and os.path.exists(score_path): - try: - with open(score_path, "r") as f: - existing_scores = json.load(f) - for label, existing_score in existing_scores.items(): - if label in labels_data: - current_score = str(labels_data[label]['result'].get('score', 0)) - # If manually modified, override the result and mark dirty - if current_score != str(existing_score): - labels_data[label]['result']['score'] = existing_score - dirty_labels.add(label) - print(f" > Overrode score for {label} to {existing_score} from existing score.json") - except json.JSONDecodeError: - print(f" > Warning: Could not read existing {score_path}") - - - # --- 2. Process Images (Cut notes, Regenerate, Concatenate) --- - concat_list = [] - concat_list_F = [] - d_notes = dict.fromkeys(all_labels, "") - - # Iterate over images defined in bnote.json to maintain order/geometry - for img_info in bnote_data.get("images", []): - label = img_info["label"] - if label not in labels_data: continue - - # Update scores dict - content = labels_data[label] - result = content['result'] - d_notes[label] = str(result.get('score', 0)) - - # A. Cut Manual Notes - hmin, hmax = img_info["hmin"], img_info["hmax"] sub_note = None - if notes_layer: + if notes_layer is not None: + hmin = int(image_info.get("hmin", 0)) + hmax = int(image_info.get("hmax", 0)) sub_note = notes_layer.crop((0, hmin, notes_layer.width, hmax)) - has_notes = has_significant_notes(sub_note) - # B. Regenerate Label Image - # We always regenerate to ensure Concat.jpg is consistent with any modifications - # pdf_path = Path(root_dir) / "Copies" / f"Copie{student_id}" / f"{label}.pdf" - pdf_path = content.get('pdf_path') # Contient le suffixe _new si nécessaire - if not os.path.exists(pdf_path): continue - - (base_img, _, _) = annotating.make_base_image(pdf_path) - - # Compose uses the result object we modified in step 1 - final_img, new_header_h = annotating.compose_label_image( - base_img, label, content['result'], content['coordinates'][0], - with_error=False + pdf_path = Path(content["pdf_path"]) + if not pdf_path.is_file(): + print(f" Missing answer PDF: {pdf_path}") + incomplete = True + continue + base_image, _, _ = annotating.make_base_image(pdf_path) + final_image, new_header_height = annotating.compose_label_image( + base_image, + label, + result, + content["coordinates"][0], + with_error=False, ) - if final_img==None: + if final_image is None: + incomplete = True continue - # Overlay manual notes - if has_notes: - old_header_h = int(img_info.get("header_height", 0)) - w, h = sub_note.size + if has_notes and sub_note is not None: + old_header_height = int(image_info.get("header_height", 0)) + width, height = sub_note.size + if old_header_height > 0: + 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) - # 1. Paste header ink at the top - if old_header_h > 0: - header_crop = sub_note.crop((0, 0, w, min(h, old_header_h))) - final_img.paste(header_crop, (0, 0), mask=header_crop) + if label in dirty_labels or has_notes: + dirty_images[label] = final_image + concatenated.append(final_image) + if float(scores[label]) != 4.0 or result.get("feedback", []): + filtered.append(final_image) - # 2. Paste student-content ink at the new header height - if h > old_header_h: - body_crop = sub_note.crop((0, old_header_h, w, h)) - final_img.paste(body_crop, (0, new_header_h), mask=body_crop) + concat_image = concatenate(concatenated) + filtered_image = concatenate(filtered) + with staged_files(output_dir) as staging: + 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_image is not None: + filtered_image.save(staging / "Concat_F.jpg") - # C. Save individual file if Modified (Dirty logic or visual notes) - if (label in dirty_labels) or has_notes: - save_path = os.path.join(output_dir, f"{label}.jpg") - final_img.save(save_path) - print(f" Saved dirty image: {label}.jpg") + print(f" Saved regenerated files in {output_dir}") + return ExitCode.PARTIAL if incomplete else ExitCode.SUCCESS - concat_list.append(final_img) - perfect_no_comment = True - if float(d_notes[label]) != 4.0: - perfect_no_comment = False - if len(result.get('feedback', [])) != 0: - perfect_no_comment = False - if not perfect_no_comment: - concat_list_F.append(final_img) +def run(workspace: EvaluationWorkspace, *, update_score: bool = False) -> ExitCode: + workspace.require_files("labels", "correction.json") + workspace.require_directories("Copies", "Par label", "Bnot") + all_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("No annotation data found.") + return ExitCode.PARTIAL - # --- 3. Save Final Outputs --- - with open(score_path, "w") as f: - json.dump(d_notes, f, indent=4) - print(f" Saved {score_path}") + status = ExitCode.PARTIAL if loaded.warnings else ExitCode.SUCCESS + for student_id in sorted(loaded.data, key=utils.natural_key): + output_dir = workspace.annotation_dir("checks") / f"Copie{student_id}" + if not output_dir.is_dir(): + print(f"Warning: missing annotation directory {output_dir}") + status = ExitCode.PARTIAL + continue + print(f"Processing annotations for: {student_id}") + actions, notes = detect_checks_and_notes(output_dir) + if notes is None and not actions and not update_score: + print(" No readable annotation input found.") + status = ExitCode.PARTIAL + continue + result = apply_actions_and_regenerate( + workspace, + loaded.data, + student_id, + actions, + notes, + all_labels, + update_score=update_score, + ) + if result != ExitCode.SUCCESS: + status = ExitCode.PARTIAL + return status - if concat_list: - max_w = max(i.width for i in concat_list) - total_h = sum(i.height for i in concat_list) - full_img = Image.new("RGB", (max_w, total_h), "white") - y = 0 - for img in concat_list: - full_img.paste(img, (0, y)) - y += img.height +def build_parser() -> argparse.ArgumentParser: + parser = evaluation_parser("Read checked annotations and regenerate copies") + parser.add_argument( + "--update-score", + action="store_true", + help="Override generated scores with values from existing score.json files", + ) + return parser - full_img.save(os.path.join(output_dir, "Concat.jpg")) - print(f" Saved regenerated Concat.jpg") - if concat_list_F: - max_w = max(i.width for i in concat_list_F) - total_h = sum(i.height for i in concat_list_F) - full_img = Image.new("RGB", (max_w, total_h), "white") - y = 0 - for img in concat_list_F: - full_img.paste(img, (0, y)) - y += img.height +def main(argv: Sequence[str] | None = None) -> int: + parser = build_parser() + + def handle(args: argparse.Namespace) -> ExitCode: + return run(workspace_from_args(args), update_score=args.update_score) + + return execute(parser, argv, handle) - full_img.save(os.path.join(output_dir, "Concat_F.jpg")) - print(f" Saved regenerated Concat_F.jpg") -from utils import read_all_labels if __name__ == "__main__": - import argparse - parser = argparse.ArgumentParser(description="Read annotations and compile PDFs") - parser.add_argument("input_path", help="Directory path") - parser.add_argument("--update-score", action="store_true", help="Override scores with values from existing score.json") - args = parser.parse_args() - - root_dir = args.input_path - - try: - all_labels = read_all_labels(Path(root_dir)) - except FileNotFoundError: - all_labels = [] - - # Load original data - original_data = annotating.make_dictionary(root_dir) - - # Process each Bnot folder - for student_id in original_data.keys(): - bnot_dir = os.path.join(root_dir, "Bnot", f"Copie{student_id}") - if os.path.exists(bnot_dir): - print(f"Processing annotations for: {student_id}") - actions, notes = detect_checks_and_notes(bnot_dir) - if actions or notes or args.update_score: - apply_actions_and_regenerate(root_dir, original_data, student_id, - actions, notes, all_labels, - update_score=args.update_score) - else: - print(" No changes detected or missing files.") + raise SystemExit(main()) diff --git a/reading_grouped_annotations.py b/reading_grouped_annotations.py index 6c6aca9..d24c83d 100644 --- a/reading_grouped_annotations.py +++ b/reading_grouped_annotations.py @@ -1,435 +1,400 @@ -import sys -import os -import json -import collections +from __future__ import annotations + +import argparse import concurrent.futures +from collections import defaultdict +from collections.abc import Sequence from pathlib import Path +from typing import Any + from PIL import Image, ImageDraw -import threading import annotating +import utils +from copienator import ( + EvaluationWorkspace, + ExitCode, + atomic_write_json, + evaluation_parser, + execute, + read_json, + 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 reading_annotations import ( + concatenate, + detect_checks_and_notes, + has_significant_notes, +) -from utils import natural_key, pdf_image_of_enonce, pdf_image_of_solution, pdf_images_of_contexts -from reading_annotations import detect_checks_and_notes, has_significant_notes +LabelNotes = dict[str, dict[str, Any]] +ScanResult = tuple[dict[str, list[dict[str, Any]]], dict[str, LabelNotes]] -def get_extra_pdfs_as_images(root_dir, label, annotating_module, all_labels): - """Fetches Text and Sol pdfs for a given label and converts them to images.""" - extra_images = [] - a, b = pdf_image_of_enonce(root_dir, label), pdf_image_of_solution(root_dir, label) - e = pdf_images_of_contexts(root_dir, label, all_labels) - for c in e + [a, b]: - if c: - img, _, _ = annotating_module.make_base_image(c) - if img: - extra_images.append(img) - return extra_images +def get_extra_pdfs_as_images( + root_dir: str | Path, + label: str, + annotating_module: Any, + all_labels: list[str], +) -> list[Image.Image]: + """Convert the context, question and solution PDFs associated with a label.""" + paths = [ + *utils.pdf_images_of_contexts(root_dir, label, all_labels), + utils.pdf_image_of_enonce(root_dir, label), + utils.pdf_image_of_solution(root_dir, label), + ] + images = [] + for path in paths: + if path: + image, _, _ = annotating_module.make_base_image(path) + if image is not None: + images.append(image) + return images -def save_paginated_pdf(image_groups, output_path): - """Concatenates groups of images vertically, adding inner borders and margins.""" - if not image_groups: + +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: return - - max_w = max(img.width for group in image_groups for img in group) - max_page_h = int(max_w * 1.414 * 1.25) - - # Calculate sizes in pixels at 100 DPI - border_px = int((0.2 / 2.54) * 100) + max_width = max(image.width for group in non_empty for image in group) + max_page_height = int(max_width * 1.414 * 1.25) + border = int((0.2 / 2.54) * 100) left_margin = int((0.3 / 2.54) * 100) - tb_margin = int((0.2 / 2.54) * 100) + vertical_margin = int((0.2 / 2.54) * 100) + max_content_height = max_page_height - 2 * vertical_margin - # Available height for images once top/bottom margins are added - max_content_h = max_page_h - (2 * tb_margin) + pages: list[Image.Image] = [] + page_images: list[Image.Image] = [] + page_height = 0 - pages = [] - current_page_imgs = [] - current_h = 0 - - for group in image_groups: - if not group: - continue - - # Process the group to add borders - processed_group = [] - for i, img in enumerate(group): - if i in (0, 1): - img = img.copy() - draw = ImageDraw.Draw(img) - color = "black" if i == 0 else "blue" - - draw.rectangle( - [0, 0, img.width - 1, img.height - 1], - outline=color, - width=border_px - ) - processed_group.append(img) - - group_h = sum(img.height for img in processed_group) - - if current_page_imgs and (current_h + group_h > max_content_h): - # Create page with margins included in dimensions - page = Image.new("RGB", (max_w + left_margin, current_h + 2 * tb_margin), "white") - y = tb_margin - for c_img in current_page_imgs: - page.paste(c_img, (left_margin, y)) - y += c_img.height - pages.append(page) - - current_page_imgs = processed_group - current_h = group_h - else: - current_page_imgs.extend(processed_group) - current_h += group_h - - if current_page_imgs: - page = Image.new("RGB", (max_w + left_margin, current_h + 2 * tb_margin), "white") - y = tb_margin - for c_img in current_page_imgs: - page.paste(c_img, (left_margin, y)) - y += c_img.height - pages.append(page) - - if pages: - pages[0].save(output_path, "PDF", resolution=100.0, save_all=True, append_images=pages[1:]) - -def apply_actions_and_regenerate_grouped(root_dir, data, student_id, - actions, label_notes, all_labels, - update_score=False): - """ - Modifies data based on actions, pastes label-specific note crops, - regenerates label images for consistency, saves dirty ones, - and generates Concat.jpg in the BGnot/Copie{id} directory. - Returns a string of accumulated log messages. - """ - logs = [f"\nProcessing compilation for: Copie{student_id}"] - output_dir = os.path.join(root_dir, "BGnot", f"Copie{student_id}") - os.makedirs(output_dir, exist_ok=True) - - score_path = os.path.join(output_dir, "score.json") - labels_data = data.get(student_id, {}) - - # --- 1. Apply Actions to Data (Update scores / Flags for deletion) --- - actions_by_label = collections.defaultdict(list) - for a in actions: - actions_by_label[a['label']].append(a) - - dirty_labels = set() - - for label, acts in actions_by_label.items(): - if label not in labels_data: continue - - content = labels_data[label] - result = content['result'] - feedbacks = result.get('feedback', []) - - # Helpers to find objects by index - global_fb = [f for f in feedbacks if not f.get('box_2d')] - local_fb = [f for f in feedbacks if f.get('box_2d')] - local_fb.sort(key=lambda x: x['box_2d'][0]) - - for act in acts: - if act['type'] == 'score': - result['score'] = act['value'] - dirty_labels.add(label) - logs.append(f" > Updated score for {label} to {act['value']}") - - elif act['type'] == 'clear_all': - for fb in feedbacks: - fb["to_delete"] = True - if fb.get("box_2d"): - fb["norectangle"] = True - dirty_labels.add(label) - logs.append(f" > Cleared all feedbacks in {label}") - - elif act['type'] == 'del_global': - if act['index'] < len(global_fb): - global_fb[act['index']]["to_delete"] = True - dirty_labels.add(label) - logs.append(f" > Deleted global feedback in {label}") - - elif act['type'] in ('del_local', 'del_local_rect'): - if act['index'] < len(local_fb): - target = local_fb[act['index']] - if act['type'] == 'del_local': - target["to_delete"] = True - logs.append(f" > Deleted local feedback in {label}") - else: - target["norectangle"] = True - logs.append(f" > Deleted rect in {label}") - dirty_labels.add(label) - - # --- 1.5 Override with existing score.json if requested --- - if update_score and os.path.exists(score_path): - try: - with open(score_path, "r") as f: - existing_scores = json.load(f) - for label, existing_score in existing_scores.items(): - if label in labels_data: - current_score = str(labels_data[label]['result'].get('score', 0)) - # If manually modified, override the result and mark dirty - if current_score != str(existing_score): - labels_data[label]['result']['score'] = existing_score - dirty_labels.add(label) - logs.append(f" > Overrode score for {label} to {existing_score} from existing score.json") - except json.JSONDecodeError: - logs.append(f" > Warning: Could not read existing {score_path}") - - - # --- 2. Process Images (Regenerate & Concatenate) --- - concat_list = [] - concat_list_F = [] - d_notes = dict.fromkeys(all_labels, "") - - # Iterate over all labels naturally to assemble a complete student profile - sorted_labels = sorted(labels_data.items(), key=lambda x: natural_key(x[0])) - - for label, content in sorted_labels: - result = content['result'] - d_notes[label] = str(result.get('score', 0)) - - # pdf_path = Path(root_dir) / "Copies" / f"Copie{student_id}" / f"{label}.pdf" - pdf_path = content.get('pdf_path') - if not os.path.exists(pdf_path): continue - - (base_img, _, _) = annotating.make_base_image(pdf_path) - - # Compose uses the result object we modified in step 1 - final_img, new_header_h = annotating.compose_label_image( - base_img, label, content['result'], content['coordinates'][0], - with_error=False + def finish_page() -> None: + nonlocal page_images, page_height + if not page_images: + return + page = Image.new( + "RGB", + (max_width + left_margin, page_height + 2 * vertical_margin), + "white", ) - if final_img is None: + current_y = vertical_margin + for image in page_images: + page.paste(image, (left_margin, current_y)) + current_y += image.height + pages.append(page) + page_images = [] + page_height = 0 + + for group in non_empty: + processed: list[Image.Image] = [] + for index, image in enumerate(group): + if index in (0, 1): + image = image.copy() + color = "black" if index == 0 else "blue" + ImageDraw.Draw(image).rectangle( + [0, 0, image.width - 1, image.height - 1], + outline=color, + width=border, + ) + processed.append(image) + group_height = sum(image.height for image in processed) + if page_images and page_height + group_height > max_content_height: + finish_page() + page_images.extend(processed) + page_height += group_height + finish_page() + pages[0].save( + output_path, + "PDF", + resolution=100.0, + save_all=True, + append_images=pages[1:], + ) + + +def _scan_annotation_directory( + directory: Path, + only_ids: set[str] | None = None, + default_student_id: str | None = None, +) -> ScanResult: + bnote_path = directory / "bnote.json" + if not bnote_path.is_file(): + raise FileNotFoundError(f"Missing {bnote_path}") + bnote = read_json(bnote_path) + if not isinstance(bnote, dict): + raise TypeError(f"Expected a JSON object in {bnote_path}") + images = [item for item in bnote.get("images", []) if isinstance(item, dict)] + if only_ids and not any( + str(item.get("id", default_student_id)) in only_ids for item in images + ): + return {}, {} + + actions, notes_image = detect_checks_and_notes(directory) + if notes_image is None: + return {}, {} + actions_by_student: dict[str, list[dict[str, Any]]] = defaultdict(list) + notes_by_student: dict[str, LabelNotes] = defaultdict(dict) + for action in actions: + raw_student_id = action.get("student_id", default_student_id) + if raw_student_id is not None: + actions_by_student[str(raw_student_id)].append(action) + for image_info in images: + student_id = str(image_info.get("id", default_student_id or "")) + label = str(image_info.get("label", "")) + hmin = int(image_info.get("hmin", 0)) + hmax = int(image_info.get("hmax", 0)) + if student_id and label and hmax > hmin: + crop = notes_image.crop((0, hmin, notes_image.width, hmax)) + if has_significant_notes(crop): + notes_by_student[student_id][label] = { + "img": crop, + "old_header_h": int(image_info.get("header_height", 0)), + } + return dict(actions_by_student), dict(notes_by_student) + + +def _merge_scan_result( + target_actions: dict[str, list[dict[str, Any]]], + target_notes: dict[str, LabelNotes], + result: ScanResult, +) -> None: + actions, notes = result + for student_id, student_actions in actions.items(): + target_actions[student_id].extend(student_actions) + for student_id, student_notes in notes.items(): + target_notes[student_id].update(student_notes) + + +def apply_actions_and_regenerate_grouped( + workspace: EvaluationWorkspace, + data: AnnotationData, + student_id: str, + actions: list[dict[str, Any]], + label_notes: LabelNotes, + all_labels: list[str], + *, + update_score: bool = False, +) -> tuple[ExitCode, str]: + """Apply grouped annotations and atomically merge regenerated student files.""" + logs = [f"\nProcessing compilation for: Copie{student_id}"] + output_dir = workspace.annotation_dir("grouped") / f"Copie{student_id}" + labels_data = data.get(student_id, {}) + dirty_labels = apply_checkbox_actions(labels_data, actions, logs.append) + if update_score: + dirty_labels |= apply_score_overrides( + labels_data, output_dir / "score.json", logs.append + ) + + 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])): + result = content["result"] + scores[label] = str(result.get("score", 0)) + pdf_path = Path(content["pdf_path"]) + if not pdf_path.is_file(): + logs.append(f" Missing answer PDF: {pdf_path}") + incomplete = True + continue + base_image, _, _ = annotating.make_base_image(pdf_path) + final_image, new_header_height = annotating.compose_label_image( + base_image, + label, + result, + content["coordinates"][0], + with_error=False, + ) + if final_image is None: + incomplete = True continue - # Overlay manual notes specific to this label has_notes = False if label in label_notes: - note_info = label_notes[label] - sub_note = note_info['img'] - old_header_h = int(note_info['old_header_h']) + sub_note = label_notes[label]["img"] + old_header_height = int(label_notes[label]["old_header_h"]) + has_notes = has_significant_notes(sub_note) + 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))) + 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 has_significant_notes(sub_note): - has_notes = True - w, h = sub_note.size - - # 1. Paste header ink at the top - if old_header_h > 0: - header_crop = sub_note.crop((0, 0, w, min(h, old_header_h))) - final_img.paste(header_crop, (0, 0), mask=header_crop) - - # 2. Paste student-content ink at the new header height - if h > old_header_h: - body_crop = sub_note.crop((0, old_header_h, w, h)) - final_img.paste(body_crop, (0, new_header_h), mask=body_crop) - - # Save individual file if Modified (Dirty logic or visual notes) - if (label in dirty_labels) or has_notes: - save_path = os.path.join(output_dir, f"{label}.jpg") - final_img.save(save_path) + if label in dirty_labels or has_notes: + dirty_images[label] = final_image logs.append(f" Saved dirty image: {label}.jpg") + concat_images.append(final_image) - concat_list.append(final_img) + feedbacks = result.get("feedback", []) + perfect = float(scores[label]) >= 4.0 and all( + 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 + ) + filtered_groups.append([*extras, final_image]) - perfect_no_comment = True - if float(d_notes[label]) < 4.0: - perfect_no_comment = False - else: - lfb = result.get('feedback', []) - for e in lfb: - if "to_delete" not in e or not e["to_delete"]: - perfect_no_comment = False + concat_image = concatenate(concat_images) + with staged_files(output_dir) as staging: + 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") + logs.append(f" Saved regenerated files in {output_dir}") + status = ExitCode.PARTIAL if incomplete else ExitCode.SUCCESS + return status, "\n".join(logs) - if not perfect_no_comment or has_notes: - extras = get_extra_pdfs_as_images(root_dir, label, annotating, all_labels) - extras.append(final_img) - concat_list_F.append(extras) - # --- 3. Save Final Outputs --- - with open(score_path, "w") as f: - json.dump(d_notes, f, indent=4) - logs.append(f" Saved {score_path}") +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): + raise TypeError(f"Malformed refaire entry: {entry!r}") + copy_name, labels = entry + student_id = str(copy_name).removeprefix("Copie") + normalized_labels = [str(label) for label in labels] + entries.append([str(copy_name), normalized_labels]) + by_student[student_id] = normalized_labels + return entries, by_student - if concat_list: - max_w = max(i.width for i in concat_list) - total_h = sum(i.height for i in concat_list) - full_img = Image.new("RGB", (max_w, total_h), "white") - y = 0 - for img in concat_list: - full_img.paste(img, (0, y)) - y += img.height +def run( + workspace: EvaluationWorkspace, + *, + refaire: bool = False, + update_score: bool = False, +) -> ExitCode: + workspace.require_files("labels", "correction.json") + workspace.require_directories("Copies", "Par label", "BGnot") + refaire_list: RefaireList | None = None + refaire_by_student: dict[str, list[str]] = {} + if refaire: + workspace.require_files("refaire.json") + workspace.require_directories("BRnot") + refaire_list, refaire_by_student = _read_refaire(workspace) - full_img.save(os.path.join(output_dir, "Concat.jpg")) - logs.append(f" Saved regenerated Concat.jpg") + all_labels = utils.read_all_labels(workspace.root) + loaded = load_annotation_data(workspace, refaire_list=refaire_list) + for warning in loaded.warnings: + print(f"Warning: {warning}") + if not loaded.data: + print("No annotation data found.") + return ExitCode.PARTIAL - if concat_list_F: - pdf_out_path = os.path.join(output_dir, "Concat_F.pdf") - save_paginated_pdf(concat_list_F, pdf_out_path) - logs.append(f" Saved regenerated Concat_F.pdf") + actions_by_student: dict[str, list[dict[str, Any]]] = defaultdict(list) + notes_by_student: dict[str, LabelNotes] = defaultdict(dict) + 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") + ] + with concurrent.futures.ThreadPoolExecutor(max_workers=6) as executor: + futures = [ + executor.submit(_scan_annotation_directory, path, only_ids) + for path in group_dirs + ] + for future in concurrent.futures.as_completed(futures): + _merge_scan_result(actions_by_student, notes_by_student, future.result()) - return "\n".join(logs) + refaire_incomplete = False + 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 + continue + actions_by_student[student_id] = [ + action + for action in actions_by_student[student_id] + if str(action.get("label")) not in selected_set + ] + 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 + ) + 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 -from utils import read_all_labels -import argparse + status = ( + ExitCode.PARTIAL + if loaded.warnings or refaire_incomplete + else ExitCode.SUCCESS + ) + 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( + apply_actions_and_regenerate_grouped, + workspace, + loaded.data, + student_id, + actions_by_student[student_id], + notes_by_student[student_id], + all_labels, + update_score=update_score, + ): student_id + for student_id in student_ids + if student_id in loaded.data + } + for future in concurrent.futures.as_completed(futures): + result, output = future.result() + print(output) + if result != ExitCode.SUCCESS: + status = ExitCode.PARTIAL + return status +def build_parser() -> argparse.ArgumentParser: + parser = evaluation_parser("Read grouped annotations and regenerate copies") + parser.add_argument( + "--refaire", + action="store_true", + help="Use refaire.json and merge annotations from BRnot", + ) + parser.add_argument( + "--update-score", + action="store_true", + help="Override generated scores with values from existing score.json files", + ) + return parser + + +def main(argv: Sequence[str] | None = None) -> int: + parser = build_parser() + + def handle(args: argparse.Namespace) -> ExitCode: + return run( + workspace_from_args(args), + refaire=args.refaire, + update_score=args.update_score, + ) + + return execute(parser, argv, handle) + if __name__ == "__main__": - parser = argparse.ArgumentParser(description="Read grouped annotations and compile PDFs") - parser.add_argument("input_path", help="Directory path") - parser.add_argument("--refaire", action="store_true", help="Merge refaire annotations from Bnot") - parser.add_argument("--update-score", action="store_true", help="Override scores with values from existing score.json") - args = parser.parse_args() - - root_dir = sys.argv[1] - bgnot_dir = os.path.join(root_dir, "BGnot") - - if not os.path.exists(bgnot_dir): - print(f"Directory {bgnot_dir} does not exist. Run annotating_by_label.py first.") - sys.exit(1) - - try: - all_labels = read_all_labels(Path(root_dir)) - except FileNotFoundError: - all_labels = [] - - refaire_dict = {} - 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) - for c_name, labels in refaire_list: - sid = c_name.replace("Copie", "") - refaire_dict[sid] = labels - else: - print(f"Warning: --refaire flag used, but {refaire_path} not found.") - - - # Load original data - if args.refaire and refaire_list: - original_data = annotating.make_dictionary(root_dir, - refaire=True, - refaire_list=refaire_list) - else: - original_data = annotating.make_dictionary(root_dir) - - lock = threading.Lock() - actions_by_student = collections.defaultdict(list) - notes_by_student = collections.defaultdict(dict) - - - def process_bgnot_entry(entry, only_ids=None): - gdir = os.path.join(bgnot_dir, entry) - if not os.path.isdir(gdir) or entry.startswith("Copie"): - return - bnote_path = os.path.join(gdir, "bnote.json") - with open(bnote_path, "r") as f: - bnote_data = json.load(f) - - if only_ids: - id_found = False - for d in bnote_data["images"]: - if d["id"] in only_ids: - id_found = True - if not id_found: - return - - actions, notes_img = detect_checks_and_notes(gdir) - if not os.path.exists(bnote_path) or notes_img is None: - return - - - with lock: - for act in actions: - sid = str(act.get("student_id")) - if sid: actions_by_student[sid].append(act) - - for img_info in bnote_data.get("images", []): - sid, lbl = str(img_info.get("id")), img_info.get("label") - hmin, hmax = img_info.get("hmin", 0), img_info.get("hmax", 0) - if hmax > hmin: - crop = notes_img.crop((0, hmin, notes_img.width, hmax)) - if has_significant_notes(crop): - notes_by_student[sid][lbl] = {'img': crop, - 'old_header_h': img_info.get("header_height", 0)} - - - def process_refaire_entry(sid, r_labels): - s_bnot_dir = os.path.join(root_dir, "BRnot", f"Copie{sid}") - if not os.path.exists(s_bnot_dir): return - if not r_labels: - r_labels = list(original_data.get(sid, {}).keys()) - - with lock: - actions_by_student[sid] = [a for a in actions_by_student[sid] - if a.get('label') not in r_labels] - for lbl in r_labels: - notes_by_student[sid].pop(lbl, None) - - b_actions, b_notes_img = detect_checks_and_notes(s_bnot_dir) - b_bnote_path = os.path.join(s_bnot_dir, "bnote.json") - if os.path.exists(b_bnote_path): - with open(b_bnote_path, "r") as f: - b_bnote_data = json.load(f) - with lock: - for act in b_actions: - act["student_id"] = sid - actions_by_student[sid].append(act) - if b_notes_img: - for img_info in b_bnote_data.get("images", []): - lbl = img_info.get("label") - hmin, hmax = img_info.get("hmin", 0), img_info.get("hmax", 0) - if hmax > hmin: - crop = b_notes_img.crop((0, hmin, b_notes_img.width, hmax)) - if has_significant_notes(crop): - notes_by_student[sid][lbl] = \ - {'img': crop, - 'old_header_h': img_info.get("header_height", 0)} - - - - # --- 0. Read refaire.json if requested --- - - if refaire_dict: - only_ids = [ids for ids in refaire_dict] - else: - only_ids = None - - - # Lecture des bgnot - with concurrent.futures.ThreadPoolExecutor(max_workers=6) as executor: - executor.map(lambda x: process_bgnot_entry(x, only_ids=only_ids), - os.listdir(bgnot_dir)) - - # Refaire - if args.refaire and refaire_dict: - for sid, labels in refaire_dict.items(): - process_refaire_entry(sid, labels) - - - def process_student(sid): - if sid not in original_data: - return "" - return apply_actions_and_regenerate_grouped( - root_dir, - original_data, - sid, - actions_by_student[sid], - notes_by_student[sid], - all_labels, - update_score=args.update_score - ) - - # --- 2. Process each student concurrently using 4 threads --- - sids = sorted(original_data.keys(), key=natural_key) - with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor: - if refaire_dict: - futures = {executor.submit(process_student, sid): sid for sid in refaire_dict} - else: - futures = {executor.submit(process_student, sid): sid for sid in sids} - - for future in concurrent.futures.as_completed(futures): - output = future.result() - if output: - print(output) + raise SystemExit(main()) diff --git a/tests/test_gui_core.py b/tests/test_gui_core.py index 4eb5724..84943c9 100644 --- a/tests/test_gui_core.py +++ b/tests/test_gui_core.py @@ -25,8 +25,9 @@ from copienator import ( read_json, workspace_from_target, ) +from copienator.annotation_actions import apply_checkbox_actions, apply_score_overrides from copienator.annotation_data import AnnotationLoadResult, load_annotation_data -from copienator.filesystem import staged_directory +from copienator.filesystem import staged_directory, staged_files from copienator_gui.app import process_status from copienator_gui.diagnostics import collect_diagnostics from copienator_gui.runner import ProcessRunner @@ -237,6 +238,52 @@ class AnnotationDataTests(unittest.TestCase): leftovers = [path for path in destination.parent.iterdir() if path.name.startswith(".output.")] self.assertEqual(leftovers, []) + def test_staged_files_preserve_inputs_and_roll_back_outputs(self) -> None: + with tempfile.TemporaryDirectory() as directory: + destination = Path(directory) / "Copie01" + destination.mkdir() + (destination / "bnote.json").write_text("input", encoding="utf-8") + (destination / "score.json").write_text("old", encoding="utf-8") + + with self.assertRaises(RuntimeError), staged_files(destination) as staging: + (staging / "score.json").write_text("broken", encoding="utf-8") + (staging / "Concat.jpg").write_text("partial", encoding="utf-8") + raise RuntimeError("rendering failed") + self.assertEqual((destination / "score.json").read_text(), "old") + self.assertFalse((destination / "Concat.jpg").exists()) + + with staged_files(destination) as staging: + (staging / "score.json").write_text("new", encoding="utf-8") + (staging / "Concat.jpg").write_text("complete", encoding="utf-8") + self.assertEqual((destination / "score.json").read_text(), "new") + self.assertEqual((destination / "bnote.json").read_text(), "input") + + def test_annotation_actions_are_shared_and_do_not_touch_json_sources(self) -> None: + with tempfile.TemporaryDirectory() as directory: + score_path = Path(directory) / "score.json" + atomic_write_json(score_path, {"Ex 1": "3"}) + labels_data = { + "Ex 1": { + "result": { + "score": 1, + "feedback": [{"text": "global"}], + } + } + } + logs = [] + dirty = apply_checkbox_actions( + labels_data, + [{"label": "Ex 1", "type": "del_global", "index": 0}], + logs.append, + ) + dirty |= apply_score_overrides(labels_data, score_path, logs.append) + self.assertEqual(dirty, {"Ex 1"}) + self.assertTrue( + labels_data["Ex 1"]["result"]["feedback"][0]["to_delete"] + ) + self.assertEqual(labels_data["Ex 1"]["result"]["score"], "3") + self.assertEqual(read_json(score_path), {"Ex 1": "3"}) + class StandardCliTests(unittest.TestCase): @classmethod @@ -249,6 +296,12 @@ class StandardCliTests(unittest.TestCase): "annotating_by_label": load_script_module( "annotating_by_label.py", "annotating_by_label" ), + "reading_annotations": load_script_module( + "reading_annotations.py", "reading_annotations" + ), + "reading_grouped_annotations": load_script_module( + "reading_grouped_annotations.py", "reading_grouped_annotations" + ), "copies_tools": load_script_module( "copies_tools.py", "copienator_copies_tools_test" ), @@ -284,6 +337,8 @@ class StandardCliTests(unittest.TestCase): "annotating": [missing], "annotating_with_checks": [missing], "annotating_by_label": [missing], + "reading_annotations": [missing], + "reading_grouped_annotations": [missing], } for name, arguments in invocations.items(): with self.subTest(script=name), redirect_stderr(io.StringIO()): @@ -366,6 +421,16 @@ class StandardCliTests(unittest.TestCase): "grouped", {"target": evaluation, "overwrite": True}, ), + "reading_annotations": ( + "read_annotations", + "standard", + {"target": evaluation, "update_score": True}, + ), + "reading_grouped_annotations": ( + "read_annotations", + "grouped", + {"target": evaluation, "update_score": True, "refaire": True}, + ), } for module_name, (step_id, variant_id, values) in cases.items(): step = steps[step_id] @@ -590,6 +655,44 @@ class StandardCliTests(unittest.TestCase): with redirect_stderr(io.StringIO()): self.assertEqual(module.main([str(evaluation), "--refaire"]), 3) + def test_grouped_reader_refaire_requires_refaire_file(self) -> None: + module = self.modules["reading_grouped_annotations"] + with tempfile.TemporaryDirectory() as directory: + evaluation = Path(directory) / "Exam" + (evaluation / "Copies").mkdir(parents=True) + (evaluation / "Par label").mkdir() + (evaluation / "BGnot").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_note_detection_accepts_a_missing_note_layer(self) -> None: + module = self.modules["reading_annotations"] + self.assertFalse(module.has_significant_notes(None)) + + def test_grouped_reader_reports_worker_failures(self) -> None: + module = self.modules["reading_grouped_annotations"] + with tempfile.TemporaryDirectory() as directory: + evaluation = Path(directory) / "Exam" + (evaluation / "Copies").mkdir(parents=True) + (evaluation / "Par label").mkdir() + (evaluation / "BGnot" / "Ex 1").mkdir(parents=True) + (evaluation / "labels").write_text("Ex 1\n", encoding="utf-8") + atomic_write_json(evaluation / "correction.json", {}) + loaded = AnnotationLoadResult({"01": {"Ex 1": {}}}, []) + with ( + patch.object(module, "load_annotation_data", return_value=loaded), + patch.object( + module, + "_scan_annotation_directory", + side_effect=RuntimeError("worker failed"), + ), + redirect_stderr(io.StringIO()) as errors, + ): + self.assertEqual(module.main([str(evaluation)]), 1) + self.assertIn("worker failed", errors.getvalue()) + def test_checked_render_failure_preserves_previous_student_output(self) -> None: module = self.modules["annotating_with_checks"] with tempfile.TemporaryDirectory() as directory: