Compare commits

..
2 Commits
21 changed files with 521 additions and 65 deletions

No files matched your search

+75
View File
@@ -0,0 +1,75 @@
"""Answer provenance and reproducible annotation spacing (200 dpi pixels)."""
from hashlib import sha256
from pathlib import Path
import shutil
import pymupdf
from copienator import configuration
from copienator.json_io import atomic_write_json, read_json
def fingerprint(path):
return sha256(Path(path).read_bytes()).hexdigest()
def metadata_path(pdf):
return Path(pdf).with_suffix('.layout.json')
def answer_parts(pdf):
"""Ignore stale sidecars after a PDF is replaced or manually edited."""
if pdf is None or not Path(pdf).is_file():
return []
try:
data = read_json(metadata_path(pdf), default={})
if data.get('sha256') == fingerprint(pdf):
parts = data.get('parts', [])
if isinstance(parts, list) and all(isinstance(part, dict) for part in parts):
return parts
except (OSError, ValueError, TypeError, AttributeError):
pass
return []
def save_parts(pdf, parts):
atomic_write_json(metadata_path(pdf), {
'version': 1, 'sha256': fingerprint(pdf), 'parts': parts,
})
def page_parts(pdf):
parts = answer_parts(pdf)
if not Path(pdf).is_file():
return []
with pymupdf.open(pdf) as document:
if len(parts) == len(document):
return parts
return [{'margin_status': 'unknown'} for _ in document]
def copy_answer_pdf(source, destination):
parts = answer_parts(source)
shutil.copy(source, destination)
save_parts(destination, parts)
def move_answer_pdf(source, destination):
parts = answer_parts(source)
shutil.move(str(source), str(destination))
save_parts(destination, parts)
metadata_path(source).unlink(missing_ok=True)
def layout_for(pdf):
low = any(part.get('margin_status') == 'low' for part in answer_parts(pdf))
margin = (configuration.ANNOTATION_LOW_MARGIN_LEFT if low
else configuration.ANNOTATION_MARGIN_LEFT)
if type(margin) is not int or margin < 0:
raise ValueError('Annotation left margins must be non-negative integers')
return {'version': 1, 'margin_left': margin, 'narrow_left_pad_limit': 300}
def legacy_layout():
# Old annotation scans were produced with these literal constants.
return {'version': 1, 'margin_left': 300, 'narrow_left_pad_limit': 300}
+12 -6
View File
@@ -155,7 +155,7 @@ def output_removals(directory: Path) -> tuple[str, ...]:
info = read_json(directory / "info.json") if (directory / "info.json").exists() else {} info = read_json(directory / "info.json") if (directory / "info.json").exists() else {}
return ("Concat.jpg", "Concat_F.jpg", "Concat_F.pdf", "touched.json", "answer_labels.json", return ("Concat.jpg", "Concat_F.jpg", "Concat_F.pdf", "touched.json", "answer_labels.json",
*(f"{label}.jpg" for label, entry in info.items() *(f"{label}.jpg" for label, entry in info.items()
if entry.get("not_empty") and Path(label).name == label)) if Path(label).name == label))
def rebuild(root: Path, mode: str, directory: Path, staging: Path, meta: dict, def rebuild(root: Path, mode: str, directory: Path, staging: Path, meta: dict,
@@ -169,7 +169,15 @@ def rebuild(root: Path, mode: str, directory: Path, staging: Path, meta: dict,
headers, bodies, changed = edited_bodies(directory, meta, instructions) headers, bodies, changed = edited_bodies(directory, meta, instructions)
info = deepcopy(meta["info"]) info = deepcopy(meta["info"])
filtered = set(meta["filtered"]) | changed for label, entry in info.items():
# Retained empty scans are images, not evidence of a non-empty answer.
entry["not_empty"] = label in bodies and (entry["not_empty"] or label in changed)
if label in changed:
entry["present"] = label in bodies
entry["score"] = scores.get(label, entry["score"])
filtered = (set(meta["filtered"]) | changed) & {
label for label, entry in info.items() if entry["not_empty"]
}
all_labels = utils.read_all_labels(root) all_labels = utils.read_all_labels(root)
images, groups = [], [] images, groups = [], []
seen_contexts = set() seen_contexts = set()
@@ -187,15 +195,13 @@ def rebuild(root: Path, mode: str, directory: Path, staging: Path, meta: dict,
fill="black", font_size=30) fill="black", font_size=30)
image = stack([header, body]) image = stack([header, body])
image.save(staging / f"{label}.jpg") image.save(staging / f"{label}.jpg")
images.append(image) if info[label]["not_empty"]:
images.append(image)
if label in filtered: if label in filtered:
extras = get_extra_pdfs_as_images(root, label, annotating, all_labels, extras = get_extra_pdfs_as_images(root, label, annotating, all_labels,
seen_contexts=seen_contexts) if mode == "BGnot" else [] seen_contexts=seen_contexts) if mode == "BGnot" else []
groups.append([*extras, image]) groups.append([*extras, image])
for label, entry in info.items(): for label, entry in info.items():
entry["not_empty"] = label in bodies
if label in changed and label in bodies:
entry["present"] = True
entry["touched"] = mode == "BGnot" and label in bodies and label in filtered entry["touched"] = mode == "BGnot" and label in bodies and label in filtered
entry["score"] = scores.get(label, entry["score"]) entry["score"] = scores.get(label, entry["score"])
atomic_write_json(staging / "info.json", info) atomic_write_json(staging / "info.json", info)
+12 -10
View File
@@ -35,9 +35,9 @@ from copienator.commands.import_annotations import clear_import_directory
from copienator.answer_info import build_answer_info from copienator.answer_info import build_answer_info
from copienator.filesystem import staged_directory from copienator.filesystem import staged_directory
from copienator.feedback_boxes import valid_feedback_box from copienator.feedback_boxes import valid_feedback_box
from copienator.annotation_layout import layout_for
from copienator.utils import natural_key from copienator.utils import natural_key
MARGIN_LEFT = 300
ANNOT_WIDTH = 600 ANNOT_WIDTH = 600
@@ -321,7 +321,7 @@ def compose_label_image(base_img, label, result, hmin,
with_error=True, with_error=True,
with_empty=False, with_empty=False,
more_right=False, more_right=False,
with_id=None, source_pdf=None): with_id=None, source_pdf=None, layout=None):
""" """
Composes the final image with annotations. Composes the final image with annotations.
@@ -350,10 +350,12 @@ def compose_label_image(base_img, label, result, hmin,
context = f"Copie : {with_id} — {context}" context = f"Copie : {with_id} — {context}"
raise AnnotationLatexError(f"{context}\n{exc}") from exc raise AnnotationLatexError(f"{context}\n{exc}") from exc
layout = dict(layout if layout is not None else layout_for(source_pdf))
margin_left = layout['margin_left']
left_pad = 0 left_pad = 0
if base_img.width < TARGET_MIN_WIDTH: if base_img.width < TARGET_MIN_WIDTH:
total_missing = TARGET_MIN_WIDTH - base_img.width total_missing = TARGET_MIN_WIDTH - base_img.width
left_pad = min(total_missing, MARGIN_LEFT) left_pad = min(total_missing, layout['narrow_left_pad_limit'])
new_base = Image.new("RGB", (TARGET_MIN_WIDTH, base_img.height), "white") new_base = Image.new("RGB", (TARGET_MIN_WIDTH, base_img.height), "white")
new_base.paste(base_img, (left_pad, 0)) new_base.paste(base_img, (left_pad, 0))
@@ -409,7 +411,7 @@ def compose_label_image(base_img, label, result, hmin,
total_height = base_img.height + header_height total_height = base_img.height + header_height
# Create Canvas # Create Canvas
final_img = Image.new("RGB", (base_img.width + MARGIN_LEFT, total_height), "white") final_img = Image.new("RGB", (base_img.width + margin_left, total_height), "white")
# Draw Headers # Draw Headers
current_y = 0 current_y = 0
@@ -431,7 +433,7 @@ def compose_label_image(base_img, label, result, hmin,
# Paste Base Image # Paste Base Image
image_offset_y = current_y image_offset_y = current_y
final_img.paste(base_img, (MARGIN_LEFT, image_offset_y)) final_img.paste(base_img, (margin_left, image_offset_y))
# 2. Draw Local Annotations # 2. Draw Local Annotations
draw = ImageDraw.Draw(final_img, "RGBA") # Refresh draw object draw = ImageDraw.Draw(final_img, "RGBA") # Refresh draw object
@@ -443,8 +445,8 @@ def compose_label_image(base_img, label, result, hmin,
target_ymin = (ymin - hmin) + image_offset_y target_ymin = (ymin - hmin) + image_offset_y
target_ymax = (ymax - hmin) + image_offset_y target_ymax = (ymax - hmin) + image_offset_y
target_xmin = xmin + MARGIN_LEFT + left_pad target_xmin = xmin + margin_left + left_pad
target_xmax = xmax + MARGIN_LEFT + left_pad target_xmax = xmax + margin_left + left_pad
# Draw Rectangle (if not suppressed) # Draw Rectangle (if not suppressed)
if "norectangle" not in fb: if "norectangle" not in fb:
@@ -485,6 +487,7 @@ def compose_label_image(base_img, label, result, hmin,
last_text_bottom = paste_y + txt_img.height last_text_bottom = paste_y + txt_img.height
final_img.info['annotation_layout'] = layout
return final_img, header_height return final_img, header_height
def process_student(student_id, labels_data, root_dir, all_labels, overwrite): def process_student(student_id, labels_data, root_dir, all_labels, overwrite):
@@ -544,14 +547,13 @@ def process_student(student_id, labels_data, root_dir, all_labels, overwrite):
if label_images: if label_images:
max_w = max(image.width for image in label_images) max_w = max(image.width for image in label_images)
total_h = sum(image.height for image in label_images) total_h = sum(image.height for image in label_images)
canvas = Image.new('RGB', (max_w, total_h)) canvas = Image.new('RGB', (max_w, total_h), 'white')
current_y = 0 current_y = 0
for image in label_images: for image in label_images:
canvas.paste(image, (0, current_y)) canvas.paste(image, (0, current_y))
current_y += image.height current_y += image.height
canvas.save(staging / "Concat.jpg") canvas.save(staging / "Concat.jpg")
elif labels_data: # An all-empty copy still has valid individual return images.
problems = True
return "partial" if problems else "success" return "partial" if problems else "success"
@@ -87,6 +87,7 @@ def save_batch(batch, prefix, group_id, output_root: Path) -> None:
"id": student_id, "id": student_id,
"label": label, "label": label,
"header_height": header_height, "header_height": header_height,
"layout": image.info.get('annotation_layout', annotating.layout_for(None)),
"hmin": current_y, "hmin": current_y,
"hmax": current_y + image.height, "hmax": current_y + image.height,
} }
@@ -196,6 +196,7 @@ def _render_student(
"id": student_id, "id": student_id,
"label": label, "label": label,
"header_height": header_height, "header_height": header_height,
"layout": final_image.info.get('annotation_layout', annotating.layout_for(None)),
"img_h": final_image.height, "img_h": final_image.height,
} }
) )
+8 -8
View File
@@ -7,7 +7,6 @@ import json
import os import os
import signal import signal
import shlex import shlex
import shutil
import sys import sys
import threading import threading
import time import time
@@ -20,6 +19,7 @@ from google.genai import types
from copienator import configuration as config from copienator import configuration as config
from copienator.commands import grouping from copienator.commands import grouping
from copienator.answer_origins import save_origins from copienator.answer_origins import save_origins
from copienator.annotation_layout import copy_answer_pdf, move_answer_pdf
from copienator import prompting from copienator import prompting
from copienator.gemini_config import fresh_seed, thinking_config from copienator.gemini_config import fresh_seed, thinking_config
from copienator import configuration as settings from copienator import configuration as settings
@@ -427,10 +427,10 @@ def handle_label_errors(pid, label, res, pdf_path):
tprint(f"\t\tCopie{pid} : moving wrong {label} to {new_label}.") tprint(f"\t\tCopie{pid} : moving wrong {label} to {new_label}.")
# Copie vers _new, puis renommage de l'original vers _old # Copie vers _new, puis renommage de l'original vers _old
shutil.copy(str(pdf_path), str(new_pdf_path)) copy_answer_pdf(pdf_path, new_pdf_path)
old_pdf_path = pdf_path.with_name(f"{label}_old.pdf") old_pdf_path = pdf_path.with_name(f"{label}_old.pdf")
if pdf_path != old_pdf_path: if pdf_path != old_pdf_path:
shutil.move(str(pdf_path), str(old_pdf_path)) move_answer_pdf(pdf_path, old_pdf_path)
idx = reserve_next_group_idx(new_label) idx = reserve_next_group_idx(new_label)
height = grouping.get_pdf_height(str(new_pdf_path)) height = grouping.get_pdf_height(str(new_pdf_path))
@@ -464,7 +464,7 @@ def handle_label_errors(pid, label, res, pdf_path):
add_pdf_path = COPIES_DIR / f"Copie{pid}" / f"{add_label}_new.pdf" add_pdf_path = COPIES_DIR / f"Copie{pid}" / f"{add_label}_new.pdf"
if not base_add_pdf_path.exists() and not add_pdf_path.exists(): if not base_add_pdf_path.exists() and not add_pdf_path.exists():
shutil.copy(str(pdf_path), str(add_pdf_path)) copy_answer_pdf(pdf_path, add_pdf_path)
save_origins(INPUT_DIR, pid, add_label, add_pdf_path, [label]) save_origins(INPUT_DIR, pid, add_label, add_pdf_path, [label])
tprint(f"\t\tCopying Copie{pid} : {label} -> {add_label}") tprint(f"\t\tCopying Copie{pid} : {label} -> {add_label}")
idx = reserve_next_group_idx(add_label) idx = reserve_next_group_idx(add_label)
@@ -578,7 +578,7 @@ def process_single_task(task_tuple, precomputed_response=None):
if res.get("error") == "empty-answer": if res.get("error") == "empty-answer":
old_path = pdf_path.with_name(f"{label}_old.pdf") old_path = pdf_path.with_name(f"{label}_old.pdf")
if pdf_path.exists() and pdf_path != old_path: if pdf_path.exists() and pdf_path != old_path:
shutil.move(str(pdf_path), str(old_path)) move_answer_pdf(pdf_path, old_path)
pdf_path = old_path pdf_path = old_path
current_suffix = "_old" current_suffix = "_old"
@@ -729,10 +729,10 @@ def resolve_delayed_moves():
res["suffix"] = "_old" # Fixed typo: was suffixe res["suffix"] = "_old" # Fixed typo: was suffixe
resolved_any = True resolved_any = True
shutil.copy(str(pdf_path), str(new_pdf_path)) copy_answer_pdf(pdf_path, new_pdf_path)
old_pdf_path = pdf_path.with_name(f"{label}_old.pdf") old_pdf_path = pdf_path.with_name(f"{label}_old.pdf")
if pdf_path != old_pdf_path: if pdf_path != old_pdf_path:
shutil.move(str(pdf_path), str(old_pdf_path)) move_answer_pdf(pdf_path, old_pdf_path)
idx = reserve_next_group_idx(target_label) idx = reserve_next_group_idx(target_label)
height = grouping.get_pdf_height(str(new_pdf_path)) height = grouping.get_pdf_height(str(new_pdf_path))
@@ -751,7 +751,7 @@ def resolve_delayed_moves():
res["error"] = res["error"].replace(f"(xx){target_label}", f"(->){target_label}") res["error"] = res["error"].replace(f"(xx){target_label}", f"(->){target_label}")
resolved_any = True resolved_any = True
shutil.copy(str(pdf_path), str(add_pdf_path)) copy_answer_pdf(pdf_path, add_pdf_path)
save_origins(INPUT_DIR, pid, target_label, add_pdf_path, [label]) save_origins(INPUT_DIR, pid, target_label, add_pdf_path, [label])
idx = reserve_next_group_idx(target_label) idx = reserve_next_group_idx(target_label)
height = grouping.get_pdf_height(str(add_pdf_path)) height = grouping.get_pdf_height(str(add_pdf_path))
+3
View File
@@ -29,6 +29,7 @@ from copienator.filesystem import staged_files
from copienator.copy_errors import clear_copy_error, mark_copy_error, marked_copy_paths from copienator.copy_errors import clear_copy_error, mark_copy_error, marked_copy_paths
from copienator.configuration import SHOW_PERSONAL_STEPS from copienator.configuration import SHOW_PERSONAL_STEPS
from copienator.left_margin import analyse_margins from copienator.left_margin import analyse_margins
from copienator.annotation_layout import fingerprint
DELIMITER_WIDTH = 5 DELIMITER_WIDTH = 5
DELIMITER_COLOR = (0, 0, 0) DELIMITER_COLOR = (0, 0, 0)
@@ -211,6 +212,8 @@ def save_results(
) -> None: ) -> None:
"""Atomically replace every Cutleft output associated with one copy.""" """Atomically replace every Cutleft output associated with one copy."""
_, splits, schema = result _, splits, schema = result
if pdf_path.is_file():
schema = {**schema, 'source_sha256': fingerprint(pdf_path)}
base_name = pdf_path.stem base_name = pdf_path.stem
previous = _previous_cutleft_outputs(output_dir, base_name) previous = _previous_cutleft_outputs(output_dir, base_name)
with staged_files(output_dir, remove=previous) as staging: with staged_files(output_dir, remove=previous) as staging:
+27 -9
View File
@@ -22,6 +22,7 @@ from copienator import (
) )
from copienator.annotation_actions import apply_checkbox_actions, apply_score_overrides from copienator.annotation_actions import apply_checkbox_actions, apply_score_overrides
from copienator.annotation_data import AnnotationData, load_annotation_data from copienator.annotation_data import AnnotationData, load_annotation_data
from copienator.annotation_layout import legacy_layout
from copienator.annotation_modifications import ( from copienator.annotation_modifications import (
BASELINE, BASELINE,
output_removals, output_removals,
@@ -157,10 +158,14 @@ def apply_actions_and_regenerate(
"""Apply annotations and atomically merge the regenerated student files.""" """Apply annotations and atomically merge the regenerated student files."""
output_dir = workspace.annotation_dir("checks") / f"Copie{student_id}" output_dir = workspace.annotation_dir("checks") / f"Copie{student_id}"
bnote_path = output_dir / "bnote.json" bnote_path = output_dir / "bnote.json"
if not bnote_path.is_file(): only_empty = bool(data[student_id]) and all(
content["result"].get("error") == "empty-answer"
for content in data[student_id].values()
)
if not bnote_path.is_file() and not only_empty:
print(f" Missing {bnote_path}") print(f" Missing {bnote_path}")
return ExitCode.PARTIAL return ExitCode.PARTIAL
bnote_data = read_json(bnote_path) bnote_data = read_json(bnote_path) if bnote_path.is_file() else {"images": []}
if not isinstance(bnote_data, dict): if not isinstance(bnote_data, dict):
raise TypeError(f"Expected a JSON object in {bnote_path}") raise TypeError(f"Expected a JSON object in {bnote_path}")
@@ -180,7 +185,14 @@ def apply_actions_and_regenerate(
filtered: list[Image.Image] = [] filtered: list[Image.Image] = []
incomplete = False incomplete = False
for image_info in bnote_data.get("images", []): review_images = list(bnote_data.get("images", []))
reviewed_labels = {str(item.get("label", "")) for item in review_images if isinstance(item, dict)}
# Empty answers are omitted from the review sheet, but still need return images.
review_images.extend(
{"label": label} for label, content in labels_data.items()
if label not in reviewed_labels and content["result"].get("error") == "empty-answer"
)
for image_info in review_images:
if not isinstance(image_info, dict): if not isinstance(image_info, dict):
incomplete = True incomplete = True
continue continue
@@ -191,11 +203,10 @@ def apply_actions_and_regenerate(
content = labels_data[label] content = labels_data[label]
result = content["result"] result = content["result"]
scores[label] = str(result.get("score", 0)) scores[label] = str(result.get("score", 0))
if result.get("error") == "empty-answer": is_empty = result.get("error") == "empty-answer"
continue
sub_note = None sub_note = None
if notes_layer is not None: if notes_layer is not None and "hmin" in image_info:
hmin = int(image_info.get("hmin", 0)) hmin = int(image_info.get("hmin", 0))
hmax = int(image_info.get("hmax", 0)) hmax = int(image_info.get("hmax", 0))
sub_note = notes_layer.crop((0, hmin, notes_layer.width, hmax)) sub_note = notes_layer.crop((0, hmin, notes_layer.width, hmax))
@@ -213,6 +224,9 @@ def apply_actions_and_regenerate(
result, result,
content["coordinates"][0], content["coordinates"][0],
with_error=False, with_error=False,
with_empty=True,
source_pdf=pdf_path,
layout=image_info.get('layout', legacy_layout()),
) )
if final_image is None: if final_image is None:
incomplete = True incomplete = True
@@ -230,6 +244,8 @@ def apply_actions_and_regenerate(
dirty_images[label] = final_image dirty_images[label] = final_image
header_heights[label] = new_header_height header_heights[label] = new_header_height
if is_empty:
continue
answer_labels.append(label) answer_labels.append(label)
concatenated.append(final_image) concatenated.append(final_image)
if float(scores[label]) != 4.0 or result.get("feedback", []): if float(scores[label]) != 4.0 or result.get("feedback", []):
@@ -302,13 +318,15 @@ def run(
status = ExitCode.PARTIAL if loaded.warnings else ExitCode.SUCCESS status = ExitCode.PARTIAL if loaded.warnings else ExitCode.SUCCESS
for student_id in sorted(loaded.data, key=utils.natural_key): for student_id in sorted(loaded.data, key=utils.natural_key):
output_dir = workspace.annotation_dir("checks") / f"Copie{student_id}" output_dir = workspace.annotation_dir("checks") / f"Copie{student_id}"
if not output_dir.is_dir(): only_empty = all(content["result"].get("error") == "empty-answer"
for content in loaded.data[student_id].values())
if not output_dir.is_dir() and not only_empty:
print(f"Warning: missing annotation directory {output_dir}") print(f"Warning: missing annotation directory {output_dir}")
status = ExitCode.PARTIAL status = ExitCode.PARTIAL
continue continue
print(f"Processing annotations for: {student_id}") print(f"Processing annotations for: {student_id}")
actions, notes = detect_checks_and_notes(output_dir) actions, notes = ([], None) if only_empty else detect_checks_and_notes(output_dir)
if notes is None and not actions and not update_score: if notes is None and not actions and not update_score and not only_empty:
print(" No readable annotation input found.") print(" No readable annotation input found.")
status = ExitCode.PARTIAL status = ExitCode.PARTIAL
continue continue
@@ -25,6 +25,7 @@ from copienator import (
) )
from copienator.annotation_actions import apply_checkbox_actions, apply_score_overrides from copienator.annotation_actions import apply_checkbox_actions, apply_score_overrides
from copienator.annotation_data import AnnotationData, RefaireList, load_annotation_data from copienator.annotation_data import AnnotationData, RefaireList, load_annotation_data
from copienator.annotation_layout import legacy_layout
from copienator.annotation_modifications import ( from copienator.annotation_modifications import (
BASELINE, BASELINE,
baseline, baseline,
@@ -168,7 +169,6 @@ def _scan_annotation_directory(
if notes_image is None: if notes_image is None:
if required: if required:
raise ValueError(f"Could not read annotations in {directory}") raise ValueError(f"Could not read annotations in {directory}")
return {}, {}
actions_by_student: dict[str, list[dict[str, Any]]] = defaultdict(list) actions_by_student: dict[str, list[dict[str, Any]]] = defaultdict(list)
notes_by_student: dict[str, LabelNotes] = defaultdict(dict) notes_by_student: dict[str, LabelNotes] = defaultdict(dict)
for action in actions: for action in actions:
@@ -181,12 +181,13 @@ def _scan_annotation_directory(
hmin = int(image_info.get("hmin", 0)) hmin = int(image_info.get("hmin", 0))
hmax = int(image_info.get("hmax", 0)) hmax = int(image_info.get("hmax", 0))
if student_id and label and hmax > hmin: if student_id and label and hmax > hmin:
crop = notes_image.crop((0, hmin, notes_image.width, hmax)) crop = (notes_image.crop((0, hmin, notes_image.width, hmax))
if has_significant_notes(crop): if notes_image is not None else None)
notes_by_student[student_id][label] = { notes_by_student[student_id][label] = {
"img": crop, "img": crop if has_significant_notes(crop) else None,
"old_header_h": int(image_info.get("header_height", 0)), "old_header_h": int(image_info.get("header_height", 0)),
} "layout": image_info.get('layout', legacy_layout()),
}
return dict(actions_by_student), dict(notes_by_student) return dict(actions_by_student), dict(notes_by_student)
@@ -292,8 +293,7 @@ def apply_actions_and_regenerate_grouped(
result["score"] = old_scores[label] result["score"] = old_scores[label]
scores[label] = str(result.get("score", 0)) scores[label] = str(result.get("score", 0))
touched[label] = False touched[label] = False
if result.get("error") == "empty-answer": is_empty = result.get("error") == "empty-answer"
continue
saved_image = output_dir / f"{label}.jpg" saved_image = output_dir / f"{label}.jpg"
baseline_entry = (previous_baseline or {}).get("images", {}).get(label) baseline_entry = (previous_baseline or {}).get("images", {}).get(label)
if selected_labels and label not in selected_labels and (saved_image.is_file() or baseline_entry): if selected_labels and label not in selected_labels and (saved_image.is_file() or baseline_entry):
@@ -319,6 +319,8 @@ def apply_actions_and_regenerate_grouped(
) )
dirty_images[label] = final_image dirty_images[label] = final_image
scores[label] = str(old_scores.get(label, scores[label])) scores[label] = str(old_scores.get(label, scores[label]))
if is_empty:
continue
concat_images.append(final_image) concat_images.append(final_image)
answer_labels.append(label) answer_labels.append(label)
filtered_labels.append(label) filtered_labels.append(label)
@@ -344,6 +346,9 @@ def apply_actions_and_regenerate_grouped(
result, result,
content["coordinates"][0], content["coordinates"][0],
with_error=False, with_error=False,
with_empty=True,
source_pdf=pdf_path,
layout=label_notes.get(label, {}).get('layout', legacy_layout()),
) )
if final_image is None: if final_image is None:
incomplete = True incomplete = True
@@ -368,6 +373,8 @@ def apply_actions_and_regenerate_grouped(
# Persist every final block, including unchanged answers, for returns. # Persist every final block, including unchanged answers, for returns.
dirty_images[label] = final_image dirty_images[label] = final_image
header_heights[label] = new_header_height header_heights[label] = new_header_height
if is_empty:
continue
answer_labels.append(label) answer_labels.append(label)
concat_images.append(final_image) concat_images.append(final_image)
+12 -4
View File
@@ -2,7 +2,6 @@ from __future__ import annotations
import argparse import argparse
import re import re
import shutil
from collections.abc import Sequence from collections.abc import Sequence
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
@@ -10,6 +9,8 @@ from typing import Any
from pypdf import PdfWriter from pypdf import PdfWriter
from copienator.pdf_cut import split_pdf from copienator.pdf_cut import split_pdf
from copienator.annotation_layout import (copy_answer_pdf, move_answer_pdf,
page_parts, save_parts, metadata_path)
from copienator.answer_origins import origins_for, save_origins from copienator.answer_origins import origins_for, save_origins
from copienator import ( from copienator import (
@@ -240,6 +241,8 @@ def resolve_manual(workspace: EvaluationWorkspace) -> ExitCode:
if not instruction.cut and instruction.operator == "->": if not instruction.cut and instruction.operator == "->":
origins[key_new] = list(dict.fromkeys([ origins[key_new] = list(dict.fromkeys([
*origins[key_new], instruction.old_label])) *origins[key_new], instruction.old_label]))
source_parts = page_parts(source)
destination_parts = page_parts(destination)
writer = PdfWriter() writer = PdfWriter()
try: try:
if instruction.pipe_first: if instruction.pipe_first:
@@ -251,8 +254,11 @@ def resolve_manual(workspace: EvaluationWorkspace) -> ExitCode:
writer.append(destination) writer.append(destination)
writer.append(source) writer.append(source)
writer.write(temp_output) writer.write(temp_output)
save_parts(temp_output, (source_parts + destination_parts
if instruction.pipe_first else destination_parts + source_parts))
except Exception: except Exception:
temp_output.unlink(missing_ok=True) temp_output.unlink(missing_ok=True)
metadata_path(temp_output).unlink(missing_ok=True)
raise raise
finally: finally:
writer.close() writer.close()
@@ -264,12 +270,13 @@ def resolve_manual(workspace: EvaluationWorkspace) -> ExitCode:
[instruction.old_label] if instruction.operator == "-x" [instruction.old_label] if instruction.operator == "-x"
and not instruction.cut else [] and not instruction.cut else []
) )
shutil.copy(source, temp_output) copy_answer_pdf(source, temp_output)
current_paths[key_new] = temp_output current_paths[key_new] = temp_output
temp_files.append(temp_output) temp_files.append(temp_output)
except Exception: except Exception:
for temporary in temp_files: for temporary in temp_files:
temporary.unlink(missing_ok=True) temporary.unlink(missing_ok=True)
metadata_path(temporary).unlink(missing_ok=True)
raise raise
for pdf in files_to_old: for pdf in files_to_old:
@@ -280,7 +287,7 @@ def resolve_manual(workspace: EvaluationWorkspace) -> ExitCode:
old_name = pdf.with_name(f"{label}_old.pdf") old_name = pdf.with_name(f"{label}_old.pdf")
if pdf != old_name: if pdf != old_name:
old_name.unlink(missing_ok=True) old_name.unlink(missing_ok=True)
shutil.move(str(pdf), str(old_name)) move_answer_pdf(pdf, old_name)
set_suffix_and_clean_error(results, copy_id, label, "_old") set_suffix_and_clean_error(results, copy_id, label, "_old")
for instruction in instructions: for instruction in instructions:
@@ -298,7 +305,7 @@ def resolve_manual(workspace: EvaluationWorkspace) -> ExitCode:
continue continue
final_name = workspace.copies_dir / f"Copie{copy_id}" / f"{label}_new.pdf" final_name = workspace.copies_dir / f"Copie{copy_id}" / f"{label}_new.pdf"
final_name.unlink(missing_ok=True) final_name.unlink(missing_ok=True)
shutil.move(str(current_path), str(final_name)) move_answer_pdf(current_path, final_name)
retained_origins = [origin for origin in origins[(copy_id, label)] retained_origins = [origin for origin in origins[(copy_id, label)]
if (copy_id, origin) not in initial_paths if (copy_id, origin) not in initial_paths
or initial_paths[(copy_id, origin)] not in files_to_old] or initial_paths[(copy_id, origin)] not in files_to_old]
@@ -312,6 +319,7 @@ def resolve_manual(workspace: EvaluationWorkspace) -> ExitCode:
for temporary in temp_files: for temporary in temp_files:
if temporary not in used_temps: if temporary not in used_temps:
temporary.unlink(missing_ok=True) temporary.unlink(missing_ok=True)
metadata_path(temporary).unlink(missing_ok=True)
for copy_id, label, target in dismissed_conflicts: for copy_id, label, target in dismissed_conflicts:
# A removed instruction acknowledges the conflict without touching PDFs. # A removed instruction acknowledges the conflict without touching PDFs.
+17
View File
@@ -22,6 +22,7 @@ from copienator import (
workspace_from_target, workspace_from_target,
) )
from copienator.filesystem import staged_directory from copienator.filesystem import staged_directory
from copienator.annotation_layout import fingerprint, metadata_path, save_parts
SQUARE = 1000 // 38 SQUARE = 1000 // 38
ANSWER_TOP_PADDING_POINTS = 4 * 72 / 25.4 ANSWER_TOP_PADDING_POINTS = 4 * 72 / 25.4
@@ -129,6 +130,11 @@ def _render_split_outputs(
visible_bounds = _prepare_split_pages(document) visible_bounds = _prepare_split_pages(document)
parsed = _parse_coordinates(coords_list) parsed = _parse_coordinates(coords_list)
parts_by_label: defaultdict[str, list[Path]] = defaultdict(list) parts_by_label: defaultdict[str, list[Path]] = defaultdict(list)
provenance: defaultdict[str, list[dict]] = defaultdict(list)
schema = read_json(input_pdf.parent.parent / 'Cutleft' / f'{input_pdf.stem}_schema.json', default={})
analysis = schema.get('auto_margin') or {}
measurements = ({item['page']: item for item in analysis.get('pages', [])}
if schema.get('source_sha256') == fingerprint(input_pdf) else {})
with tempfile.TemporaryDirectory(prefix="copienator-split-") as temp_directory: with tempfile.TemporaryDirectory(prefix="copienator-split-") as temp_directory:
temporary = Path(temp_directory) temporary = Path(temp_directory)
for index, item in enumerate(parsed): for index, item in enumerate(parsed):
@@ -200,6 +206,15 @@ def _render_split_outputs(
part_path, part_path,
) )
parts_by_label[clean_label].append(part_path) parts_by_label[clean_label].append(part_path)
edge = measurements.get(page_number + 1, {}).get('edge_mm')
remaining = None if edge is None else max(0, edge - clip.x0 * 25.4 / 72)
provenance[clean_label].append({
'source': input_pdf.name, 'source_sha256': schema.get('source_sha256'),
'page': page_number + 1, 'crop_points': list(clip),
'margin_mm': remaining,
'margin_status': ('unknown' if remaining is None else
'low' if remaining < 20 else 'normal'),
})
generated: set[str] = set() generated: set[str] = set()
for label, parts in parts_by_label.items(): for label, parts in parts_by_label.items():
@@ -212,6 +227,8 @@ def _render_split_outputs(
finally: finally:
merger.close() merger.close()
generated.add(filename) generated.add(filename)
save_parts(staging / filename, provenance[label])
generated.add(metadata_path(staging / filename).name)
return generated return generated
finally: finally:
document.close() document.close()
+2
View File
@@ -37,6 +37,8 @@ THINKING_LEVEL_FOR_LABEL = "medium"
THINKING_LEVEL_FLASH = "medium" THINKING_LEVEL_FLASH = "medium"
THINKING_LEVEL_PRO = "high" THINKING_LEVEL_PRO = "high"
ALWAYS_CROP = False ALWAYS_CROP = False
ANNOTATION_MARGIN_LEFT = 300
ANNOTATION_LOW_MARGIN_LEFT = 400
SHOW_PERSONAL_STEPS = False SHOW_PERSONAL_STEPS = False
RETURN_JPEG_ENABLED = True RETURN_JPEG_ENABLED = True
RETURN_PDF_ENABLED = True RETURN_PDF_ENABLED = True
+9 -1
View File
@@ -144,7 +144,15 @@ def analyse_margins(images: list[Image.Image], render_dpi: int, reference_width:
starts[run_start:] = [0] * (len(images) - run_start) starts[run_start:] = [0] * (len(images) - run_start)
break break
return {'version': 5, 'starts_px': starts, 'groups': groups, pages = []
for index in range(len(images)):
estimate = page_estimate(index)
edge = estimate['edge_mm']
pages.append({'page': index + 1, **estimate,
'status': 'unknown' if edge is None else
('low' if edge < MIN_MARGIN_MM else 'normal')})
return {'version': 6, 'pages': pages, 'starts_px': starts, 'groups': groups,
'no_margin_from_page': no_margin_from_page, 'no_margin_from_page': no_margin_from_page,
'discard_left_fraction': DISCARD_LEFT_FRACTION, 'cut_fraction_of_margin': .5, 'discard_left_fraction': DISCARD_LEFT_FRACTION, 'cut_fraction_of_margin': .5,
'standard_margin_mm': STANDARD_MARGIN_MM, 'minimum_margin_mm': MIN_MARGIN_MM, 'standard_margin_mm': STANDARD_MARGIN_MM, 'minimum_margin_mm': MIN_MARGIN_MM,
+10
View File
@@ -6,6 +6,7 @@ from pathlib import Path
import pymupdf import pymupdf
from copienator.commands.splitting_int import _prepare_split_pages from copienator.commands.splitting_int import _prepare_split_pages
from copienator.annotation_layout import page_parts, save_parts
# Percentages emitted by the GUI have six decimal places. Recover exact page # Percentages emitted by the GUI have six decimal places. Recover exact page
# boundaries despite rounding, without turning nearby in-page cuts into breaks. # boundaries despite rounding, without turning nearby in-page cuts into breaks.
@@ -35,6 +36,7 @@ def cut_position(heights: list[float], percent: float) -> tuple[int, float]:
def split_pdf(source: Path, percent: float, first_path: Path, second_path: Path) -> None: def split_pdf(source: Path, percent: float, first_path: Path, second_path: Path) -> None:
"""Preserve whole pages; clip only the page actually crossed by the cut.""" """Preserve whole pages; clip only the page actually crossed by the cut."""
parts = page_parts(source)
with pymupdf.open(source) as document, pymupdf.open() as first, pymupdf.open() as second: with pymupdf.open(source) as document, pymupdf.open() as first, pymupdf.open() as second:
index, offset = cut_position([page.rect.height for page in document], percent) index, offset = cut_position([page.rect.height for page in document], percent)
if index: if index:
@@ -55,3 +57,11 @@ def split_pdf(source: Path, percent: float, first_path: Path, second_path: Path)
second.insert_pdf(document, from_page=index + 1) second.insert_pdf(document, from_page=index + 1)
first.save(first_path) first.save(first_path)
second.save(second_path) second.save(second_path)
first_parts = [dict(part) for part in parts[:index + bool(offset)]]
second_parts = [dict(part) for part in parts[index:]]
if offset and 'crop_points' in parts[index]:
x0, y0, x1, y1 = parts[index]['crop_points']
first_parts[-1]['crop_points'] = [x0, y0, x1, y0 + offset]
second_parts[0]['crop_points'] = [x0, y0 + offset, x1, y1]
save_parts(first_path, first_parts)
save_parts(second_path, second_parts)
+1 -1
View File
@@ -84,7 +84,7 @@ def publish_answer_returns(
raise ValueError(f"Expected a real answer directory: {answers_dir}") raise ValueError(f"Expected a real answer directory: {answers_dir}")
if configuration.RETURN_ANSWERS_ENABLED: if configuration.RETURN_ANSWERS_ENABLED:
options = load_return_answer_options(root) options = load_return_answer_options(root)
labels = [label for label, entry in info.items() if entry["present"] and entry["not_empty"]] labels = [label for label, entry in info.items() if entry["present"]]
# Import the rendering backend only when individual images are requested. # Import the rendering backend only when individual images are requested.
from copienator.commands.annotating import make_base_image from copienator.commands.annotating import make_base_image
+4
View File
@@ -6,6 +6,10 @@ API_KEY = os.environ.get("GEMINI_API_KEY")
EXPORT_DIR = Path("Export") EXPORT_DIR = Path("Export")
IMPORT_DIR = Path("Import") IMPORT_DIR = Path("Import")
# Espace ajouté à gauche des réponses annotées, en pixels à 200 dpi.
ANNOTATION_MARGIN_LEFT = 300
ANNOTATION_LOW_MARGIN_LEFT = 400
# Fichiers à inclure dans A Rendre (les sources d'annotations sont conservées). # Fichiers à inclure dans A Rendre (les sources d'annotations sont conservées).
RETURN_JPEG_ENABLED = True RETURN_JPEG_ENABLED = True
RETURN_PDF_ENABLED = True RETURN_PDF_ENABLED = True
+46 -9
View File
@@ -4,6 +4,40 @@ This documents the current implementation, as of 2026-09-12. The return folder
referred to as « À rendre » is named **`A Rendre`** on disk. JPEG files use the referred to as « À rendre » is named **`A Rendre`** on disk. JPEG files use the
extension **`.jpg`**, not `.jpeg`. extension **`.jpg`**, not `.jpeg`.
## Left space for Gemini comments
`config.py` (or the file selected by `COPIENATOR_CONFIG`) can override:
```python
ANNOTATION_MARGIN_LEFT = 300
ANNOTATION_LOW_MARGIN_LEFT = 400
```
These are added pixels at the 200 dpi annotation rendering scale. An answer
uses the larger margin if any of its source fragments has a measured left
margin below 20 mm (half the standard 40 mm). Unknown or insufficiently written
pages retain the normal spacing. The 600 px comment width is unchanged, so
comments can still overlap handwriting on pages with very little whitespace.
Automatic `crop-labels` saves per-page measurements in the Cutleft schema.
`split-answers` records source pages, crop rectangles and remaining margins in
`*.layout.json` files beside answer PDFs. Those files follow copies, manual
merges and vertical cuts. PDF fingerprints prevent using stale measurements
after files are replaced. Legacy schemas and manual/full-page Cutleft runs
without measurements keep the normal spacing.
For an existing evaluation, rerun automatic `crop-labels`, `split-answers`, and
annotation generation with overwrite to populate and use this metadata. Do
this before handwritten review: existing Bnot/BGnot/BRnot annotation batches
save their rendering geometry and reuse it when read back, even after settings
change. Older batches without saved geometry keep their original 300 px offset.
Do not overwrite a batch that still contains handwritten corrections to import.
All fragments of one answer share one left offset. BGnot pads narrower answer
images with white without resizing handwriting; fitting the wider document to
screen or paper may reduce the displayed scale, and return-PDF page breaks can
change. The extra space is also used by Anot and Bnot.
## Files and their sources ## Files and their sources
After grouped correction and review: After grouped correction and review:
@@ -39,7 +73,7 @@ answer JPEGs.
| `Student Name.pdf` | `Concat_F.pdf` | Filtered, paginated correction with context, questions and solutions. | | `Student Name.pdf` | `Concat_F.pdf` | Filtered, paginated correction with context, questions and solutions. |
| `score.json` | `score.json` | Per-question scores, including questions omitted from the filtered PDF. | | `score.json` | `score.json` | Per-question scores, including questions omitted from the filtered PDF. |
| `info.json` | `info.json` | Answer presence, empty-answer classification, PDF membership and score. | | `info.json` | `info.json` | Answer presence, empty-answer classification, PDF membership and score. |
| `answers/*.jpg` | Final per-label JPEGs selected by `info.json` | One annotated non-empty answer, with optional supplementary material. | | `answers/*.jpg` | Final per-label JPEGs selected by `info.json` | One annotated answer, including empty scans, with optional supplementary material. |
## Enabling or disabling outputs ## Enabling or disabling outputs
@@ -69,9 +103,9 @@ and preserves return PDFs when present.
## Individual answer JPEGs ## Individual answer JPEGs
With `RETURN_ANSWERS_ENABLED = True`, `giving-names` generates an `answers/` With `RETURN_ANSWERS_ENABLED = True`, `giving-names` generates an `answers/`
subdirectory inside each student's return folder. It includes **every non-empty subdirectory inside each student's return folder. It includes **every present
compiled answer**, even a perfect answer omitted from the filtered PDF. Labels compiled answer**, even a perfect answer omitted from the filtered PDF. Scans
marked `empty-answer` and labels without an answer are excluded. Every JPEG marked `empty-answer` are included; labels without an answer are excluded. Every JPEG
contains the final annotated student answer, including retained feedback and contains the final annotated student answer, including retained feedback and
extracted handwriting. extracted handwriting.
@@ -92,7 +126,7 @@ Filenames use natural label order, a three-digit minimum sequence number, and a
sanitized label (`001 - Ex 1.jpg`). Numbering prevents filename collisions when sanitized label (`001 - Ex 1.jpg`). Numbering prevents filename collisions when
labels differ only by characters forbidden in filenames. JSON keys retain exact labels differ only by characters forbidden in filenames. JSON keys retain exact
labels. The managed `answers/` directory is replaced on successful generation, labels. The managed `answers/` directory is replaced on successful generation,
so removed/empty answers do not leave stale images; failures preserve the previous so removed answers do not leave stale images; failures preserve the previous
directory. Disabling the option clears this directory on the next `giving-names` directory. Disabling the option clears this directory on the next `giving-names`
run. Separate storage keeps these images out of the personal final-mark stamping run. Separate storage keeps these images out of the personal final-mark stamping
step, which reads only JPEGs directly inside the student's folder. step, which reads only JPEGs directly inside the student's folder.
@@ -116,11 +150,13 @@ Every label in `score.json` has an object containing exactly four fields:
``` ```
- `present`: an answer entry exists for this student and label in the compilation - `present`: an answer entry exists for this student and label in the compilation
data. A supplied answer judged empty still has `present: true`. data. A supplied answer judged empty still has `present: true`. A body removed by
manual annotation edits has `present: false`.
- `not_empty`: the answer was not marked `empty-answer` and was successfully - `not_empty`: the answer was not marked `empty-answer` and was successfully
compiled. Absent answers have `not_empty: false`. When individual export is compiled. Absent answers have `not_empty: false`. When individual export is
enabled, a JPEG is generated if and only if both `present` and `not_empty` are enabled, every `present` answer gets a JPEG, including those with `not_empty: false`.
true. These fields describe the answer regardless of export settings. Image retention alone does not change the empty classification. These fields
describe the answer regardless of export settings.
- `touched`: the answer appears in the compiled filtered `Concat_F.pdf`, using - `touched`: the answer appears in the compiled filtered `Concat_F.pdf`, using
the actual selection including handwriting and selective redo preservation. the actual selection including handwriting and selective redo preservation.
It is not a flag for human edits. Empty and absent answers have `false`. It is not a flag for human edits. Empty and absent answers have `false`.
@@ -151,7 +187,8 @@ The result is one RGB image of variable height, with no page breaks. It contains
all successfully compiled answer blocks, including answers scored 4 with no all successfully compiled answer blocks, including answers scored 4 with no
remaining feedback. “Full” refers to those answer blocks, not the original scan remaining feedback. “Full” refers to those answer blocks, not the original scan
pages or every question in the statement. Missing/unrenderable answers cannot pages or every question in the statement. Missing/unrenderable answers cannot
be included, and the renderer normally suppresses `empty-answer` results. be included. Empty-answer scans are retained individually but excluded from
the combined and filtered documents.
The grouped compiler refuses to publish a new set when compilation is incomplete. The grouped compiler refuses to publish a new set when compilation is incomplete.
The JPEG does not prepend the question, context or model solution PDFs. The JPEG does not prepend the question, context or model solution PDFs.
+139
View File
@@ -0,0 +1,139 @@
import tempfile
import unittest
from pathlib import Path
from unittest.mock import Mock, patch
import pymupdf
from PIL import Image
from copienator import atomic_write_json, configuration, read_json
from copienator.annotation_layout import (
answer_parts, copy_answer_pdf, fingerprint, layout_for, legacy_layout,
move_answer_pdf, save_parts,
)
from copienator.commands import annotating, annotating_by_label, splitting_int
from copienator.commands import reading_grouped_annotations as reader
from copienator.left_margin import analyse_margins
from copienator.pdf_cut import split_pdf
class AnnotationLayoutTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.addCleanup(self.temp.cleanup)
self.root = Path(self.temp.name)
self.pdf = self.root / 'Copies' / 'Copie01.pdf'
self.pdf.parent.mkdir()
with pymupdf.open() as document:
for _ in range(2):
page = document.new_page(width=600, height=800)
page.insert_text((100, 100), 'Answer')
document.save(self.pdf)
def test_all_pages_measured_after_early_no_margin_run(self):
images = [Image.new('RGB', (600, 100)) for _ in range(6)]
values = [10, 10, 40, None, 19.9, 20]
estimates = {id(image): value for image, value in zip(images, values)}
def estimate(image, dpi):
edge = estimates[id(image)]
return {'edge_mm': edge, 'bands': 0 if edge is None else 8,
'band_starts_mm': [] if edge is None else [edge] * 8}
with patch('copienator.left_margin.estimate_page', side_effect=estimate):
result = analyse_margins(images, 200, 600)
self.assertEqual([p['status'] for p in result['pages']],
['low', 'low', 'normal', 'unknown', 'low', 'normal'])
def test_split_records_pages_and_rejects_stale_measurements(self):
schema_path = self.root / 'Cutleft' / 'Copie01_schema.json'
atomic_write_json(schema_path, {
'source_sha256': fingerprint(self.pdf),
'auto_margin': {'pages': [{'page': 1, 'edge_mm': 40}, {'page': 2, 'edge_mm': 10}]},
})
staging = self.root / 'split'
staging.mkdir()
splitting_int._render_split_outputs(self.pdf, [('Ex 1', 0, 0, 10, 0, 10)], staging)
answer = staging / 'Ex 1.pdf'
self.assertEqual([p['page'] for p in answer_parts(answer)], [1, 2])
self.assertEqual(layout_for(answer)['margin_left'], 400)
atomic_write_json(schema_path, {'source_sha256': 'stale', 'auto_margin': {
'pages': [{'page': 1, 'edge_mm': 10}]}})
splitting_int._render_split_outputs(self.pdf, [('Ex 1', 0, 0, 10, 0, 10)], staging)
self.assertEqual(layout_for(answer)['margin_left'], 300)
def test_configurable_margin_unknown_fallback_and_stale_sidecar(self):
self.assertEqual(layout_for(self.pdf)['margin_left'], 300)
save_parts(self.pdf, [{'margin_status': 'normal'}, {'margin_status': 'low'}])
with patch.object(configuration, 'ANNOTATION_LOW_MARGIN_LEFT', 450):
self.assertEqual(layout_for(self.pdf)['margin_left'], 450)
self.pdf.write_bytes(self.pdf.read_bytes() + b'\n')
self.assertEqual(layout_for(self.pdf)['margin_left'], 300)
def test_horizontal_answer_crop_can_remove_an_original_margin(self):
atomic_write_json(self.root / 'Cutleft' / 'Copie01_schema.json', {
'source_sha256': fingerprint(self.pdf),
'auto_margin': {'pages': [{'page': 1, 'edge_mm': 40}, {'page': 2, 'edge_mm': 40}]},
})
staging = self.root / 'horizontal'
staging.mkdir()
# 20% of a 600-point page removes 42.3 mm from the left.
splitting_int._render_split_outputs(self.pdf, [('|Ex 1', 0, 0, 10, 100, 110)], staging)
answer = staging / 'Ex 1.pdf'
self.assertEqual(layout_for(answer)['margin_left'], 400)
self.assertEqual(answer_parts(answer)[0]['margin_mm'], 0)
def test_copy_move_and_vertical_cut_preserve_page_classification(self):
save_parts(self.pdf, [{'margin_status': 'normal'}, {'margin_status': 'low'}])
copied = self.root / 'copied.pdf'
moved = self.root / 'moved.pdf'
copy_answer_pdf(self.pdf, copied)
move_answer_pdf(copied, moved)
self.assertEqual(layout_for(moved)['margin_left'], 400)
first, second = self.root / 'first.pdf', self.root / 'second.pdf'
split_pdf(moved, 50, first, second)
self.assertEqual(layout_for(first)['margin_left'], 300)
self.assertEqual(layout_for(second)['margin_left'], 400)
def render(self, layout):
callback = Mock()
result = {'score': 2, 'feedback': [{'text': 'Feedback', 'box_2d': [10, 50, 30, 100]}]}
with patch.object(annotating, 'render_score_text', return_value=Image.new('RGBA', (200, 40))):
image, header = annotating.compose_label_image(
Image.new('RGB', (1654, 100), 'blue'), 'Ex 1', result, 0,
render_fn=lambda *args, **kwargs: Image.new('RGBA', (200, 10)),
draw_callback=callback, layout=layout)
box = next(call.args[2]['box'] for call in callback.call_args_list if call.args[0] == 'local_rect')
return image, header, box
def test_wider_layout_translates_copy_and_feedback_without_scaling(self):
normal, height, box = self.render(legacy_layout())
low, low_height, low_box = self.render({**legacy_layout(), 'margin_left': 400})
self.assertEqual(low.width, normal.width + 100)
self.assertEqual(height, low_height)
self.assertEqual(low_box, [box[0] + 100, box[1], box[2] + 100, box[3]])
self.assertEqual(low.getpixel((400, height + 80)), (0, 0, 255))
self.assertEqual(low.getpixel((399, height + 80)), (255, 255, 255))
def test_mixed_bgnot_widths_and_saved_geometry_survive_config_change(self):
normal, height, _ = self.render(legacy_layout())
low, _, _ = self.render({**legacy_layout(), 'margin_left': 400})
annotating_by_label.save_batch([
('01', 'Ex 1', normal, height, []), ('02', 'Ex 1', low, height, []),
], 'Ex 1', 1, self.root)
output = self.root / 'Ex 1 G1'
meta = read_json(output / 'bnote.json')
self.assertEqual(meta['width'], low.width)
with Image.open(output / 'Reference.jpg') as reference:
self.assertGreater(min(reference.getpixel((low.width - 10, 80))), 245)
with patch.object(reader, 'detect_checks_and_notes', return_value=([], Image.new('RGBA', (low.width, low.height * 2)))), patch.object(configuration, 'ANNOTATION_LOW_MARGIN_LEFT', 500):
_, notes = reader._scan_annotation_directory(output)
self.assertEqual(notes['02']['Ex 1']['layout']['margin_left'], 400)
rerender, _, _ = self.render(notes['02']['Ex 1']['layout'])
self.assertEqual(rerender.width, low.width)
with patch.object(reader, 'detect_checks_and_notes', return_value=([], None)):
_, notes = reader._scan_annotation_directory(output)
self.assertEqual(notes['02']['Ex 1']['layout']['margin_left'], 400)
self.assertIsNone(notes['02']['Ex 1']['img'])
if __name__ == '__main__':
unittest.main()
+23
View File
@@ -115,6 +115,29 @@ class ModificationTests(unittest.TestCase):
self.assertFalse((self.output / "Empty.jpg").exists()) self.assertFalse((self.output / "Empty.jpg").exists())
self.assertFalse(read_json(self.output / "info.json")["Empty"]["not_empty"]) self.assertFalse(read_json(self.output / "info.json")["Empty"]["not_empty"])
def test_retained_empty_scan_stays_empty_after_unrelated_edits_and_can_be_replaced(self):
self.images["Empty"] = Image.new("RGB", (100, 60), "white")
self.heights["Empty"] = 10
self.info["Empty"]["present"] = True
self.seed()
self.apply("Copie43 Ex 1 -> Ex 2|")
info = read_json(self.output / "info.json")
self.assertTrue((self.output / "Empty.jpg").is_file())
self.assertEqual(info["Empty"], self.info["Empty"])
with Image.open(self.output / "Concat.jpg") as image:
self.assertEqual(image.height, 120 + 180 + 40)
self.apply("Copie43 Ex 1 -> Empty")
self.assertTrue(read_json(self.output / "info.json")["Empty"]["not_empty"])
self.apply()
self.assertEqual(read_json(self.output / "info.json")["Empty"], self.info["Empty"])
def test_removed_body_is_absent_and_not_exported_as_an_empty_scan(self):
self.apply("Copie43 Ex 1 x> Ex 2")
info = read_json(self.output / "info.json")["Ex 1"]
self.assertFalse(info["present"])
self.assertFalse(info["not_empty"])
self.assertFalse((self.output / "Ex 1.jpg").exists())
def test_invalid_file_does_not_apply_valid_first_instruction(self): def test_invalid_file_does_not_apply_valid_first_instruction(self):
before = {p.name: p.read_bytes() for p in self.output.iterdir() if p.is_file()} before = {p.name: p.read_bytes() for p in self.output.iterdir() if p.is_file()}
with self.assertRaises(CliError): with self.assertRaises(CliError):
+85 -8
View File
@@ -57,8 +57,8 @@ class AnswerReturnTests(unittest.TestCase):
), patch.object(annotating, "make_base_image", side_effect=self.supplement): ), patch.object(annotating, "make_base_image", side_effect=self.supplement):
publish_answer_returns(self.root, self.source, self.destination) publish_answer_returns(self.root, self.source, self.destination)
files = sorted((self.destination / "answers").glob("*.jpg")) files = sorted((self.destination / "answers").glob("*.jpg"))
self.assertEqual([p.name for p in files], ["Ex 1.jpg", "Ex 2.jpg"]) self.assertEqual([p.name for p in files], ["Empty.jpg", "Ex 1.jpg", "Ex 2.jpg"])
with Image.open(files[0]) as image: with Image.open(self.destination / "answers" / "Ex 1.jpg") as image:
self.assertEqual(image.size, (100, 30 + 20 * sum((context, question, solution)))) self.assertEqual(image.size, (100, 30 + 20 * sum((context, question, solution))))
colors = [] colors = []
if context: if context:
@@ -212,7 +212,7 @@ class AnswerReturnTests(unittest.TestCase):
) )
self.assertEqual( self.assertEqual(
sorted(path.name for path in answers.iterdir()), sorted(path.name for path in answers.iterdir()),
["Ex 1.jpg", "Ex 2.jpg"], ["Empty.jpg", "Ex 1.jpg", "Ex 2.jpg"],
) )
@@ -304,7 +304,7 @@ class CompiledMembershipTests(unittest.TestCase):
self.assertEqual(info["Empty"], {"present": True, "not_empty": False, "touched": False, "score": "0"}) self.assertEqual(info["Empty"], {"present": True, "not_empty": False, "touched": False, "score": "0"})
self.assertEqual(info["Absent"], {"present": False, "not_empty": False, "touched": False, "score": ""}) self.assertEqual(info["Absent"], {"present": False, "not_empty": False, "touched": False, "score": ""})
self.assertEqual(info["Perfect"], {"present": True, "not_empty": True, "touched": False, "score": "4"}) self.assertEqual(info["Perfect"], {"present": True, "not_empty": True, "touched": False, "score": "4"})
self.assertNotIn("Empty", rendered) self.assertIn("Empty", rendered)
for label in rendered: for label in rendered:
self.assertTrue((output / f"{label}.jpg").is_file()) self.assertTrue((output / f"{label}.jpg").is_file())
# Selective redo keeps unselected saved answers in the PDF, # Selective redo keeps unselected saved answers in the PDF,
@@ -316,6 +316,76 @@ class CompiledMembershipTests(unittest.TestCase):
self.assertTrue(read_json(output / "info.json")["Perfect"]["touched"]) self.assertTrue(read_json(output / "info.json")["Perfect"]["touched"])
self.assertFalse(read_json(output / "info.json")["Empty"]["not_empty"]) self.assertFalse(read_json(output / "info.json")["Empty"]["not_empty"])
def test_real_empty_pdf_is_published_in_all_annotation_modes(self):
import fitz
from copienator.commands import reading_annotations as standard
for mode in ("Anot", "Bnot", "BGnot"):
with self.subTest(mode=mode), tempfile.TemporaryDirectory() as directory:
root = Path(directory)
workspace = EvaluationWorkspace(root)
(root / "labels").write_text("Empty\nAbsent\n")
pdf = root / "answer.pdf"
with fitz.open() as document:
page = document.new_page(width=200, height=60)
page.insert_text((10, 30), "Copied question only")
document.save(pdf)
data = {"01": {"Empty": {"pdf_path": pdf, "coordinates": (0, 0),
"result": {"score": 0, "error": "empty-answer", "feedback": []}}}}
if mode == "Anot":
status = annotating.process_student("01", data["01"], root, ["Empty", "Absent"], True)
self.assertEqual(status, "success")
elif mode == "Bnot":
status = standard.apply_actions_and_regenerate(
workspace, data, "01", [], None, ["Empty", "Absent"])
self.assertEqual(status, 0)
else:
status, _ = reader.apply_actions_and_regenerate_grouped(
workspace, data, "01", [], {}, ["Empty", "Absent"])
self.assertEqual(status, 0)
output = root / mode / "Copie01"
destination = root / "return"
destination.mkdir()
with patch.multiple(configuration, RETURN_ANSWERS_ENABLED=True,
RETURN_ANSWERS_CONTEXT=False, RETURN_ANSWERS_QUESTION=False,
RETURN_ANSWERS_SOLUTION=False):
publish_answer_returns(root, output, destination)
self.assertEqual([p.name for p in (destination / "answers").iterdir()], ["Empty.jpg"])
with Image.open(destination / "answers" / "Empty.jpg") as image:
self.assertGreater(image.height, 0)
self.assertLess(image.convert("L").getextrema()[0], 100)
self.assertFalse(read_json(destination / "info.json")["Empty"]["not_empty"])
self.assertFalse((output / "Concat.jpg").exists())
self.assertFalse((output / "Concat_F.pdf").exists())
def test_bnot_recovers_empty_scans_missing_from_review_layout(self):
from copienator.commands import reading_annotations as standard
for only_empty in (False, True):
with self.subTest(only_empty=only_empty), tempfile.TemporaryDirectory() as directory:
workspace = EvaluationWorkspace(Path(directory))
output = workspace.root / "Bnot" / "Copie01"
output.mkdir(parents=True)
answer = workspace.root / "answer.pdf"
answer.touch()
labels = {"Empty": {"pdf_path": answer, "coordinates": (0, 0),
"result": {"score": 0, "error": "empty-answer"}}}
if not only_empty:
labels["Full"] = {"pdf_path": answer, "coordinates": (0, 0),
"result": {"score": 4}}
atomic_write_json(output / "bnote.json", {"images": [
{"label": "Full", "hmin": 0, "hmax": 50, "header_height": 0}]})
with patch.object(annotating, "make_base_image", return_value=(None, 0, 0)), patch.object(
annotating, "compose_label_image", return_value=(Image.new("RGB", (100, 50)), 0)
) as render:
status = standard.apply_actions_and_regenerate(
workspace, {"01": labels}, "01", [], None, [*labels, "Absent"])
self.assertEqual(status, 0)
self.assertTrue(all(call.kwargs["with_empty"] for call in render.call_args_list))
self.assertTrue((output / "Empty.jpg").is_file())
self.assertFalse((output / "Absent.jpg").exists())
self.assertFalse(read_json(output / "info.json")["Empty"]["not_empty"])
self.assertEqual((output / "Concat.jpg").exists(), not only_empty)
self.assertFalse((output / "Concat_F.jpg").exists())
def test_all_empty_removes_stale_concat_and_records_false(self): def test_all_empty_removes_stale_concat_and_records_false(self):
with tempfile.TemporaryDirectory() as directory: with tempfile.TemporaryDirectory() as directory:
workspace = EvaluationWorkspace(Path(directory)) workspace = EvaluationWorkspace(Path(directory))
@@ -325,10 +395,17 @@ class CompiledMembershipTests(unittest.TestCase):
(output / name).write_bytes(b"stale") (output / name).write_bytes(b"stale")
atomic_write_json(output / "touched.json", {"Empty": True}) atomic_write_json(output / "touched.json", {"Empty": True})
atomic_write_json(output / "answer_labels.json", ["Empty"]) atomic_write_json(output / "answer_labels.json", ["Empty"])
status, _ = reader.apply_actions_and_regenerate_grouped( answer = workspace.root / "answer.pdf"
workspace, {"01": {"Empty": {"result": {"score": 0, "error": "empty-answer"}}}}, answer.touch()
"01", [], {}, ["Empty"] with patch.object(annotating, "make_base_image", return_value=(Image.new("RGB", (100, 50)), 0, 0)), patch.object(
) annotating, "compose_label_image", return_value=(Image.new("RGB", (100, 50)), 0)
):
status, _ = reader.apply_actions_and_regenerate_grouped(
workspace, {"01": {"Empty": {"result": {"score": 0, "error": "empty-answer"},
"pdf_path": answer, "coordinates": (0, 0)}}},
"01", [], {}, ["Empty"]
)
self.assertTrue((output / "Empty.jpg").is_file())
self.assertEqual(status, 0) self.assertEqual(status, 0)
self.assertEqual(read_json(output / "info.json"), { self.assertEqual(read_json(output / "info.json"), {
"Empty": {"present": True, "not_empty": False, "touched": False, "score": "0"} "Empty": {"present": True, "not_empty": False, "touched": False, "score": "0"}
+18
View File
@@ -9,11 +9,29 @@ import pymupdf
from copienator import CliError, EvaluationWorkspace, atomic_write_json, read_json from copienator import CliError, EvaluationWorkspace, atomic_write_json, read_json
from copienator.commands import correction, resolve_manual from copienator.commands import correction, resolve_manual
from copienator.answer_origins import origins_for, save_origins from copienator.answer_origins import origins_for, save_origins
from copienator.annotation_layout import save_parts, answer_parts, layout_for, page_parts
from copienator.annotation_data import load_annotation_data from copienator.annotation_data import load_annotation_data
from copienator_gui.app import has_manual_conflicts from copienator_gui.app import has_manual_conflicts
class ManualResolutionStateTests(unittest.TestCase): class ManualResolutionStateTests(unittest.TestCase):
def test_merge_preserves_low_margin_and_page_order(self):
for pipe_first in (False, True):
with self.subTest(pipe_first=pipe_first), tempfile.TemporaryDirectory() as temporary:
workspace = self.fixture(Path(temporary), '->')
workspace.manual_resolutions_file.write_text(
'Copie01 A -> |B\n' if pipe_first else 'Copie01 A -> B|\n')
copies = workspace.copies_dir / 'Copie01'
for label, status in (('A', 'low'), ('B', 'normal')):
pdf = copies / f'{label}.pdf'
save_parts(pdf, [{'margin_status': status} for _ in page_parts(pdf)])
resolve_manual.resolve_manual(workspace)
target = resolve_manual.get_actual_pdf(workspace.copies_dir, '01', 'B')
self.assertEqual(layout_for(target)['margin_left'], 400)
statuses = [part['margin_status'] for part in answer_parts(target)]
self.assertEqual(statuses[0], 'low' if pipe_first else 'normal')
self.assertEqual(statuses[-1], 'normal' if pipe_first else 'low')
def test_automatic_additional_content_records_origin_but_wrong_label_does_not(self): def test_automatic_additional_content_records_origin_but_wrong_label_does_not(self):
for kind in ("additional-answer", "wrong-label"): for kind in ("additional-answer", "wrong-label"):
with self.subTest(kind=kind), tempfile.TemporaryDirectory() as temporary: with self.subTest(kind=kind), tempfile.TemporaryDirectory() as temporary: