When no margin detected, reserve a bit more space for comments
This commit is contained in:
1 parent
e7094d850e
commit
78a7b7679c
17 files changed
+354
-25
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}
|
||||
@@ -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,7 +547,7 @@ 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))
|
||||
|
||||
@@ -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,
|
||||
@@ -224,6 +225,8 @@ def apply_actions_and_regenerate(
|
||||
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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -346,6 +347,8 @@ def apply_actions_and_regenerate_grouped(
|
||||
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
|
||||
|
||||
@@ -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)
|
||||
@@ -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
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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()
|
||||
@@ -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