From 9b22a8a137c2108317e05cc85c9c537eca8babab Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Miquel?= Date: Tue, 15 Sep 2026 14:18:22 +0200 Subject: [PATCH] Miscs improvements (Interro02) --- copienator/commands/add_final_score.py | 127 +++++++++++------ copienator/commands/correction.py | 23 ++- copienator/commands/giving_names.py | 95 +++++++++++-- copienator/commands/post_correction.py | 58 ++++++-- copienator/commands/reading_annotations.py | 98 +++++++++++-- .../commands/reading_grouped_annotations.py | 65 ++++++++- copienator/configuration.py | 1 + copienator/return_answers.py | 73 ++++++++-- copienator/utils.py | 27 +++- copienator_gui/workflow.py | 53 ++++++- default_config.py | 2 + tests/test_add_final_score.py | 99 +++++++++++++ tests/test_answer_returns.py | 134 +++++++++++++++++- tests/test_gui_core.py | 108 +++++++++++++- tests/test_utils_compile.py | 48 +++++++ 15 files changed, 893 insertions(+), 118 deletions(-) create mode 100644 tests/test_add_final_score.py create mode 100644 tests/test_utils_compile.py diff --git a/copienator/commands/add_final_score.py b/copienator/commands/add_final_score.py index b090cef..769c041 100644 --- a/copienator/commands/add_final_score.py +++ b/copienator/commands/add_final_score.py @@ -7,11 +7,17 @@ from pathlib import Path import pandas as pd from PIL import Image, ImageDraw, ImageFont -from copienator.configuration import FINAL_SCORE_FONT_PATH, FINAL_SCORE_ODS_PATH, FINAL_SCORE_OUTPUT_DIR +from copienator.configuration import ( + FINAL_SCORE_FONT_PATH, + FINAL_SCORE_HISTOGRAM_PATH, + FINAL_SCORE_ODS_PATH, + FINAL_SCORE_OUTPUT_DIR, +) # Configuration constants ODS_PATH = Path(FINAL_SCORE_ODS_PATH).expanduser() OUTPUT_DIR = Path(FINAL_SCORE_OUTPUT_DIR).expanduser() +HISTOGRAM_PATH = Path(FINAL_SCORE_HISTOGRAM_PATH).expanduser() def score_font(size): @@ -33,6 +39,35 @@ def get_rounded_score(score): except (ValueError, TypeError): return None + +def copy_return_artifacts(source_dir: Path, destination_dir: Path) -> None: + """Copy the metadata and optional individual answers for one student.""" + for filename in ("score.json", "info.json"): + source = source_dir / filename + if source.is_file(): + shutil.copy2(source, destination_dir / filename) + else: + print(f"Warning: Missing '{source}'.") + + answers_source = source_dir / "answers" + answers_destination = destination_dir / "answers" + if answers_destination.is_symlink(): + answers_destination.unlink() + elif answers_destination.is_dir(): + shutil.rmtree(answers_destination) + if answers_source.is_dir(): + shutil.copytree(answers_source, answers_destination) + + +def copy_histogram(output_dir: Path) -> None: + """Copy the score histogram beside the per-student output folders.""" + if not HISTOGRAM_PATH.is_file(): + print(f"Warning: Missing histogram '{HISTOGRAM_PATH}'.") + return + destination = output_dir / "histogramme.pdf" + shutil.copy2(HISTOGRAM_PATH, destination) + print(f"Copied histogram: {destination}") + def process_images(base_dir, output_dir): # 1. Load Data try: @@ -56,60 +91,64 @@ def process_images(base_dir, output_dir): print(f"Error: Directory '{search_path}' not found.") sys.exit(1) - for img_path in sorted(search_path.glob("*/*.jpg")): - student_name = img_path.stem # Filename without extension + for student_source in sorted(path for path in search_path.iterdir() if path.is_dir()): + image_paths = sorted(student_source.glob("*.jpg")) + pdf_paths = sorted(student_source.glob("*.pdf")) + media_paths = image_paths or pdf_paths + if not media_paths: + print(f"Error: No JPG or PDF found in '{student_source}'.") + continue + + student_name = media_paths[0].stem + student_output = output_dir / student_name + student_output.mkdir(parents=True, exist_ok=True) + # Remove files produced by the former flat output layout when migrating + # an existing export directory. + for suffix in (".jpg", ".pdf"): + legacy_output = output_dir / f"{student_name}{suffix}" + if legacy_output.is_file() or legacy_output.is_symlink(): + legacy_output.unlink() + copy_return_artifacts(student_source, student_output) # 4. Find Score if student_name not in score_db: print(f"Error: Student '{student_name}' not found in ODS file.") - continue + else: + raw_score = score_db[student_name] + score = get_rounded_score(raw_score) - raw_score = score_db[student_name] - score = get_rounded_score(raw_score) + if score is None: + print(f"Error: Invalid score '{raw_score}' for '{student_name}'.") + else: + # 5. Process Images + for img_path in image_paths: + try: + with Image.open(img_path) as img: + img = img.convert("RGB") + draw = ImageDraw.Draw(img) + width, _height = img.size - if score is None: - print(f"Error: Invalid score '{raw_score}' for '{student_name}'.") - continue + font_size = int(width * 0.08) + font = score_font(font_size) + text = str(score) - # 5. Process Image - try: - with Image.open(img_path) as img: - img = img.convert("RGB") - draw = ImageDraw.Draw(img) - width, height = img.size + bbox = draw.textbbox((0, 0), text, font=font) + text_w = bbox[2] - bbox[0] - # Dynamic font size (15% of image height) - font_size = int(width * 0.08) + # 30px padding, top right. + x = width - text_w - 30 + y = 30 + draw.text((x, y), text, fill=(255, 0, 0), font=font) - font = score_font(font_size) + img.save(student_output / img_path.name) + print(f"Processed: {student_name} -> {score}") + except Exception as e: + print(f"Error processing image for '{student_name}': {e}") - text = str(score) + for pdf_path in pdf_paths: + shutil.copy2(pdf_path, student_output / pdf_path.name) - # Calculate text size and position (Top Right) - bbox = draw.textbbox((0, 0), text, font=font) - text_w = bbox[2] - bbox[0] - text_h = bbox[3] - bbox[1] - - # 30px padding - x = width - text_w - 30 - y = 30 - - # Draw Text (Red) - draw.text((x, y), text, fill=(255, 0, 0), font=font) - - # Save - save_path = output_dir / f"{student_name}.jpg" - img.save(save_path) - print(f"Processed: {student_name} -> {score}") - - except Exception as e: - print(f"Error processing image for '{student_name}': {e}") - - for pdf_path in sorted(search_path.glob("*/*.pdf")): - student_name = pdf_path.stem # Filename without extension - save_path = output_dir / f"{student_name}.pdf" - - shutil.copy(str(pdf_path), str(save_path)) + copy_histogram(output_dir) def main(argv=None): diff --git a/copienator/commands/correction.py b/copienator/commands/correction.py index 72f6e9c..42e1c7b 100644 --- a/copienator/commands/correction.py +++ b/copienator/commands/correction.py @@ -85,6 +85,8 @@ def flush_thread_log(tid=None): # --- Lock for thread-safe file writing --- io_lock = threading.Lock() pro_lock = threading.Lock() +group_index_lock = threading.Lock() +reserved_group_indices: dict[str, int] = {} pro_count = 0 flash_count = 0 pro_quota_exhausted = False @@ -155,6 +157,7 @@ def configure_runtime( completed_tasks = [] results = {label: [] for _file, label in tasks} thread_logs.clear() + reserved_group_indices.clear() pro_count = 0 flash_count = 0 pro_quota_exhausted = False @@ -294,6 +297,16 @@ def get_next_group_idx(label): if not existing: return 0 return max([int(f.stem.split("_")[1]) for f in existing]) + +def reserve_next_group_idx(label: str) -> int: + """Reserve a unique zero-based group index for this correction run.""" + with group_index_lock: + if label not in reserved_group_indices: + reserved_group_indices[label] = get_next_group_idx(label) + idx = reserved_group_indices[label] + reserved_group_indices[label] = idx + 1 + return idx + def handle_label_errors(pid, label, res, pdf_path): """Handles Gemini labeling errors, moves/copies files, and returns new tasks.""" new_tasks = [] @@ -334,7 +347,7 @@ def handle_label_errors(pid, label, res, pdf_path): if pdf_path != old_pdf_path: shutil.move(str(pdf_path), str(old_pdf_path)) - idx = get_next_group_idx(new_label) + idx = reserve_next_group_idx(new_label) height = grouping.get_pdf_height(str(new_pdf_path)) grouping.create_jpg(new_label, idx, [(pid, str(new_pdf_path), height)], GROUPS_DIR) tprint(f"\t\tMaking {new_label} group {idx+1}") @@ -366,7 +379,7 @@ def handle_label_errors(pid, label, res, pdf_path): if not base_add_pdf_path.exists() and not add_pdf_path.exists(): shutil.copy(str(pdf_path), str(add_pdf_path)) tprint(f"\t\tCopying Copie{pid} : {label} -> {add_label}") - idx = get_next_group_idx(add_label) + idx = reserve_next_group_idx(add_label) tprint(f"\t\tMaking {add_label} group {idx+1}") height = grouping.get_pdf_height(str(add_pdf_path)) grouping.create_jpg(add_label, idx, [(pid, str(add_pdf_path), height)], GROUPS_DIR) @@ -584,7 +597,7 @@ def resolve_delayed_moves(): if pdf_path != old_pdf_path: shutil.move(str(pdf_path), str(old_pdf_path)) - idx = get_next_group_idx(target_label) + idx = reserve_next_group_idx(target_label) height = grouping.get_pdf_height(str(new_pdf_path)) grouping.create_jpg(target_label, idx, [(pid, str(new_pdf_path), height)], GROUPS_DIR) new_tasks.append((str(GROUPS_DIR / target_label / f"Group_{idx+1}.jpg"), target_label, False)) @@ -602,7 +615,7 @@ def resolve_delayed_moves(): resolved_any = True shutil.copy(str(pdf_path), str(add_pdf_path)) - idx = get_next_group_idx(target_label) + idx = reserve_next_group_idx(target_label) height = grouping.get_pdf_height(str(add_pdf_path)) grouping.create_jpg(target_label, idx, [(pid, str(add_pdf_path), height)], GROUPS_DIR) new_tasks.append((str(GROUPS_DIR / target_label / f"Group_{idx+1}.jpg"), target_label, False)) @@ -688,7 +701,7 @@ def run_configured(args: argparse.Namespace) -> ExitCode: # pdf_path = copie_dir / f"{label}_old.pdf" if pdf_path.exists(): - idx = get_next_group_idx(label) + idx = reserve_next_group_idx(label) height = grouping.get_pdf_height(str(pdf_path)) grouping.create_jpg(label, idx, [(pid, str(pdf_path), height)], GROUPS_DIR) new_group_path = str(GROUPS_DIR / label / f"Group_{idx+1}.jpg") diff --git a/copienator/commands/giving_names.py b/copienator/commands/giving_names.py index 6211134..6b919cf 100644 --- a/copienator/commands/giving_names.py +++ b/copienator/commands/giving_names.py @@ -29,9 +29,20 @@ def build_parser() -> argparse.ArgumentParser: choices=ANNOTATION_CHOICES, help="Annotation directory to use", ) + parser.add_argument( + "--update", + action="store_true", + help=( + "Update only the individual images in existing A Rendre/answers " + "directories, matching folders by their trailing copy ID" + ), + ) return parser +RETURN_COPY_ID = re.compile(r"\((\d+)\)$") + + def _read_expected_names(workspace: EvaluationWorkspace) -> set[str]: names_path = workspace.names_file() if not names_path.exists(): @@ -47,6 +58,70 @@ def _read_expected_names(workspace: EvaluationWorkspace) -> set[str]: } +def _annotation_source( + workspace: EvaluationWorkspace, + annotation_dir_name: str, + copy_id: str, +) -> Path | None: + selected = workspace.root / annotation_dir_name / f"Copie{copy_id}" + fallback = workspace.annotation_dir("simple") / f"Copie{copy_id}" + for candidate in (selected, fallback): + if (candidate / "score.json").is_file() and ( + (candidate / "Concat.jpg").is_file() + or (candidate / "info.json").is_file() + ): + return candidate + return None + + +def update_named_return_answers( + workspace: EvaluationWorkspace, + annotation_dir_name: str, +) -> ExitCode: + """Refresh only answers/ in existing returns, preserving manual names.""" + workspace.require_directories(annotation_dir_name, "A Rendre") + had_errors = False + found = False + for destination in sorted(workspace.return_dir.iterdir()): + if not destination.is_dir(): + continue + match = RETURN_COPY_ID.search(destination.name) + if match is None: + print( + f"Warning: cannot identify a copy ID in {destination.name!r}; skipped", + file=sys.stderr, + ) + had_errors = True + continue + found = True + copy_id = match.group(1) + source_folder = _annotation_source( + workspace, annotation_dir_name, copy_id + ) + if source_folder is None: + print( + f"Warning: no annotation source found for Copie{copy_id}; skipped", + file=sys.stderr, + ) + had_errors = True + continue + try: + publish_answer_returns( + workspace.root, + source_folder, + destination, + answers_only=True, + ) + print(f"Updated answers for {destination.name} from Copie{copy_id}") + except (OSError, TypeError, ValueError) as exc: + print(f"Error updating answers for {destination.name}: {exc}", file=sys.stderr) + had_errors = True + if not found: + print("Warning: no identifiable student folders found in A Rendre", file=sys.stderr) + had_errors = True + return ExitCode.PARTIAL if had_errors else ExitCode.SUCCESS + + def prepare_named_returns( workspace: EvaluationWorkspace, annotation_dir_name: str, @@ -74,9 +149,6 @@ def prepare_named_returns( had_errors = True assigned_names: set[str] = set() - selected_annotations = workspace.root / annotation_dir_name - fallback_annotations = workspace.annotation_dir("simple") - for name, copy_ids in copies_map.items(): if name == "Unknown": print( @@ -92,16 +164,9 @@ def prepare_named_returns( safe_name = safe_filename(name) for copy_id in copy_ids: - selected = selected_annotations / f"Copie{copy_id}" - fallback = fallback_annotations / f"Copie{copy_id}" - source_folder = None - for candidate in (selected, fallback): - if (candidate / "score.json").is_file() and ( - (candidate / "Concat.jpg").is_file() - or (candidate / "info.json").is_file() - ): - source_folder = candidate - break + source_folder = _annotation_source( + workspace, annotation_dir_name, copy_id + ) if source_folder is None: continue @@ -160,7 +225,10 @@ def run( workspace: EvaluationWorkspace, *, annotation_dir: str, + update: bool = False, ) -> ExitCode: + if update: + return update_named_return_answers(workspace, annotation_dir) return prepare_named_returns(workspace, annotation_dir) @@ -172,6 +240,7 @@ def main(argv: Sequence[str] | None = None) -> int: lambda args: run( workspace_from_args(args, repository=Path.cwd()), annotation_dir=args.annotation_dir, + update=args.update, ), ) diff --git a/copienator/commands/post_correction.py b/copienator/commands/post_correction.py index f070a0e..7e24a52 100644 --- a/copienator/commands/post_correction.py +++ b/copienator/commands/post_correction.py @@ -19,6 +19,10 @@ from copienator import ( WORD_LIST_FILE = Path(__file__).resolve().parents[1] / "data" / "liste_francais.txt" ACCENT_PATTERN = re.compile(r"[éèêëàâäîïôöùûüçœÉÈÊËÀÂÄÎÏÔÖÙÛÜÇŒ]") +MATH_PATTERN = re.compile( + r"(\$\$.*?\$\$|\$.*?\$|\\\(.*?\\\)|\\\[.*?\\\])", + re.DOTALL, +) def build_parser() -> argparse.ArgumentParser: @@ -26,22 +30,37 @@ def build_parser() -> argparse.ArgumentParser: def escape_latex_underscores(text: str) -> str: - r"""Escape underscores outside LaTeX math environments.""" - math_pattern = re.compile( - r"(\$\$.*?\$\$|\$.*?\$|\\\(.*?\\\)|\\\[.*?\\\])", - re.DOTALL, - ) + r"""Escape underscores outside math without double-escaping existing ones.""" + + def escape_plain(value: str) -> str: + # Collapse any existing escape run as well, making cleanup idempotent. + return re.sub(r"\\*_", lambda _match: r"\_", value) + parts: list[str] = [] last_end = 0 - for match in math_pattern.finditer(text): + for match in MATH_PATTERN.finditer(text): start, end = match.span() - parts.append(text[last_end:start].replace("_", r"\_")) + parts.append(escape_plain(text[last_end:start])) parts.append(match.group(0)) last_end = end - parts.append(text[last_end:].replace("_", r"\_")) + parts.append(escape_plain(text[last_end:])) return "".join(parts) +def normalize_overescaped_latex_commands(text: str) -> str: + r"""Collapse doubled command escapes inside LaTeX math environments. + + Model responses occasionally contain ``\\mathbb`` after JSON decoding where + LaTeX requires ``\mathbb``. A doubled backslash followed by whitespace is a + legitimate row break (for example in ``cases``), so it must be preserved. + """ + + def normalize_math(match: re.Match[str]) -> str: + return re.sub(r"\\\\(?=[A-Za-z{}])", r"\\", match.group(0)) + + return MATH_PATTERN.sub(normalize_math, text) + + def build_lookup_map(word_list_path: Path = WORD_LIST_FILE) -> dict[str, str]: words = word_list_path.read_text(encoding="utf-8").splitlines() lookup: dict[str, str] = {} @@ -68,8 +87,23 @@ def fix_hex_corruption_safe(text: str) -> str: ) -def some_other_replacements(text: str) -> str: - return text.replace("\neq", "\\neq").replace("\not", "\\not") +def repair_json_escape_corruption(text: str) -> str: + r"""Restore observed LaTeX commands consumed as JSON control escapes.""" + replacements = ( + ("\x0crac", r"\frac"), + ("\x0ceuille", r"\equiv"), + ("\theta", r"\theta"), + ("\times", r"\times"), + ("\textbackslash ", "\\"), + ("\negthinspace", r"\negthinspace"), + ("\neq", r"\neq"), + ("\not", r"\not"), + ("∈", r"\ensuremath{\in}"), + ("⊂", r"\ensuremath{\subset}"), + ) + for broken, repaired in replacements: + text = text.replace(broken, repaired) + return text def clean_string(text: str, lookup: dict[str, str]) -> str: @@ -80,7 +114,9 @@ def clean_string(text: str, lookup: dict[str, str]) -> str: text = re.sub(r" \x00{1,2} ", " à ", text) if "\x00" in text: text = fast_fix(text, lookup).replace("\x00", "") - return escape_latex_underscores(some_other_replacements(text)) + text = repair_json_escape_corruption(text) + text = normalize_overescaped_latex_commands(text) + return escape_latex_underscores(text) def clean_obj(value: Any, lookup: dict[str, str]) -> Any: diff --git a/copienator/commands/reading_annotations.py b/copienator/commands/reading_annotations.py index 1e35f6e..92898ad 100644 --- a/copienator/commands/reading_annotations.py +++ b/copienator/commands/reading_annotations.py @@ -10,7 +10,7 @@ from pdf2image import convert_from_path from PIL import Image, ImageChops, ImageDraw, ImageFilter from copienator.commands import annotating -from copienator import utils +from copienator import configuration, utils from copienator import ( EvaluationWorkspace, ExitCode, @@ -24,6 +24,7 @@ from copienator.annotation_actions import apply_checkbox_actions, apply_score_ov from copienator.answer_info import build_answer_info from copienator.annotation_data import AnnotationData, load_annotation_data from copienator.filesystem import staged_files +from copienator.return_answers import save_return_answer_options Image.MAX_IMAGE_PIXELS = None @@ -69,10 +70,11 @@ def detect_checks_and_notes( print(f" Resizing annotated PDF from {user_image.size} to {reference.size}") user_image = user_image.resize(reference.size, Image.Resampling.LANCZOS) - 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 the full-size difference in uint8. Converting both tall group images + # to the platform ``int`` dtype used several gigabytes per scan worker. + difference = np.asarray( + ImageChops.difference(reference, user_image), dtype=np.uint8 + ) keep_mask = Image.new("L", reference.size, 255) mask_draw = ImageDraw.Draw(keep_mask) actions: list[dict[str, Any]] = [] @@ -83,10 +85,14 @@ def detect_checks_and_notes( x1, y1, x2, y2 = map(int, raw_box["global_box"]) x1, y1 = max(0, x1), max(0, y1) x2, y2 = min(reference.width, x2), min(reference.height, y2) - region = difference_gray[y1 + 5 : y2 - 5, x1 + 5 : x2 - 5] + region = difference[y1 + 5 : y2 - 5, x1 + 5 : x2 - 5] if region.size == 0: continue - density = np.sum(region > 30) / region.size + # Preserve the previous mean-across-RGB threshold, but allocate its + # temporary float array only for the small checkbox region. + density = np.count_nonzero(np.mean(region, axis=2) > 30) / ( + region.shape[0] * region.shape[1] + ) if density > 0.05: actions.append(raw_box) mask_draw.rectangle([x1 - 15, y1 - 15, x2 + 15, y2 + 15], fill=0) @@ -95,13 +101,17 @@ def detect_checks_and_notes( 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) + del difference 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)) + del reference_blur, user_blur + alpha = np.asarray(diff_image, dtype=np.uint8).copy() + np.greater(alpha, 50, out=alpha) + alpha *= np.uint8(255) + np.minimum(alpha, np.asarray(keep_mask, dtype=np.uint8), out=alpha) notes = user_image.convert("RGBA") - notes.putalpha(Image.fromarray(final_alpha)) + notes.putalpha(Image.fromarray(alpha)) return actions, notes @@ -150,8 +160,10 @@ def apply_actions_and_regenerate( labels_data = data[student_id] apply_checkbox_actions(labels_data, actions, print) + score_path = output_dir / "score.json" + preserve_score_file = update_score and score_path.is_file() if update_score: - apply_score_overrides(labels_data, output_dir / "score.json", print) + apply_score_overrides(labels_data, score_path, print) scores = dict.fromkeys(all_labels, "") answer_labels: list[str] = [] @@ -220,7 +232,8 @@ def apply_actions_and_regenerate( "touched.json", "answer_labels.json")) as staging: for label, image in dirty_images.items(): image.save(staging / f"{label}.jpg") - atomic_write_json(staging / "score.json", scores) + if not preserve_score_file: + atomic_write_json(staging / "score.json", scores) atomic_write_json(staging / "info.json", build_answer_info( scores, labels_data, answer_labels )) @@ -229,13 +242,41 @@ def apply_actions_and_regenerate( if filtered_image is not None: filtered_image.save(staging / "Concat_F.jpg") + if preserve_score_file: + print(f" Preserved existing score.json in {output_dir}") print(f" Saved regenerated files in {output_dir}") return ExitCode.PARTIAL if incomplete else ExitCode.SUCCESS -def run(workspace: EvaluationWorkspace, *, update_score: bool = False) -> ExitCode: +def run( + workspace: EvaluationWorkspace, + *, + update_score: bool = False, + return_answers_context: bool | None = None, + return_answers_question: bool | None = None, + return_answers_solution: bool | None = None, +) -> ExitCode: workspace.require_files("labels", "correction.json") workspace.require_directories("Copies", "Par label", "Bnot") + if configuration.RETURN_ANSWERS_ENABLED: + save_return_answer_options( + workspace.root, + context=( + configuration.RETURN_ANSWERS_CONTEXT + if return_answers_context is None + else return_answers_context + ), + question=( + configuration.RETURN_ANSWERS_QUESTION + if return_answers_question is None + else return_answers_question + ), + solution=( + configuration.RETURN_ANSWERS_SOLUTION + if return_answers_solution is None + else return_answers_solution + ), + ) all_labels = utils.read_all_labels(workspace.root) loaded = load_annotation_data(workspace) for warning in loaded.warnings: @@ -276,7 +317,28 @@ def build_parser() -> argparse.ArgumentParser: parser.add_argument( "--update-score", action="store_true", - help="Override generated scores with values from existing score.json files", + help=( + "Regenerate images with current statement/solution PDFs while " + "preserving and applying existing score.json values" + ), + ) + parser.add_argument( + "--return-answers-context", + action=argparse.BooleanOptionalAction, + default=configuration.RETURN_ANSWERS_CONTEXT, + help="Include applicable context pages in individual answer exports", + ) + parser.add_argument( + "--return-answers-question", + action=argparse.BooleanOptionalAction, + default=configuration.RETURN_ANSWERS_QUESTION, + help="Include the current question PDF in individual answer exports", + ) + parser.add_argument( + "--return-answers-solution", + action=argparse.BooleanOptionalAction, + default=configuration.RETURN_ANSWERS_SOLUTION, + help="Include the current solution PDF in individual answer exports", ) return parser @@ -285,7 +347,13 @@ 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 run( + workspace_from_args(args), + update_score=args.update_score, + return_answers_context=args.return_answers_context, + return_answers_question=args.return_answers_question, + return_answers_solution=args.return_answers_solution, + ) return execute(parser, argv, handle) diff --git a/copienator/commands/reading_grouped_annotations.py b/copienator/commands/reading_grouped_annotations.py index c4f688c..141b9aa 100644 --- a/copienator/commands/reading_grouped_annotations.py +++ b/copienator/commands/reading_grouped_annotations.py @@ -9,6 +9,7 @@ from typing import Any from PIL import Image, ImageDraw +from copienator import configuration from copienator import ( EvaluationWorkspace, ExitCode, @@ -29,9 +30,11 @@ from copienator.commands.reading_annotations import ( has_significant_notes, ) from copienator.filesystem import staged_files +from copienator.return_answers import save_return_answer_options LabelNotes = dict[str, dict[str, Any]] ScanResult = tuple[dict[str, list[dict[str, Any]]], dict[str, LabelNotes]] +SCAN_WORKERS = 2 def get_extra_pdfs_as_images( @@ -191,9 +194,11 @@ def apply_actions_and_regenerate_grouped( output_dir = workspace.root / annotation_dir / f"Copie{student_id}" labels_data = data.get(student_id, {}) apply_checkbox_actions(labels_data, actions, logs.append) + score_path = output_dir / "score.json" + preserve_score_file = update_score and score_path.is_file() if update_score: apply_score_overrides( - labels_data, output_dir / "score.json", logs.append + labels_data, score_path, logs.append ) selected_labels = selected_labels if selected_labels is not None else set() @@ -350,7 +355,8 @@ def apply_actions_and_regenerate_grouped( atomic_write_json(staging / "refaire_simple_layout.json", simple_layout) for label, image in dirty_images.items(): image.save(staging / f"{label}.jpg") - atomic_write_json(staging / "score.json", scores) + if not preserve_score_file: + atomic_write_json(staging / "score.json", scores) atomic_write_json(staging / "info.json", build_answer_info( scores, labels_data, answer_labels, touched )) @@ -364,6 +370,8 @@ def apply_actions_and_regenerate_grouped( [image for group in filtered_groups for image in group] ) filtered_image.save(staging / "Concat_F.jpg") + if preserve_score_file: + logs.append(f" Preserved existing score.json in {output_dir}") logs.append(f" Saved regenerated files in {output_dir}") status = ExitCode.PARTIAL if incomplete else ExitCode.SUCCESS return status, "\n".join(logs) @@ -461,6 +469,9 @@ def run( refaire: bool = False, update_score: bool = False, annotation_dir: str = "BGnot", + return_answers_context: bool | None = None, + return_answers_question: bool | None = None, + return_answers_solution: bool | None = None, ) -> ExitCode: workspace.require_files("labels", "correction.json") workspace.require_directories("Copies", "Par label", annotation_dir) @@ -470,6 +481,25 @@ def run( workspace.require_files("refaire.json") workspace.require_directories("BRnot") refaire_list, refaire_by_student = _read_refaire(workspace) + if configuration.RETURN_ANSWERS_ENABLED: + save_return_answer_options( + workspace.root, + context=( + configuration.RETURN_ANSWERS_CONTEXT + if return_answers_context is None + else return_answers_context + ), + question=( + configuration.RETURN_ANSWERS_QUESTION + if return_answers_question is None + else return_answers_question + ), + solution=( + configuration.RETURN_ANSWERS_SOLUTION + if return_answers_solution is None + else return_answers_solution + ), + ) all_labels = utils.read_all_labels(workspace.root) loaded = load_annotation_data(workspace) @@ -494,7 +524,10 @@ def run( and path.is_dir() and not path.name.startswith("Copie") ] - with concurrent.futures.ThreadPoolExecutor(max_workers=6) as executor: + # Each worker decodes a full-height returned group and its reference image. + # Keep this stage deliberately narrow; answer regeneration below has its own + # parallel executor and a much smaller per-task memory footprint. + with concurrent.futures.ThreadPoolExecutor(max_workers=SCAN_WORKERS) as executor: futures = [ executor.submit(_scan_annotation_directory, path, only_ids) for path in group_dirs @@ -599,7 +632,28 @@ def build_parser() -> argparse.ArgumentParser: parser.add_argument( "--update-score", action="store_true", - help="Override generated scores with values from existing score.json files", + help=( + "Regenerate images with current statement/solution PDFs while " + "preserving and applying existing score.json values" + ), + ) + parser.add_argument( + "--return-answers-context", + action=argparse.BooleanOptionalAction, + default=configuration.RETURN_ANSWERS_CONTEXT, + help="Include applicable context pages in individual answer exports", + ) + parser.add_argument( + "--return-answers-question", + action=argparse.BooleanOptionalAction, + default=configuration.RETURN_ANSWERS_QUESTION, + help="Include the current question PDF in individual answer exports", + ) + parser.add_argument( + "--return-answers-solution", + action=argparse.BooleanOptionalAction, + default=configuration.RETURN_ANSWERS_SOLUTION, + help="Include the current solution PDF in individual answer exports", ) return parser @@ -615,6 +669,9 @@ def main(argv: Sequence[str] | None = None) -> int: refaire=args.refaire, update_score=args.update_score, annotation_dir=args.annotation_dir, + return_answers_context=args.return_answers_context, + return_answers_question=args.return_answers_question, + return_answers_solution=args.return_answers_solution, ) return execute(parser, argv, handle) diff --git a/copienator/configuration.py b/copienator/configuration.py index 0ea6731..1a6554d 100644 --- a/copienator/configuration.py +++ b/copienator/configuration.py @@ -39,6 +39,7 @@ RETURN_ANSWERS_ENABLED = False RETURN_ANSWERS_CONTEXT = False RETURN_ANSWERS_QUESTION = True RETURN_ANSWERS_SOLUTION = False +FINAL_SCORE_HISTOGRAM_PATH = Path("histogramme.pdf") for _name in dir(_configuration): if not _name.startswith("_"): globals()[_name] = getattr(_configuration, _name) diff --git a/copienator/return_answers.py b/copienator/return_answers.py index 89af872..f06bd4f 100644 --- a/copienator/return_answers.py +++ b/copienator/return_answers.py @@ -8,9 +8,56 @@ from copienator import atomic_write_json, configuration, read_json, utils from copienator.filesystem import staged_directory from copienator.platform import safe_filename +RETURN_ANSWER_OPTIONS_FILE = Path(".copienator") / "return_answers.json" -def publish_answer_returns(root: Path, source: Path, destination: Path) -> None: - """Publish individual reviewed answers and per-question information.""" + +def configured_return_answer_options() -> dict[str, bool]: + return { + "context": bool(configuration.RETURN_ANSWERS_CONTEXT), + "question": bool(configuration.RETURN_ANSWERS_QUESTION), + "solution": bool(configuration.RETURN_ANSWERS_SOLUTION), + } + + +def save_return_answer_options( + root: Path, + *, + context: bool, + question: bool, + solution: bool, +) -> None: + path = Path(root) / RETURN_ANSWER_OPTIONS_FILE + path.parent.mkdir(parents=True, exist_ok=True) + atomic_write_json( + path, + {"context": context, "question": question, "solution": solution}, + ) + + +def load_return_answer_options(root: Path) -> dict[str, bool]: + options = configured_return_answer_options() + path = Path(root) / RETURN_ANSWER_OPTIONS_FILE + if not path.is_file(): + return options + loaded = read_json(path) + if not isinstance(loaded, dict): + raise ValueError(f"Expected a return-answer options object in {path}") + for name in options: + if name in loaded: + if type(loaded[name]) is not bool: + raise ValueError(f"Expected a boolean for {name!r} in {path}") + options[name] = loaded[name] + return options + + +def publish_answer_returns( + root: Path, + source: Path, + destination: Path, + *, + answers_only: bool = False, +) -> None: + """Publish reviewed answers, optionally without touching return metadata.""" scores = read_json(source / "score.json") if not isinstance(scores, dict): raise ValueError(f"Expected a score object in {source}") @@ -36,6 +83,7 @@ def publish_answer_returns(root: Path, source: Path, destination: Path) -> None: if answers_dir.is_symlink(): raise ValueError(f"Expected a real answer directory: {answers_dir}") if configuration.RETURN_ANSWERS_ENABLED: + options = load_return_answer_options(root) labels = [label for label, entry in info.items() if entry["present"] and entry["not_empty"]] # Import the rendering backend only when individual images are requested. @@ -44,13 +92,13 @@ def publish_answer_returns(root: Path, source: Path, destination: Path) -> None: all_labels = utils.read_all_labels(root) with staged_directory(answers_dir) as staging: - for index, label in enumerate(sorted(labels, key=utils.natural_key), 1): + for label in sorted(labels, key=utils.natural_key): paths = [] - if configuration.RETURN_ANSWERS_CONTEXT: + if options["context"]: paths.extend(utils.pdf_images_of_contexts(root, label, all_labels)) - if configuration.RETURN_ANSWERS_QUESTION: + if options["question"]: paths.append(utils.pdf_image_of_enonce(root, label)) - if configuration.RETURN_ANSWERS_SOLUTION: + if options["solution"]: paths.append(utils.pdf_image_of_solution(root, label)) images = [] for path in paths: @@ -62,11 +110,16 @@ def publish_answer_returns(root: Path, source: Path, destination: Path) -> None: with Image.open(source / f"{label}.jpg") as answer: images.append(answer.convert("RGB")) image = concatenate(images) - # Numbering avoids collisions between sanitized label filenames. - image.save(staging / f"{index:03d} - {safe_filename(label)}.jpg") + output = staging / f"{safe_filename(label)}.jpg" + if output.exists(): + raise ValueError( + f"Answer labels produce the same filename in {answers_dir}: {label}" + ) + image.save(output) elif answers_dir.exists(): # Replace the managed directory with an empty one to remove stale exports. with staged_directory(answers_dir): pass - atomic_write_json(destination / "info.json", info) - (destination / "touched.json").unlink(missing_ok=True) + if not answers_only: + atomic_write_json(destination / "info.json", info) + (destination / "touched.json").unlink(missing_ok=True) diff --git a/copienator/utils.py b/copienator/utils.py index f99a4cb..74acd59 100644 --- a/copienator/utils.py +++ b/copienator/utils.py @@ -160,23 +160,38 @@ def compile_to_pdf(text, output_pdf_path): # env['TEXINPUTS'] = f".:{current_dir}:" try: - subprocess.run( + result = subprocess.run( ['pdflatex', '-interaction=nonstopmode', tex_filename], cwd=temp_dir, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, check=False ) if "minted" in text: - subprocess.run( + result = subprocess.run( ['pdflatex', '-interaction=nonstopmode', tex_filename], cwd=temp_dir, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, check=False) + if result.returncode != 0: + error_lines = [ + line.strip() for line in result.stdout.splitlines() + if line.lstrip().startswith("!") + ] + detail = f": {error_lines[0]}" if error_lines else "" + print( + f"Warning: LaTeX compilation failed for {output_pdf_path} " + f"(exit code {result.returncode}){detail}" + ) + generated_pdf = os.path.join(temp_dir, pdf_filename) if os.path.exists(generated_pdf): shutil.move(generated_pdf, output_pdf_path) + else: + print(f"Warning: LaTeX compilation produced no PDF for {output_pdf_path}") except Exception as e: print(f"Compilation error for {output_pdf_path}: {e}") diff --git a/copienator_gui/workflow.py b/copienator_gui/workflow.py index d5a9378..6a19e74 100644 --- a/copienator_gui/workflow.py +++ b/copienator_gui/workflow.py @@ -7,6 +7,7 @@ from collections.abc import Iterable from dataclasses import dataclass from pathlib import Path +from copienator import configuration from copienator.configuration import ALWAYS_CROP EVALUATION = "${evaluation}" @@ -18,6 +19,7 @@ class ArgumentSpec: label: str kind: str = "text" # text, int, bool, choice, path flag: str | None = None + false_flag: str | None = None default: object = "" choices: tuple[str, ...] = () help: str = "" @@ -87,6 +89,38 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]: ident, label, None, "manual" ) + return_answer_arguments = () + if configuration.RETURN_ANSWERS_ENABLED: + return_answer_arguments = ( + ArgumentSpec( + "return_answers_context", + "Inclure le contexte dans answers", + "bool", + "--return-answers-context", + "--no-return-answers-context", + default=configuration.RETURN_ANSWERS_CONTEXT, + help="Inclut les pages de contexte applicables avant chaque réponse individuelle publiée dans answers.", + ), + ArgumentSpec( + "return_answers_question", + "Inclure l’énoncé dans answers", + "bool", + "--return-answers-question", + "--no-return-answers-question", + default=configuration.RETURN_ANSWERS_QUESTION, + help="Inclut l’énoncé actuel avant chaque réponse individuelle publiée dans answers.", + ), + ArgumentSpec( + "return_answers_solution", + "Inclure la correction dans answers", + "bool", + "--return-answers-solution", + "--no-return-answers-solution", + default=configuration.RETURN_ANSWERS_SOLUTION, + help="Inclut la correction actuelle avant chaque réponse individuelle publiée dans answers.", + ), + ) + steps = [ StepDefinition( "inputs", @@ -533,10 +567,10 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]: arg_target("le dossier de l’évaluation"), ArgumentSpec( "update_score", - "Réappliquer les score.json", + "Régénérer avec les nouveaux énoncés/corrigés et les score.json", "bool", "--update-score", - help="Réutilise les valeurs présentes dans les fichiers score.json pour remplacer les scores lus dans les annotations.", + help="Régénère les images avec les PDF actuels d’énoncé et de correction ; pour chaque label, le score.json existant prévaut sur le score relu dans l’annotation manuscrite.", ), ArgumentSpec( "refaire", @@ -556,7 +590,7 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]: help="Indique le dossier d’annotations du passage principal dans lequel intégrer les questions refaites.", variants=("grouped",), ), - ), + ) + return_answer_arguments, ), StepDefinition( "giving_names", @@ -575,6 +609,13 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]: help="Choisissez le dossier d’annotations utilisé pour construire les fichiers nommés dans A Rendre.", positional=True, ), + ArgumentSpec( + "update", + "Mettre à jour uniquement les images de answers", + "bool", + "--update", + help="Met à jour uniquement les images du dossier answers de chaque élève existant, en identifiant la copie par le numéro final entre parenthèses et sans modifier le nom du dossier ni les autres fichiers.", + ), ), artifacts=("A Rendre",), ), @@ -632,7 +673,9 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]: "final_score", "Étapes personnelles", "Ajouter le score final", - "Génère les fichiers de diffusion avec le score final.", + "Prérequis immédiat : gestion_classe wse doit avoir été exécuté juste avant. " + "Génère ensuite les dossiers de diffusion avec le score final, puis copie " + "gestion_classe/Staging/histogramme.pdf dans le dossier Server/copies de l’évaluation.", (python("default", "Score final", "add-final-score", supports_verbose=False),), arguments=(arg_target("le dossier de l’évaluation"),), personal=True, @@ -782,6 +825,8 @@ def build_command( if spec.kind == "bool": if bool(value) and spec.flag: options.append(spec.flag) + elif not bool(value) and spec.false_flag: + options.append(spec.false_flag) continue if value is None or str(value).strip() == "": continue diff --git a/default_config.py b/default_config.py index 5248be6..b1f1f26 100644 --- a/default_config.py +++ b/default_config.py @@ -24,6 +24,7 @@ ALWAYS_CROP = False CURRENT_SCORE_ODS_PATH = Path("current_eval.ods") FINAL_SCORE_ODS_PATH = Path("simple_eval.ods") FINAL_SCORE_OUTPUT_DIR = Path("Server") / "copies" +FINAL_SCORE_HISTOGRAM_PATH = Path("histogramme.pdf") FINAL_SCORE_FONT_PATH = None # Modèle pour des choses très légères @@ -82,6 +83,7 @@ LATEX_BEFORE = r"""\documentclass[varwidth=24.8cm,margin=0.4cm]{standalone} \usepackage{minted} \usepackage{graphicx} \usepackage{enumitem} +\usepackage{multicol} \begin{document} \begin{minipage}{24.8cm} """ diff --git a/tests/test_add_final_score.py b/tests/test_add_final_score.py new file mode 100644 index 0000000..59fe8a3 --- /dev/null +++ b/tests/test_add_final_score.py @@ -0,0 +1,99 @@ +import contextlib +import io +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +import pandas as pd +from PIL import Image + +from copienator.commands import add_final_score + + +class AddFinalScoreTests(unittest.TestCase): + def test_creates_complete_student_directory(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "A Rendre" / "Student (01)" + answers = source / "answers" + answers.mkdir(parents=True) + Image.new("RGB", (300, 200), "white").save(source / "Student.jpg") + (source / "Student.pdf").write_bytes(b"pdf") + (source / "score.json").write_bytes(b'{"Ex 1": "4"}') + (source / "info.json").write_bytes(b'{"Ex 1": {}}') + (answers / "Ex 1.jpg").write_bytes(b"answer") + + output = root / "output" + output.mkdir() + (output / "Student.jpg").write_bytes(b"legacy jpeg") + (output / "Student.pdf").write_bytes(b"legacy pdf") + stale_answers = output / "Student" / "answers" + stale_answers.mkdir(parents=True) + (stale_answers / "001 - Ex 1.jpg").write_bytes(b"stale") + scores = pd.DataFrame({0: ["Student"], 1: [12.39]}) + histogram = root / "histogramme.pdf" + histogram.write_bytes(b"histogram") + with patch.object(add_final_score.pd, "read_excel", return_value=scores), \ + patch.object(add_final_score, "HISTOGRAM_PATH", histogram), \ + contextlib.redirect_stdout(io.StringIO()): + add_final_score.process_images(root, output) + + student = output / "Student" + self.assertEqual( + {path.name for path in student.iterdir()}, + {"Student.jpg", "Student.pdf", "score.json", "info.json", "answers"}, + ) + self.assertEqual((student / "Student.pdf").read_bytes(), b"pdf") + self.assertEqual((student / "score.json").read_bytes(), b'{"Ex 1": "4"}') + self.assertEqual((student / "info.json").read_bytes(), b'{"Ex 1": {}}') + self.assertEqual( + (student / "answers" / "Ex 1.jpg").read_bytes(), b"answer" + ) + self.assertFalse((output / "Student.jpg").exists()) + self.assertFalse((output / "Student.pdf").exists()) + self.assertFalse((student / "answers" / "001 - Ex 1.jpg").exists()) + self.assertEqual((output / "histogramme.pdf").read_bytes(), b"histogram") + + def test_omits_answers_directory_when_it_was_not_generated(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "A Rendre" / "Student (01)" + source.mkdir(parents=True) + Image.new("RGB", (300, 200), "white").save(source / "Student.jpg") + (source / "Student.pdf").write_bytes(b"pdf") + (source / "score.json").write_text("{}") + (source / "info.json").write_text("{}") + + output = root / "output" + scores = pd.DataFrame({0: ["Student"], 1: [10]}) + with patch.object(add_final_score.pd, "read_excel", return_value=scores), \ + contextlib.redirect_stdout(io.StringIO()): + add_final_score.process_images(root, output) + + self.assertFalse((output / "Student" / "answers").exists()) + + def test_missing_histogram_warns_without_discarding_student_outputs(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "A Rendre" / "Student (01)" + source.mkdir(parents=True) + Image.new("RGB", (300, 200), "white").save(source / "Student.jpg") + scores = pd.DataFrame({0: ["Student"], 1: [10]}) + output = root / "output" + messages = io.StringIO() + + with patch.object( + add_final_score.pd, "read_excel", return_value=scores + ), patch.object( + add_final_score, "HISTOGRAM_PATH", root / "missing.pdf" + ), contextlib.redirect_stdout(messages): + add_final_score.process_images(root, output) + + self.assertTrue((output / "Student" / "Student.jpg").is_file()) + self.assertFalse((output / "histogramme.pdf").exists()) + self.assertIn("Missing histogram", messages.getvalue()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_answer_returns.py b/tests/test_answer_returns.py index 5d9ed9a..f4c4f52 100644 --- a/tests/test_answer_returns.py +++ b/tests/test_answer_returns.py @@ -11,7 +11,10 @@ from PIL import Image from copienator import EvaluationWorkspace, atomic_write_json, configuration, read_json from copienator.commands import annotating, giving_names from copienator.commands import reading_grouped_annotations as reader -from copienator.return_answers import publish_answer_returns +from copienator.return_answers import ( + publish_answer_returns, + save_return_answer_options, +) class AnswerReturnTests(unittest.TestCase): @@ -54,7 +57,7 @@ class AnswerReturnTests(unittest.TestCase): ), patch.object(annotating, "make_base_image", side_effect=self.supplement): publish_answer_returns(self.root, self.source, self.destination) files = sorted((self.destination / "answers").glob("*.jpg")) - self.assertEqual([p.name for p in files], ["001 - Ex 1.jpg", "002 - Ex 2.jpg"]) + self.assertEqual([p.name for p in files], ["Ex 1.jpg", "Ex 2.jpg"]) with Image.open(files[0]) as image: self.assertEqual(image.size, (100, 30 + 20 * sum((context, question, solution)))) colors = [] @@ -70,6 +73,38 @@ class AnswerReturnTests(unittest.TestCase): self.assertTrue(all(abs(a - b) < 10 for a, b in zip(pixel, color))) self.assertEqual(read_json(self.destination / "info.json"), read_json(self.source / "info.json")) + def test_saved_reading_options_override_configuration_for_answers(self): + save_return_answer_options( + self.root, + context=True, + question=False, + solution=True, + ) + with patch.multiple( + configuration, + RETURN_ANSWERS_ENABLED=True, + RETURN_ANSWERS_CONTEXT=False, + RETURN_ANSWERS_QUESTION=True, + RETURN_ANSWERS_SOLUTION=False, + ), patch.object( + annotating, "make_base_image", side_effect=self.supplement + ): + publish_answer_returns(self.root, self.source, self.destination) + + with Image.open(self.destination / "answers" / "Ex 1.jpg") as image: + self.assertEqual(image.size, (100, 70)) + for y, color in ( + (10, (0, 0, 255)), + (30, (255, 255, 0)), + (50, (255, 0, 0)), + ): + self.assertTrue( + all( + abs(actual - expected) < 10 + for actual, expected in zip(image.getpixel((50, y)), color) + ) + ) + def test_missing_supplement_is_optional_and_failure_preserves_old_answers(self): with patch.multiple(configuration, RETURN_ANSWERS_ENABLED=True, RETURN_ANSWERS_CONTEXT=False, RETURN_ANSWERS_QUESTION=True, @@ -78,7 +113,7 @@ class AnswerReturnTests(unittest.TestCase): ): (self.root / "Text2" / "Ex 1.pdf").unlink() publish_answer_returns(self.root, self.source, self.destination) - path = self.destination / "answers" / "001 - Ex 1.jpg" + path = self.destination / "answers" / "Ex 1.jpg" with Image.open(path) as image: self.assertEqual(image.size, (100, 30)) original = path.read_bytes() @@ -86,7 +121,7 @@ class AnswerReturnTests(unittest.TestCase): with self.assertRaises(FileNotFoundError): publish_answer_returns(self.root, self.source, self.destination) self.assertEqual(path.read_bytes(), original) - self.assertTrue((self.destination / "answers" / "002 - Ex 2.jpg").exists()) + self.assertTrue((self.destination / "answers" / "Ex 2.jpg").exists()) def test_disabling_clears_individual_images_but_retains_info(self): with patch.multiple(configuration, RETURN_ANSWERS_ENABLED=True, @@ -135,8 +170,99 @@ class AnswerReturnTests(unittest.TestCase): self.assertEqual(giving_names.run(workspace, annotation_dir="BGnot"), 0) self.assertEqual({p.name for p in self.destination.iterdir()}, {"answers", "score.json", "info.json"}) + def test_update_uses_copy_id_from_renamed_folder_and_touches_only_answers(self): + workspace = EvaluationWorkspace(self.root) + workspace.copies_dir.mkdir() + atomic_write_json(workspace.copies_dir / "Copie01.json", {"name": "Student"}) + renamed = self.destination.with_name("Nom modifié manuellement (01)") + self.destination.rename(renamed) + (renamed / "Nom personnalisé.jpg").write_bytes(b"keep-jpeg") + (renamed / "Nom personnalisé.pdf").write_bytes(b"keep-pdf") + (renamed / "score.json").write_bytes(b"keep-score") + (renamed / "info.json").write_bytes(b"keep-info") + answers = renamed / "answers" + answers.mkdir() + (answers / "obsolete.jpg").write_bytes(b"obsolete") + preserved = { + path.name: path.read_bytes() + for path in renamed.iterdir() + if path.is_file() + } + + with patch.multiple( + configuration, + RETURN_ANSWERS_ENABLED=True, + RETURN_ANSWERS_CONTEXT=False, + RETURN_ANSWERS_QUESTION=False, + RETURN_ANSWERS_SOLUTION=False, + ), contextlib.redirect_stdout(io.StringIO()): + self.assertEqual( + giving_names.run(workspace, annotation_dir="BGnot", update=True), + 0, + ) + + self.assertFalse((workspace.return_dir / "Student (01)").exists()) + self.assertEqual( + { + path.name: path.read_bytes() + for path in renamed.iterdir() + if path.is_file() + }, + preserved, + ) + self.assertEqual( + sorted(path.name for path in answers.iterdir()), + ["Ex 1.jpg", "Ex 2.jpg"], + ) + class CompiledMembershipTests(unittest.TestCase): + def test_update_score_preserves_file_and_manual_value_wins(self): + with tempfile.TemporaryDirectory() as directory: + workspace = EvaluationWorkspace(Path(directory)) + output = workspace.root / "BGnot" / "Copie01" + output.mkdir(parents=True) + original_score = b'{\n "Ex 1": "3.5"\n}\n' + (output / "score.json").write_bytes(original_score) + answer = workspace.root / "answer.pdf" + answer.touch() + data = { + "01": { + "Ex 1": { + "result": {"score": 1, "feedback": []}, + "pdf_path": answer, + "coordinates": (0, 0), + } + } + } + rendered_scores = [] + + def compose(base, label, result, *args, **kwargs): + rendered_scores.append(result["score"]) + return Image.new("RGB", (100, 50), "white"), 0 + + with patch.object( + annotating, "make_base_image", return_value=(None, 0, 0) + ), patch.object( + annotating, "compose_label_image", side_effect=compose + ), patch.object( + reader, "get_extra_pdfs_as_images", return_value=[] + ), patch.object(reader, "save_paginated_pdf"): + status, _ = reader.apply_actions_and_regenerate_grouped( + workspace, + data, + "01", + [{"label": "Ex 1", "type": "score", "value": "2"}], + {}, + ["Ex 1"], + update_score=True, + ) + + self.assertEqual(status, 0) + self.assertEqual(rendered_scores, ["3.5"]) + self.assertEqual((output / "score.json").read_bytes(), original_score) + self.assertEqual(read_json(output / "info.json")["Ex 1"]["score"], "3.5") + def test_membership_matches_actual_groups_and_saves_every_nonempty_answer(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) diff --git a/tests/test_gui_core.py b/tests/test_gui_core.py index b068599..e03926b 100644 --- a/tests/test_gui_core.py +++ b/tests/test_gui_core.py @@ -46,6 +46,7 @@ from copienator_gui.app import ( from copienator_gui.diagnostics import collect_diagnostics from copienator_gui.runner import ProcessRunner from copienator_gui.state import StateStore +from copienator_gui import workflow as workflow_module from copienator_gui.workflow import build_command, build_workflow, evaluation_argument from copienator.commands.copies_tools import rename_all, rotate_all from copienator.platform import ( @@ -490,7 +491,7 @@ class StandardCliTests(unittest.TestCase): "giving_names": ( "giving_names", "default", - {"target": evaluation, "annotation_dir": "BGnot"}, + {"target": evaluation, "annotation_dir": "BGnot", "update": True}, ), "grouping": ("grouping", "default", {"target": evaluation}), "post_correction": ( @@ -1225,6 +1226,27 @@ class StandardCliTests(unittest.TestCase): self.assertEqual(module.results, {"Ex 1": []}) self.assertEqual(module.tasks_to_process, [("new.jpg", "Ex 1")]) + def test_correction_reserves_unique_group_indices_concurrently(self) -> None: + module = self.modules["correction"] + with tempfile.TemporaryDirectory() as directory: + groups = Path(directory) / "Par label" + label_dir = groups / "Ex 8 : 2)" + label_dir.mkdir(parents=True) + (label_dir / "Group_3.jpg").write_bytes(b"existing") + + with patch.object(module, "GROUPS_DIR", groups): + module.reserved_group_indices.clear() + with ThreadPoolExecutor(max_workers=8) as executor: + indices = list( + executor.map( + module.reserve_next_group_idx, + ["Ex 8 : 2)"] * 8, + ) + ) + + self.assertEqual(sorted(indices), list(range(3, 11))) + self.assertEqual(len(indices), len(set(indices))) + def test_correction_reset_restores_old_and_deletes_new_files(self) -> None: module = self.modules["correction"] with tempfile.TemporaryDirectory() as directory: @@ -1333,6 +1355,43 @@ class StandardCliTests(unittest.TestCase): def test_post_correction_main_cleans_json_atomically(self) -> None: module = self.modules["post_correction"] + cleaned_text = module.clean_string( + r"outside_name already\_escaped; " + r"bad $\\mathbb{U}_n = \\{z \\in \\mathbb{C}\\}$; " + r"valid $\begin{cases} A=0 \\ B=1 \end{cases}$", + {}, + ) + self.assertEqual( + cleaned_text, + r"outside\_name already\_escaped; " + r"bad $\mathbb{U}_n = \{z \in \mathbb{C}\}$; " + r"valid $\begin{cases} A=0 \\ B=1 \end{cases}$", + ) + self.assertEqual(module.clean_string(cleaned_text, {}), cleaned_text) + corrupted_json_escapes = ( + "Formulas $" + + "\x0c" + + "rac{1}{2}$, $" + + "\t" + + "heta " + + "\x0c" + + "euille 0 [2" + + "\t" + + "extbackslash pi]$, $e " + + "\t" + + "imes x$, and [" + + "\n" + + "egthinspace[0,n]" + ) + self.assertEqual( + module.clean_string(corrupted_json_escapes, {}), + r"Formulas $\frac{1}{2}$, $\theta \equiv 0 [2\pi]$, " + r"$e \times x$, and [\negthinspace[0,n]", + ) + self.assertEqual( + module.clean_string("x ∈ A and B ⊂ C", {}), + r"x \ensuremath{\in} A and B \ensuremath{\subset} C", + ) with tempfile.TemporaryDirectory() as directory: evaluation = Path(directory) / "Exam" evaluation.mkdir() @@ -1892,6 +1951,12 @@ class WorkflowTests(unittest.TestCase): self.assertTrue(spec.help.strip()) self.assertTrue(spec.help.rstrip().endswith(".")) + def test_final_score_documents_immediate_gestion_classe_prerequisite(self) -> None: + description = self.steps["final_score"].description + self.assertIn("gestion_classe wse", description) + self.assertIn("juste avant", description) + self.assertIn("histogramme.pdf", description) + def test_options_previously_requiring_free_form_arguments_have_controls(self) -> None: status = self.command( "batch_status", @@ -1914,7 +1979,46 @@ class WorkflowTests(unittest.TestCase): ) self.assertEqual( command_arguments(grouped), - [self.evaluation, "--refaire", "--annotation-dir", "Bnot"], + [ + self.evaluation, + "--refaire", + "--annotation-dir", + "Bnot", + "--no-return-answers-context", + "--return-answers-question", + "--return-answers-solution", + ], + ) + + def test_return_answer_controls_follow_configuration_and_can_be_hidden(self) -> None: + with patch.multiple( + workflow_module.configuration, + RETURN_ANSWERS_ENABLED=True, + RETURN_ANSWERS_CONTEXT=True, + RETURN_ANSWERS_QUESTION=False, + RETURN_ANSWERS_SOLUTION=True, + ): + step = next( + item for item in build_workflow(True) if item.id == "read_annotations" + ) + specs = {spec.name: spec for spec in step.arguments} + self.assertIs(specs["return_answers_context"].default, True) + self.assertIs(specs["return_answers_question"].default, False) + self.assertIs(specs["return_answers_solution"].default, True) + + with patch.object( + workflow_module.configuration, "RETURN_ANSWERS_ENABLED", False + ): + hidden_step = next( + item for item in build_workflow(True) if item.id == "read_annotations" + ) + self.assertFalse( + { + "return_answers_context", + "return_answers_question", + "return_answers_solution", + } + & {spec.name for spec in hidden_step.arguments} ) def test_verbose_is_added_only_to_commands_that_support_it(self) -> None: diff --git a/tests/test_utils_compile.py b/tests/test_utils_compile.py new file mode 100644 index 0000000..0763df3 --- /dev/null +++ b/tests/test_utils_compile.py @@ -0,0 +1,48 @@ +import contextlib +import io +import subprocess +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from copienator import utils + + +class CompileToPdfTests(unittest.TestCase): + def test_warns_when_latex_fails_even_if_partial_pdf_exists(self): + with tempfile.TemporaryDirectory() as temporary: + output = Path(temporary) / "result.pdf" + + def failed_run(*_args, **kwargs): + (Path(kwargs["cwd"]) / "text.pdf").write_bytes(b"partial") + return subprocess.CompletedProcess( + args=[], returncode=1, + stdout="! LaTeX Error: Something's wrong--perhaps a missing \\item.\n", + ) + + console = io.StringIO() + with patch.object(utils.subprocess, "run", side_effect=failed_run), \ + contextlib.redirect_stdout(console): + utils.compile_to_pdf("broken", output) + + self.assertEqual(output.read_bytes(), b"partial") + self.assertIn("Warning: LaTeX compilation failed", console.getvalue()) + self.assertIn("missing \\item", console.getvalue()) + + def test_warns_when_no_pdf_is_produced(self): + with tempfile.TemporaryDirectory() as temporary: + output = Path(temporary) / "result.pdf" + result = subprocess.CompletedProcess(args=[], returncode=0, stdout="") + + console = io.StringIO() + with patch.object(utils.subprocess, "run", return_value=result), \ + contextlib.redirect_stdout(console): + utils.compile_to_pdf("valid", output) + + self.assertFalse(output.exists()) + self.assertIn("produced no PDF", console.getvalue()) + + +if __name__ == "__main__": + unittest.main()