Compare commits
2
Commits
b1f1e0bc86
...
78a7b7679c
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
78a7b7679c | ||
|
|
e7094d850e |
No files matched your search
@@ -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}
|
||||
@@ -155,7 +155,7 @@ def output_removals(directory: Path) -> tuple[str, ...]:
|
||||
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",
|
||||
*(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,
|
||||
@@ -169,7 +169,15 @@ def rebuild(root: Path, mode: str, directory: Path, staging: Path, meta: dict,
|
||||
|
||||
headers, bodies, changed = edited_bodies(directory, meta, instructions)
|
||||
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)
|
||||
images, groups = [], []
|
||||
seen_contexts = set()
|
||||
@@ -187,15 +195,13 @@ def rebuild(root: Path, mode: str, directory: Path, staging: Path, meta: dict,
|
||||
fill="black", font_size=30)
|
||||
image = stack([header, body])
|
||||
image.save(staging / f"{label}.jpg")
|
||||
if info[label]["not_empty"]:
|
||||
images.append(image)
|
||||
if label in filtered:
|
||||
extras = get_extra_pdfs_as_images(root, label, annotating, all_labels,
|
||||
seen_contexts=seen_contexts) if mode == "BGnot" else []
|
||||
groups.append([*extras, image])
|
||||
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["score"] = scores.get(label, entry["score"])
|
||||
atomic_write_json(staging / "info.json", info)
|
||||
|
||||
@@ -35,9 +35,9 @@ from copienator.commands.import_annotations import clear_import_directory
|
||||
from copienator.answer_info import build_answer_info
|
||||
from copienator.filesystem import staged_directory
|
||||
from copienator.feedback_boxes import valid_feedback_box
|
||||
from copienator.annotation_layout import layout_for
|
||||
from copienator.utils import natural_key
|
||||
|
||||
MARGIN_LEFT = 300
|
||||
ANNOT_WIDTH = 600
|
||||
|
||||
|
||||
@@ -321,7 +321,7 @@ def compose_label_image(base_img, label, result, hmin,
|
||||
with_error=True,
|
||||
with_empty=False,
|
||||
more_right=False,
|
||||
with_id=None, source_pdf=None):
|
||||
with_id=None, source_pdf=None, layout=None):
|
||||
"""
|
||||
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}"
|
||||
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
|
||||
if base_img.width < TARGET_MIN_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.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
|
||||
|
||||
# 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
|
||||
current_y = 0
|
||||
@@ -431,7 +433,7 @@ def compose_label_image(base_img, label, result, hmin,
|
||||
|
||||
# Paste Base Image
|
||||
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
|
||||
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_ymax = (ymax - hmin) + image_offset_y
|
||||
target_xmin = xmin + MARGIN_LEFT + left_pad
|
||||
target_xmax = xmax + MARGIN_LEFT + left_pad
|
||||
target_xmin = xmin + margin_left + left_pad
|
||||
target_xmax = xmax + margin_left + left_pad
|
||||
|
||||
# Draw Rectangle (if not suppressed)
|
||||
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
|
||||
|
||||
final_img.info['annotation_layout'] = layout
|
||||
return final_img, header_height
|
||||
|
||||
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:
|
||||
max_w = max(image.width for image in label_images)
|
||||
total_h = sum(image.height for image in label_images)
|
||||
canvas = Image.new('RGB', (max_w, total_h))
|
||||
canvas = Image.new('RGB', (max_w, total_h), 'white')
|
||||
current_y = 0
|
||||
for image in label_images:
|
||||
canvas.paste(image, (0, current_y))
|
||||
current_y += image.height
|
||||
canvas.save(staging / "Concat.jpg")
|
||||
elif labels_data:
|
||||
problems = True
|
||||
# An all-empty copy still has valid individual return images.
|
||||
return "partial" if problems else "success"
|
||||
|
||||
|
||||
|
||||
@@ -87,6 +87,7 @@ def save_batch(batch, prefix, group_id, output_root: Path) -> None:
|
||||
"id": student_id,
|
||||
"label": label,
|
||||
"header_height": header_height,
|
||||
"layout": image.info.get('annotation_layout', annotating.layout_for(None)),
|
||||
"hmin": current_y,
|
||||
"hmax": current_y + image.height,
|
||||
}
|
||||
|
||||
@@ -196,6 +196,7 @@ def _render_student(
|
||||
"id": student_id,
|
||||
"label": label,
|
||||
"header_height": header_height,
|
||||
"layout": final_image.info.get('annotation_layout', annotating.layout_for(None)),
|
||||
"img_h": final_image.height,
|
||||
}
|
||||
)
|
||||
|
||||
@@ -7,7 +7,6 @@ import json
|
||||
import os
|
||||
import signal
|
||||
import shlex
|
||||
import shutil
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
@@ -20,6 +19,7 @@ from google.genai import types
|
||||
from copienator import configuration as config
|
||||
from copienator.commands import grouping
|
||||
from copienator.answer_origins import save_origins
|
||||
from copienator.annotation_layout import copy_answer_pdf, move_answer_pdf
|
||||
from copienator import prompting
|
||||
from copienator.gemini_config import fresh_seed, thinking_config
|
||||
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}.")
|
||||
|
||||
# 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")
|
||||
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)
|
||||
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"
|
||||
|
||||
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])
|
||||
tprint(f"\t\tCopying Copie{pid} : {label} -> {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":
|
||||
old_path = pdf_path.with_name(f"{label}_old.pdf")
|
||||
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
|
||||
current_suffix = "_old"
|
||||
|
||||
@@ -729,10 +729,10 @@ def resolve_delayed_moves():
|
||||
res["suffix"] = "_old" # Fixed typo: was suffixe
|
||||
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")
|
||||
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)
|
||||
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}")
|
||||
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])
|
||||
idx = reserve_next_group_idx(target_label)
|
||||
height = grouping.get_pdf_height(str(add_pdf_path))
|
||||
|
||||
@@ -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.configuration import SHOW_PERSONAL_STEPS
|
||||
from copienator.left_margin import analyse_margins
|
||||
from copienator.annotation_layout import fingerprint
|
||||
|
||||
DELIMITER_WIDTH = 5
|
||||
DELIMITER_COLOR = (0, 0, 0)
|
||||
@@ -211,6 +212,8 @@ def save_results(
|
||||
) -> None:
|
||||
"""Atomically replace every Cutleft output associated with one copy."""
|
||||
_, splits, schema = result
|
||||
if pdf_path.is_file():
|
||||
schema = {**schema, 'source_sha256': fingerprint(pdf_path)}
|
||||
base_name = pdf_path.stem
|
||||
previous = _previous_cutleft_outputs(output_dir, base_name)
|
||||
with staged_files(output_dir, remove=previous) as staging:
|
||||
|
||||
@@ -22,6 +22,7 @@ from copienator import (
|
||||
)
|
||||
from copienator.annotation_actions import apply_checkbox_actions, apply_score_overrides
|
||||
from copienator.annotation_data import AnnotationData, load_annotation_data
|
||||
from copienator.annotation_layout import legacy_layout
|
||||
from copienator.annotation_modifications import (
|
||||
BASELINE,
|
||||
output_removals,
|
||||
@@ -157,10 +158,14 @@ def apply_actions_and_regenerate(
|
||||
"""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():
|
||||
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}")
|
||||
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):
|
||||
raise TypeError(f"Expected a JSON object in {bnote_path}")
|
||||
|
||||
@@ -180,7 +185,14 @@ def apply_actions_and_regenerate(
|
||||
filtered: list[Image.Image] = []
|
||||
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):
|
||||
incomplete = True
|
||||
continue
|
||||
@@ -191,11 +203,10 @@ def apply_actions_and_regenerate(
|
||||
content = labels_data[label]
|
||||
result = content["result"]
|
||||
scores[label] = str(result.get("score", 0))
|
||||
if result.get("error") == "empty-answer":
|
||||
continue
|
||||
is_empty = result.get("error") == "empty-answer"
|
||||
|
||||
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))
|
||||
hmax = int(image_info.get("hmax", 0))
|
||||
sub_note = notes_layer.crop((0, hmin, notes_layer.width, hmax))
|
||||
@@ -213,6 +224,9 @@ def apply_actions_and_regenerate(
|
||||
result,
|
||||
content["coordinates"][0],
|
||||
with_error=False,
|
||||
with_empty=True,
|
||||
source_pdf=pdf_path,
|
||||
layout=image_info.get('layout', legacy_layout()),
|
||||
)
|
||||
if final_image is None:
|
||||
incomplete = True
|
||||
@@ -230,6 +244,8 @@ def apply_actions_and_regenerate(
|
||||
|
||||
dirty_images[label] = final_image
|
||||
header_heights[label] = new_header_height
|
||||
if is_empty:
|
||||
continue
|
||||
answer_labels.append(label)
|
||||
concatenated.append(final_image)
|
||||
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
|
||||
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():
|
||||
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}")
|
||||
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:
|
||||
actions, notes = ([], None) if only_empty else detect_checks_and_notes(output_dir)
|
||||
if notes is None and not actions and not update_score and not only_empty:
|
||||
print(" No readable annotation input found.")
|
||||
status = ExitCode.PARTIAL
|
||||
continue
|
||||
|
||||
@@ -25,6 +25,7 @@ from copienator import (
|
||||
)
|
||||
from copienator.annotation_actions import apply_checkbox_actions, apply_score_overrides
|
||||
from copienator.annotation_data import AnnotationData, RefaireList, load_annotation_data
|
||||
from copienator.annotation_layout import legacy_layout
|
||||
from copienator.annotation_modifications import (
|
||||
BASELINE,
|
||||
baseline,
|
||||
@@ -168,7 +169,6 @@ def _scan_annotation_directory(
|
||||
if notes_image is None:
|
||||
if required:
|
||||
raise ValueError(f"Could not read annotations in {directory}")
|
||||
return {}, {}
|
||||
actions_by_student: dict[str, list[dict[str, Any]]] = defaultdict(list)
|
||||
notes_by_student: dict[str, LabelNotes] = defaultdict(dict)
|
||||
for action in actions:
|
||||
@@ -181,11 +181,12 @@ def _scan_annotation_directory(
|
||||
hmin = int(image_info.get("hmin", 0))
|
||||
hmax = int(image_info.get("hmax", 0))
|
||||
if student_id and label and hmax > hmin:
|
||||
crop = notes_image.crop((0, hmin, notes_image.width, hmax))
|
||||
if has_significant_notes(crop):
|
||||
crop = (notes_image.crop((0, hmin, notes_image.width, hmax))
|
||||
if notes_image is not None else None)
|
||||
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)),
|
||||
"layout": image_info.get('layout', legacy_layout()),
|
||||
}
|
||||
return dict(actions_by_student), dict(notes_by_student)
|
||||
|
||||
@@ -292,8 +293,7 @@ def apply_actions_and_regenerate_grouped(
|
||||
result["score"] = old_scores[label]
|
||||
scores[label] = str(result.get("score", 0))
|
||||
touched[label] = False
|
||||
if result.get("error") == "empty-answer":
|
||||
continue
|
||||
is_empty = result.get("error") == "empty-answer"
|
||||
saved_image = output_dir / f"{label}.jpg"
|
||||
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):
|
||||
@@ -319,6 +319,8 @@ def apply_actions_and_regenerate_grouped(
|
||||
)
|
||||
dirty_images[label] = final_image
|
||||
scores[label] = str(old_scores.get(label, scores[label]))
|
||||
if is_empty:
|
||||
continue
|
||||
concat_images.append(final_image)
|
||||
answer_labels.append(label)
|
||||
filtered_labels.append(label)
|
||||
@@ -344,6 +346,9 @@ def apply_actions_and_regenerate_grouped(
|
||||
result,
|
||||
content["coordinates"][0],
|
||||
with_error=False,
|
||||
with_empty=True,
|
||||
source_pdf=pdf_path,
|
||||
layout=label_notes.get(label, {}).get('layout', legacy_layout()),
|
||||
)
|
||||
if final_image is None:
|
||||
incomplete = True
|
||||
@@ -368,6 +373,8 @@ def apply_actions_and_regenerate_grouped(
|
||||
# Persist every final block, including unchanged answers, for returns.
|
||||
dirty_images[label] = final_image
|
||||
header_heights[label] = new_header_height
|
||||
if is_empty:
|
||||
continue
|
||||
answer_labels.append(label)
|
||||
concat_images.append(final_image)
|
||||
|
||||
|
||||
@@ -2,7 +2,6 @@ from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import re
|
||||
import shutil
|
||||
from collections.abc import Sequence
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
@@ -10,6 +9,8 @@ from typing import Any
|
||||
|
||||
from pypdf import PdfWriter
|
||||
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 import (
|
||||
@@ -240,6 +241,8 @@ def resolve_manual(workspace: EvaluationWorkspace) -> ExitCode:
|
||||
if not instruction.cut and instruction.operator == "->":
|
||||
origins[key_new] = list(dict.fromkeys([
|
||||
*origins[key_new], instruction.old_label]))
|
||||
source_parts = page_parts(source)
|
||||
destination_parts = page_parts(destination)
|
||||
writer = PdfWriter()
|
||||
try:
|
||||
if instruction.pipe_first:
|
||||
@@ -251,8 +254,11 @@ def resolve_manual(workspace: EvaluationWorkspace) -> ExitCode:
|
||||
writer.append(destination)
|
||||
writer.append(source)
|
||||
writer.write(temp_output)
|
||||
save_parts(temp_output, (source_parts + destination_parts
|
||||
if instruction.pipe_first else destination_parts + source_parts))
|
||||
except Exception:
|
||||
temp_output.unlink(missing_ok=True)
|
||||
metadata_path(temp_output).unlink(missing_ok=True)
|
||||
raise
|
||||
finally:
|
||||
writer.close()
|
||||
@@ -264,12 +270,13 @@ def resolve_manual(workspace: EvaluationWorkspace) -> ExitCode:
|
||||
[instruction.old_label] if instruction.operator == "-x"
|
||||
and not instruction.cut else []
|
||||
)
|
||||
shutil.copy(source, temp_output)
|
||||
copy_answer_pdf(source, temp_output)
|
||||
current_paths[key_new] = temp_output
|
||||
temp_files.append(temp_output)
|
||||
except Exception:
|
||||
for temporary in temp_files:
|
||||
temporary.unlink(missing_ok=True)
|
||||
metadata_path(temporary).unlink(missing_ok=True)
|
||||
raise
|
||||
|
||||
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")
|
||||
if pdf != old_name:
|
||||
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")
|
||||
|
||||
for instruction in instructions:
|
||||
@@ -298,7 +305,7 @@ def resolve_manual(workspace: EvaluationWorkspace) -> ExitCode:
|
||||
continue
|
||||
final_name = workspace.copies_dir / f"Copie{copy_id}" / f"{label}_new.pdf"
|
||||
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)]
|
||||
if (copy_id, origin) not in initial_paths
|
||||
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:
|
||||
if temporary not in used_temps:
|
||||
temporary.unlink(missing_ok=True)
|
||||
metadata_path(temporary).unlink(missing_ok=True)
|
||||
|
||||
for copy_id, label, target in dismissed_conflicts:
|
||||
# A removed instruction acknowledges the conflict without touching PDFs.
|
||||
|
||||
@@ -22,6 +22,7 @@ from copienator import (
|
||||
workspace_from_target,
|
||||
)
|
||||
from copienator.filesystem import staged_directory
|
||||
from copienator.annotation_layout import fingerprint, metadata_path, save_parts
|
||||
|
||||
SQUARE = 1000 // 38
|
||||
ANSWER_TOP_PADDING_POINTS = 4 * 72 / 25.4
|
||||
@@ -129,6 +130,11 @@ def _render_split_outputs(
|
||||
visible_bounds = _prepare_split_pages(document)
|
||||
parsed = _parse_coordinates(coords_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:
|
||||
temporary = Path(temp_directory)
|
||||
for index, item in enumerate(parsed):
|
||||
@@ -200,6 +206,15 @@ def _render_split_outputs(
|
||||
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()
|
||||
for label, parts in parts_by_label.items():
|
||||
@@ -212,6 +227,8 @@ def _render_split_outputs(
|
||||
finally:
|
||||
merger.close()
|
||||
generated.add(filename)
|
||||
save_parts(staging / filename, provenance[label])
|
||||
generated.add(metadata_path(staging / filename).name)
|
||||
return generated
|
||||
finally:
|
||||
document.close()
|
||||
|
||||
@@ -37,6 +37,8 @@ THINKING_LEVEL_FOR_LABEL = "medium"
|
||||
THINKING_LEVEL_FLASH = "medium"
|
||||
THINKING_LEVEL_PRO = "high"
|
||||
ALWAYS_CROP = False
|
||||
ANNOTATION_MARGIN_LEFT = 300
|
||||
ANNOTATION_LOW_MARGIN_LEFT = 400
|
||||
SHOW_PERSONAL_STEPS = False
|
||||
RETURN_JPEG_ENABLED = True
|
||||
RETURN_PDF_ENABLED = True
|
||||
|
||||
@@ -144,7 +144,15 @@ def analyse_margins(images: list[Image.Image], render_dpi: int, reference_width:
|
||||
starts[run_start:] = [0] * (len(images) - run_start)
|
||||
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,
|
||||
'discard_left_fraction': DISCARD_LEFT_FRACTION, 'cut_fraction_of_margin': .5,
|
||||
'standard_margin_mm': STANDARD_MARGIN_MM, 'minimum_margin_mm': MIN_MARGIN_MM,
|
||||
|
||||
@@ -6,6 +6,7 @@ from pathlib import Path
|
||||
import pymupdf
|
||||
|
||||
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
|
||||
# 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:
|
||||
"""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:
|
||||
index, offset = cut_position([page.rect.height for page in document], percent)
|
||||
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)
|
||||
first.save(first_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)
|
||||
@@ -84,7 +84,7 @@ def publish_answer_returns(
|
||||
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"]]
|
||||
labels = [label for label, entry in info.items() if entry["present"]]
|
||||
|
||||
# Import the rendering backend only when individual images are requested.
|
||||
from copienator.commands.annotating import make_base_image
|
||||
|
||||
@@ -6,6 +6,10 @@ API_KEY = os.environ.get("GEMINI_API_KEY")
|
||||
EXPORT_DIR = Path("Export")
|
||||
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).
|
||||
RETURN_JPEG_ENABLED = True
|
||||
RETURN_PDF_ENABLED = True
|
||||
|
||||
+46
-9
@@ -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
|
||||
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
|
||||
|
||||
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. |
|
||||
| `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. |
|
||||
| `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
|
||||
|
||||
@@ -69,9 +103,9 @@ and preserves return PDFs when present.
|
||||
## Individual answer JPEGs
|
||||
|
||||
With `RETURN_ANSWERS_ENABLED = True`, `giving-names` generates an `answers/`
|
||||
subdirectory inside each student's return folder. It includes **every non-empty
|
||||
compiled answer**, even a perfect answer omitted from the filtered PDF. Labels
|
||||
marked `empty-answer` and labels without an answer are excluded. Every JPEG
|
||||
subdirectory inside each student's return folder. It includes **every present
|
||||
compiled answer**, even a perfect answer omitted from the filtered PDF. Scans
|
||||
marked `empty-answer` are included; labels without an answer are excluded. Every JPEG
|
||||
contains the final annotated student answer, including retained feedback and
|
||||
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
|
||||
labels differ only by characters forbidden in filenames. JSON keys retain exact
|
||||
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`
|
||||
run. Separate storage keeps these images out of the personal final-mark stamping
|
||||
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
|
||||
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
|
||||
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
|
||||
true. These fields describe the answer regardless of export settings.
|
||||
enabled, every `present` answer gets a JPEG, including those with `not_empty: false`.
|
||||
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
|
||||
the actual selection including handwriting and selective redo preservation.
|
||||
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
|
||||
remaining feedback. “Full” refers to those answer blocks, not the original scan
|
||||
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 JPEG does not prepend the question, context or model solution PDFs.
|
||||
|
||||
@@ -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()
|
||||
@@ -115,6 +115,29 @@ class ModificationTests(unittest.TestCase):
|
||||
self.assertFalse((self.output / "Empty.jpg").exists())
|
||||
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):
|
||||
before = {p.name: p.read_bytes() for p in self.output.iterdir() if p.is_file()}
|
||||
with self.assertRaises(CliError):
|
||||
|
||||
@@ -57,8 +57,8 @@ 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], ["Ex 1.jpg", "Ex 2.jpg"])
|
||||
with Image.open(files[0]) as image:
|
||||
self.assertEqual([p.name for p in files], ["Empty.jpg", "Ex 1.jpg", "Ex 2.jpg"])
|
||||
with Image.open(self.destination / "answers" / "Ex 1.jpg") as image:
|
||||
self.assertEqual(image.size, (100, 30 + 20 * sum((context, question, solution))))
|
||||
colors = []
|
||||
if context:
|
||||
@@ -212,7 +212,7 @@ class AnswerReturnTests(unittest.TestCase):
|
||||
)
|
||||
self.assertEqual(
|
||||
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["Absent"], {"present": False, "not_empty": False, "touched": False, "score": ""})
|
||||
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:
|
||||
self.assertTrue((output / f"{label}.jpg").is_file())
|
||||
# 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.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):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
workspace = EvaluationWorkspace(Path(directory))
|
||||
@@ -325,10 +395,17 @@ class CompiledMembershipTests(unittest.TestCase):
|
||||
(output / name).write_bytes(b"stale")
|
||||
atomic_write_json(output / "touched.json", {"Empty": True})
|
||||
atomic_write_json(output / "answer_labels.json", ["Empty"])
|
||||
answer = workspace.root / "answer.pdf"
|
||||
answer.touch()
|
||||
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"}}}},
|
||||
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(read_json(output / "info.json"), {
|
||||
"Empty": {"present": True, "not_empty": False, "touched": False, "score": "0"}
|
||||
|
||||
@@ -9,11 +9,29 @@ import pymupdf
|
||||
from copienator import CliError, EvaluationWorkspace, atomic_write_json, read_json
|
||||
from copienator.commands import correction, resolve_manual
|
||||
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_gui.app import has_manual_conflicts
|
||||
|
||||
|
||||
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):
|
||||
for kind in ("additional-answer", "wrong-label"):
|
||||
with self.subTest(kind=kind), tempfile.TemporaryDirectory() as temporary:
|
||||
|
||||
Reference in new issue
Block a user