from __future__ import annotations import argparse from collections.abc import Sequence from pathlib import Path from typing import Any import numpy as np from pdf2image import convert_from_path from PIL import Image, ImageChops, ImageDraw, ImageFilter from copienator.commands import annotating from copienator import configuration, 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.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 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 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() try: 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 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) # 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]] = [] 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(reference.width, x2), min(reference.height, y2) region = difference[y1 + 5 : y2 - 5, x1 + 5 : x2 - 5] if region.size == 0: continue # 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) else: 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) 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") 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(alpha)) return actions, notes 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] return bool(np.sum(alpha > 50) > threshold) 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 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] 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, score_path, print) scores = dict.fromkeys(all_labels, "") answer_labels: list[str] = [] 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"] scores[label] = str(result.get("score", 0)) if result.get("error") == "empty-answer": continue sub_note = None 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) 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_image is None: incomplete = True continue 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) dirty_images[label] = final_image answer_labels.append(label) concatenated.append(final_image) if float(scores[label]) != 4.0 or result.get("feedback", []): filtered.append(final_image) concat_image = concatenate(concatenated) filtered_image = concatenate(filtered) with staged_files(output_dir, remove=("Concat.jpg", "Concat_F.jpg", "Concat_F.pdf", "touched.json", "answer_labels.json")) as staging: for label, image in dirty_images.items(): image.save(staging / f"{label}.jpg") 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 )) 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") 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, 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: print(f"Warning: {warning}") if not loaded.data: print("No annotation data found.") return ExitCode.PARTIAL 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 def build_parser() -> argparse.ArgumentParser: parser = evaluation_parser("Read checked annotations and regenerate copies") parser.add_argument( "--update-score", action="store_true", 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 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_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) if __name__ == "__main__": raise SystemExit(main())