Bnot : split large copies into pieces.

This commit is contained in:
sebastien committed 2026-10-23 08:32:17 +02:00
1 parent 834ffe32f2
commit 1959c5d3bb
10 files changed
+333 -52

No files matched your search

+11
View File
@@ -235,6 +235,13 @@ OU
=python -m copienator annotate-checks Interro/Copies/Copie01.pdf=.
Le mode =--refaire= exige un fichier =refaire.json= et écrit dans
=BRnot=.
Les documents sont limités à 32 000 pixels de haut, comme pour =BGnot=.
Les grandes copies sont découpées entre les labels en =Part1=, =Part2=,
etc., dans leur dossier =CopieXX=. L'export les nomme =CopieXX Part1.pdf=,
etc. Importer toutes les parties avant de lancer =read-annotations= :
elles sont réunies dans les résultats habituels de la copie. Une réponse
individuelle dépassant la limite produit une erreur explicite.
OU
2. =python -m copienator annotate-grouped Interro= dans =BGnot=
@@ -242,6 +249,10 @@ OU
réponses par question.
Enregistrées dans =BGnot=
Les groupes visent une hauteur maximale de 32 000 pixels. Si le bloc
d'un seul élève dépasse cette limite, il reste entier dans un groupe
dédié, même lorsqu'il ne contient qu'une seule réponse.
_Needs_ : label_groups file (made automatically by this function),
qui dit quelles questions regrouper.
+27
View File
@@ -0,0 +1,27 @@
"""Locate tablet documents, including the parts of a large checked copy."""
from pathlib import Path
from copienator.json_io import read_json
MAX_HEIGHT_PX = 32000
def copy_documents(directory: Path) -> list[Path]:
try:
metadata = read_json(directory / "bnote.json", default={})
except (OSError, ValueError):
# Ordinary documents need no manifest. Let the reader report bad metadata.
return [directory]
parts = metadata.get("parts", []) if isinstance(metadata, dict) else []
return [directory / name for name in parts] if parts else [directory]
def tablet_documents(root: Path) -> list[tuple[str, Path]]:
documents = []
for directory in sorted(root.iterdir()):
if not directory.is_dir():
continue
for part in copy_documents(directory):
name = directory.name if part == directory else f"{directory.name} {part.name}"
documents.append((name, part))
return documents
+8 -10
View File
@@ -23,13 +23,12 @@ from copienator import (
workspace_from_args,
)
from copienator.annotation_data import load_annotation_data
from copienator.annotation_documents import MAX_HEIGHT_PX
from copienator.commands import annotating, annotating_with_checks
from copienator.commands.import_annotations import clear_import_directory
from copienator.filesystem import staged_directory
from copienator.utils import natural_key
MAX_HEIGHT_PX = 32000
def render_item(item):
student_id, label, content = item
@@ -65,10 +64,10 @@ def save_batch(batch, prefix, group_id, output_root: Path) -> None:
print(f"Generating group PDF: {prefix} G{group_id} ({len(batch)} elements)")
max_width = max(item[2].width for item in batch)
total_height = sum(item[2].height for item in batch)
if total_height > MAX_HEIGHT_PX:
if total_height > MAX_HEIGHT_PX and len({item[0] for item in batch}) > 1:
raise CliError(
f"{prefix} G{group_id} would be {total_height} px high; "
f"the maximum is {MAX_HEIGHT_PX} px."
f"only a single student's block may exceed {MAX_HEIGHT_PX} px."
)
concatenated = Image.new("RGB", (max_width, total_height), "white")
draw = ImageDraw.Draw(concatenated)
@@ -293,12 +292,11 @@ def split_batches(rendered):
student_height = sum(item[2].height for item in student_items)
if student_height > MAX_HEIGHT_PX:
labels = ", ".join(item[1] for item in student_items)
raise CliError(
f"Copie{student_id} ({labels}) is {student_height} px high; "
f"one student's labels cannot be split across groups and the "
f"maximum group height is {MAX_HEIGHT_PX} px."
)
# Oversized blocks remain intact in a group of their own.
batches.append(list(student_items))
heights.append(student_height)
index = end
continue
# Reuse space in earlier groups, keeping each student's block intact.
for batch_index, height in enumerate(heights):
+64 -22
View File
@@ -26,6 +26,7 @@ from copienator import (
workspace_from_target,
)
from copienator.annotation_data import load_annotation_data
from copienator.annotation_documents import MAX_HEIGHT_PX, copy_documents
from copienator.commands import annotating
from copienator.commands.import_annotations import clear_import_directory
from copienator.filesystem import staged_directory
@@ -143,7 +144,11 @@ class CheckboxRenderer:
def _output_complete(output_dir: Path) -> bool:
return all((output_dir / name).is_file() for name in EXPECTED_OUTPUTS)
return all(
(document / name).is_file()
for document in copy_documents(output_dir)
for name in EXPECTED_OUTPUTS
)
def _render_student(
@@ -197,7 +202,6 @@ def _render_student(
"label": label,
"header_height": header_height,
"layout": final_image.info.get('annotation_layout', annotating.layout_for(None)),
"img_h": final_image.height,
}
)
@@ -205,6 +209,46 @@ def _render_student(
print(f"Warning: no annotations could be rendered for Copie{student_id}")
return "partial"
batches = []
start = 0
height = 0
for index, image in enumerate(label_images):
if image.height > MAX_HEIGHT_PX:
raise CliError(
f"Copie{student_id} ({bnote_entries[index]['label']}) is "
f"{image.height} px high; a single answer cannot be split and "
f"the maximum document height is {MAX_HEIGHT_PX} px."
)
if height + image.height > MAX_HEIGHT_PX:
batches.append((start, index))
start = index
height = 0
height += image.height
batches.append((start, len(label_images)))
with staged_directory(output_dir) as staging:
parts = []
for index, (start, end) in enumerate(batches, start=1):
destination = staging
if len(batches) > 1:
name = f"Part{index}"
parts.append(name)
destination /= name
_save_document(destination, label_images[start:end],
checkbox_groups[start:end], bnote_entries[start:end])
if parts:
atomic_write_json(staging / "bnote.json", {
"parts": parts, "images": bnote_entries,
})
return "partial" if problems else "success"
def _save_document(
output_dir: Path,
label_images: list[Image.Image],
checkbox_groups: list[list[dict[str, Any]]],
bnote_entries: list[dict[str, Any]],
) -> None:
max_width = max(image.width for image in label_images)
total_height = sum(image.height for image in label_images)
concatenated = Image.new("RGB", (max_width, total_height), "white")
@@ -216,7 +260,6 @@ def _render_student(
concatenated.paste(image, (0, current_y))
bnote_entries[index]["hmin"] = current_y
bnote_entries[index]["hmax"] = current_y + image.height
del bnote_entries[index]["img_h"]
for item in checkboxes:
box = item.get("final_box") or item.get("rel_box")
item["global_box"] = [
@@ -228,25 +271,24 @@ def _render_student(
checkbox_map.append(item)
current_y += image.height
with staged_directory(output_dir) as staging:
atomic_write_json(
staging / "bnote.json",
{"width": max_width, "height": total_height, "images": bnote_entries},
)
atomic_write_json(staging / "checkboxes.json", checkbox_map)
reference = staging / "Reference.jpg"
concatenated.save(reference, quality=90)
pdf_path = staging / "Concat.pdf"
pdf_canvas = canvas.Canvas(str(pdf_path), pagesize=(max_width, total_height))
pdf_canvas.drawImage(
str(reference),
0,
0,
width=max_width,
height=total_height,
)
pdf_canvas.save()
return "partial" if problems else "success"
output_dir.mkdir(parents=True, exist_ok=True)
atomic_write_json(
output_dir / "bnote.json",
{"width": max_width, "height": total_height, "images": bnote_entries},
)
atomic_write_json(output_dir / "checkboxes.json", checkbox_map)
reference = output_dir / "Reference.jpg"
concatenated.save(reference, quality=90)
pdf_path = output_dir / "Concat.pdf"
pdf_canvas = canvas.Canvas(str(pdf_path), pagesize=(max_width, total_height))
pdf_canvas.drawImage(
str(reference),
0,
0,
width=max_width,
height=total_height,
)
pdf_canvas.save()
def _copy_id_from_target(workspace: EvaluationWorkspace, target: Path) -> str | None:
+5 -2
View File
@@ -11,6 +11,7 @@ from copienator import (
workspace_from_args,
)
from copienator.configuration import EXPORT_DIR
from copienator.annotation_documents import tablet_documents
from copienator.platform import replace_with_link_or_copy
ANNOTATION_DIRECTORIES = ("BGnot", "Bnot", "Anot")
@@ -36,7 +37,9 @@ def export_directory(
subdirs = [directory for directory in subdirs if not directory.name.startswith("Copie")]
missing_outputs = 0
for subdir in subdirs:
for name, subdir in tablet_documents(source_dir):
if subdir not in subdirs and subdir.parent not in subdirs:
continue
concat_file = next(
(
candidate
@@ -52,7 +55,7 @@ def export_directory(
)
missing_outputs += 1
continue
destination = sync_dir / f"{prefix}{subdir.name}{concat_file.suffix.lower()}"
destination = sync_dir / f"{prefix}{name}{concat_file.suffix.lower()}"
method = replace_with_link_or_copy(concat_file, destination, prefer="hardlink")
print(f"Exported: {destination} ({method})")
return ExitCode.PARTIAL if missing_outputs else ExitCode.SUCCESS
+6 -3
View File
@@ -13,6 +13,7 @@ from copienator import (
workspace_from_args,
)
from copienator import configuration
from copienator.annotation_documents import tablet_documents
ANNOTATION_DIRECTORIES = ("BGnot", "Bnot", "Anot")
IMPORT_DIR = getattr(configuration, "IMPORT_DIR", None)
@@ -58,6 +59,7 @@ def sync_annotated(
return ExitCode.INVALID_WORKSPACE
missing_targets = 0
targets = dict(tablet_documents(annotation_dir))
annotated_files = sorted(
(
path
@@ -70,10 +72,11 @@ def sync_annotated(
if prefix and not annotated_file.stem.startswith(prefix):
print(f"Ignoring return from another pass: {annotated_file.name}")
continue
target_subdir = annotation_dir / annotated_file.stem.removeprefix(prefix)
name = annotated_file.stem.removeprefix(prefix)
target_subdir = targets.get(name)
if not target_subdir.is_dir():
print(f"Warning: directory not found: {target_subdir}", file=sys.stderr)
if target_subdir is None or not target_subdir.is_dir():
print(f"Warning: annotation document not found: {name}", file=sys.stderr)
missing_targets += 1
else:
suffix = annotated_file.suffix.lower()
+28 -3
View File
@@ -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_documents import copy_documents
from copienator.annotation_layout import legacy_layout
from copienator.annotation_modifications import (
BASELINE,
@@ -154,6 +155,7 @@ def apply_actions_and_regenerate(
all_labels: list[str],
*,
update_score: bool = False,
label_notes: dict[str, Image.Image] | None = None,
) -> ExitCode:
"""Apply annotations and atomically merge the regenerated student files."""
output_dir = workspace.annotation_dir("checks") / f"Copie{student_id}"
@@ -205,7 +207,7 @@ def apply_actions_and_regenerate(
scores[label] = str(result.get("score", 0))
is_empty = result.get("error") == "empty-answer"
sub_note = None
sub_note = label_notes.get(label) if label_notes is not None else 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))
@@ -325,11 +327,33 @@ def run(
status = ExitCode.PARTIAL
continue
print(f"Processing annotations for: {student_id}")
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:
documents = copy_documents(output_dir)
label_notes = None
actions, notes = [], None
if not only_empty and documents == [output_dir]:
actions, notes = detect_checks_and_notes(output_dir)
elif not only_empty:
label_notes = {}
incomplete = False
for document in documents:
part_actions, part_notes = detect_checks_and_notes(document)
if part_notes is None:
incomplete = True
continue
actions.extend(part_actions)
for item in read_json(document / "bnote.json")["images"]:
label_notes[item["label"]] = part_notes.crop(
(0, item["hmin"], part_notes.width, item["hmax"])
)
if incomplete:
print(" Missing a readable annotation part; copy was preserved.")
status = ExitCode.PARTIAL
continue
if notes is None and not label_notes and not actions and not update_score and not only_empty:
print(" No readable annotation input found.")
status = ExitCode.PARTIAL
continue
extra = {"label_notes": label_notes} if label_notes is not None else {}
result = apply_actions_and_regenerate(
workspace,
loaded.data,
@@ -338,6 +362,7 @@ def run(
notes,
all_labels,
update_score=update_score,
**extra,
)
if result != ExitCode.SUCCESS:
status = ExitCode.PARTIAL
@@ -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_documents import copy_documents, tablet_documents
from copienator.annotation_layout import legacy_layout
from copienator.annotation_modifications import (
BASELINE,
@@ -463,11 +464,10 @@ def _scan_redo_annotations(
incomplete: set[str] = set()
plans = []
required = ("checkboxes.json", "Reference.jpg", "Concat_annotated.pdf")
for path in sorted(directory.iterdir()):
if not path.is_dir():
continue
for _name, path in tablet_documents(directory):
copy_name = path.name if path.name.startswith("Copie") else path.parent.name
default_id = (
path.name.removeprefix("Copie") if path.name.startswith("Copie") else None
copy_name.removeprefix("Copie") if copy_name.startswith("Copie") else None
)
try:
metadata = read_json(path / "bnote.json")
@@ -590,12 +590,12 @@ def run(
if annotation_dir == "Bnot":
for student_id in refaire_by_student:
directory = workspace.root / annotation_dir / f"Copie{student_id}"
if directory.is_dir():
for document in copy_documents(directory) if directory.is_dir() else []:
_merge_scan_result(
actions_by_student,
notes_by_student,
_scan_annotation_directory(
directory, default_student_id=student_id
document, default_student_id=student_id
),
)
+156
View File
@@ -0,0 +1,156 @@
"""Large checked copies survive the tablet round trip as separate documents."""
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from PIL import Image, ImageDraw
from copienator import CliError, EvaluationWorkspace, ExitCode, read_json
from copienator.annotation_data import AnnotationLoadResult
from copienator.annotation_documents import MAX_HEIGHT_PX, copy_documents
from copienator.commands import annotating_with_checks as checks
from copienator.commands import export, import_annotations, reading_annotations as reader
from copienator.commands import reading_grouped_annotations as grouped_reader
class CheckedDocumentTests(unittest.TestCase):
def setUp(self):
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
self.root = Path(temporary.name)
self.workspace = EvaluationWorkspace(self.root / 'Exam')
for name in ('Copies', 'Par label'):
(self.workspace.root / name).mkdir(parents=True)
(self.workspace.root / 'labels').write_text('Ex 1\nEx 2\nEx 3\n')
(self.workspace.root / 'correction.json').write_text('{}')
self.heights = {'Ex 1': 60, 'Ex 2': 40, 'Ex 3': 60}
self.labels = {}
for label in self.heights:
pdf = self.workspace.root / 'Copies' / f'{label}.pdf'
pdf.touch()
self.labels[label] = {'pdf_path': pdf, 'result': {'score': 2, 'feedback': []},
'coordinates': (0, 0)}
self.output = self.workspace.root / 'Bnot' / 'Copie01'
def compose(self, base, label, result, coordinates, **kwargs):
image = Image.new('RGB', (100, self.heights[label]), 'white')
callback = kwargs.get('draw_callback')
if callback:
callback.__self__.checkboxes.append({
'type': 'score', 'label': label, 'value': 3,
'final_box': [10, 5, 30, 25],
})
ImageDraw.Draw(image).rectangle((10, 5, 30, 25), outline='black', width=2)
return image, 0
def generate(self, overwrite=False):
with (
patch.object(checks, 'MAX_HEIGHT_PX', 100),
patch.object(checks.annotating, 'make_base_image', return_value=(None, 0, 0)),
patch.object(checks.annotating, 'compose_label_image', side_effect=self.compose),
):
return checks._render_student(self.workspace, '01', self.labels,
overwrite=overwrite, output_mode='checks')
def test_splits_on_labels_and_resets_geometry_at_each_part(self):
self.assertEqual(MAX_HEIGHT_PX, 32000)
self.assertEqual(self.generate(), 'success')
parts = copy_documents(self.output)
self.assertEqual([part.name for part in parts], ['Part1', 'Part2'])
self.assertEqual([read_json(p / 'bnote.json')['height'] for p in parts], [100, 60])
self.assertEqual([item['hmin'] for item in read_json(self.output / 'bnote.json')['images']],
[0, 60, 0])
self.assertEqual(read_json(parts[1] / 'checkboxes.json')[0]['global_box'], [10, 5, 30, 25])
self.assertTrue(checks._output_complete(self.output))
self.assertEqual(self.generate(), 'skipped')
(parts[1] / 'Reference.jpg').unlink()
self.assertFalse(checks._output_complete(self.output))
def test_overwrite_removes_stale_parts_and_keeps_small_copy_layout(self):
self.generate()
self.labels = {'Ex 1': self.labels['Ex 1']}
self.generate(overwrite=True)
self.assertEqual(copy_documents(self.output), [self.output])
self.assertTrue((self.output / 'Concat.pdf').is_file())
self.assertFalse((self.output / 'Part1').exists())
def test_single_oversized_answer_and_save_failure_preserve_previous_copy(self):
self.generate()
sentinel = self.output / 'sentinel'
sentinel.write_text('previous')
self.heights['Ex 3'] = 101
with self.assertRaisesRegex(CliError, 'Copie01.*Ex 3'):
self.generate(overwrite=True)
self.assertEqual(sentinel.read_text(), 'previous')
self.heights['Ex 3'] = 60
original = checks._save_document
def fail_second(directory, *args):
if directory.name == 'Part2':
raise OSError('disk full')
original(directory, *args)
with patch.object(checks, '_save_document', side_effect=fail_second):
with self.assertRaisesRegex(OSError, 'disk full'):
self.generate(overwrite=True)
self.assertEqual(sentinel.read_text(), 'previous')
def returned_parts(self):
self.generate()
export_root = self.root / 'Export'
with patch.object(export, 'EXPORT_DIR', export_root):
self.assertEqual(export.run(self.workspace, annotation_dir='Bnot'), ExitCode.SUCCESS)
self.assertEqual(sorted(p.name for p in (export_root / 'Exam').iterdir()),
['Copie01 Part1.pdf', 'Copie01 Part2.pdf'])
import_root = self.root / 'Import'
import_root.mkdir()
for part in copy_documents(self.output):
with Image.open(part / 'Reference.jpg') as reference:
image = reference.convert('RGB')
draw = ImageDraw.Draw(image)
# Change Ex 3's score and add handwriting in its body.
if part.name == 'Part2':
draw.rectangle((14, 9, 26, 21), fill='black')
draw.rectangle((70, 40, 85, 50), fill='black')
image.save(import_root / f'Copie01 {part.name}.pdf', 'PDF', resolution=72)
self.assertEqual(import_annotations.sync_annotated(
self.workspace, annotation_dir_name='Bnot', import_dir=import_root), ExitCode.SUCCESS)
def test_export_import_and_reader_merge_parts_without_moving_notes(self):
self.returned_parts()
with (
patch.object(reader, 'load_annotation_data', return_value=AnnotationLoadResult({'01': self.labels}, [])),
patch.object(reader.annotating, 'make_base_image', return_value=(None, 0, 0)),
patch.object(reader.annotating, 'compose_label_image', side_effect=self.compose),
):
self.assertEqual(reader.run(self.workspace), ExitCode.SUCCESS)
self.assertEqual(read_json(self.output / 'score.json'), {'Ex 1': '2', 'Ex 2': '2', 'Ex 3': '3'})
with Image.open(self.output / 'Ex 3.jpg') as image:
self.assertLess(max(image.getpixel((77, 45))), 80)
with Image.open(self.output / 'Ex 1.jpg') as image:
self.assertGreater(min(image.getpixel((77, 45))), 230)
def test_missing_return_part_preserves_previous_regenerated_files(self):
self.returned_parts()
(self.output / 'Part2' / 'Concat_annotated.pdf').unlink()
(self.output / 'score.json').write_text('{"previous": true}')
with patch.object(reader, 'load_annotation_data', return_value=AnnotationLoadResult({'01': self.labels}, [])):
self.assertEqual(reader.run(self.workspace), ExitCode.PARTIAL)
self.assertEqual(read_json(self.output / 'score.json'), {'previous': True})
def test_redo_reader_accepts_split_copy_documents(self):
self.returned_parts()
# Redo uses the same document layout under BRnot.
self.output.parent.rename(self.workspace.root / 'BRnot')
actions, notes, incomplete = grouped_reader._scan_redo_annotations(
self.workspace.root / 'BRnot', {'01': set(self.labels)})
self.assertFalse(incomplete)
self.assertEqual(set(notes['01']), set(self.labels))
self.assertEqual(actions['01'][0]['label'], 'Ex 3')
(self.workspace.root / 'BRnot/Copie01/Part2/Concat_annotated.pdf').unlink()
_, _, incomplete = grouped_reader._scan_redo_annotations(
self.workspace.root / 'BRnot', {'01': set(self.labels)})
self.assertEqual(incomplete, {'01'})
if __name__ == '__main__':
unittest.main()
+22 -6
View File
@@ -2338,12 +2338,28 @@ class StandardCliTests(unittest.TestCase):
("01", "Ex 2", image, 0, []),
("02", "Ex 1", image, 0, []),
]
with patch.object(module, "MAX_HEIGHT_PX", 100), self.assertRaises(
CliError
) as raised:
module.split_batches(rendered)
self.assertIn("Copie01", str(raised.exception))
self.assertIn("cannot be split", str(raised.exception))
with patch.object(module, "MAX_HEIGHT_PX", 100):
batches = module.split_batches(rendered)
self.assertEqual(batches, [rendered[:2], rendered[2:]])
def test_grouped_oversized_answer_is_saved_as_its_own_group(self) -> None:
module = self.modules["annotating_by_label"]
rendered = [
("01", "Ex 1", Image.new("RGB", (10, 40)), 0, []),
("02", "Ex 1", Image.new("RGB", (10, 120)), 0, []),
("03", "Ex 1", Image.new("RGB", (10, 50)), 0, []),
]
with tempfile.TemporaryDirectory() as directory, patch.object(module, "MAX_HEIGHT_PX", 100):
batches = module.split_batches(rendered)
self.assertEqual(batches, [[rendered[0], rendered[2]], [rendered[1]]])
for index, batch in enumerate(batches, start=1):
module.save_batch(batch, "Ex 1", index, Path(directory))
metadata = read_json(Path(directory) / "Ex 1 G2/bnote.json")
self.assertEqual(metadata["height"], 120)
self.assertEqual([item["id"] for item in metadata["images"]], ["02"])
self.assertTrue((Path(directory) / "Ex 1 G2/Concat.pdf").is_file())
with self.assertRaises(CliError):
module.save_batch(rendered[:2], "Ex 1", 3, Path(directory))
def test_grouped_batching_reuses_space_in_earlier_groups(self) -> None:
module = self.modules["annotating_by_label"]