diff --git a/copienator/commands/crop_exercise_bottoms.py b/copienator/commands/crop_exercise_bottoms.py new file mode 100644 index 0000000..b12ab18 --- /dev/null +++ b/copienator/commands/crop_exercise_bottoms.py @@ -0,0 +1,197 @@ +"""Optionally replace split-answer PDFs with large bottom crops.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import multiprocessing +import shutil +import signal +import tempfile +from collections.abc import Sequence +from pathlib import Path + +import cv2 + +from copienator.cli import CliError, ExitCode, execute, target_parser, workspace_from_target +from copienator.crop_exercise_bottoms import _full_page_height, process_exercise_pdf +from copienator.workspace import EvaluationWorkspace + + +def selected_files(workspace: EvaluationWorkspace, target: Path) -> list[Path]: + workspace.require_directories("Copies") + resolved = target.resolve() + if resolved not in {workspace.root.resolve(), workspace.copies_dir.resolve()}: + raise CliError( + "La cible doit être l’évaluation ou son dossier Copies.", + ExitCode.INVALID_ARGUMENTS, + ) + files = sorted( + workspace.copies_dir.glob("Copie*/*.pdf"), + key=lambda path: (path.parent.name.casefold(), path.name.casefold()), + ) + for source in files: + if source.is_symlink(): + raise CliError(f"Lien symbolique non pris en charge : {source}") + return files + + +def _initialize_worker() -> None: + cv2.setNumThreads(1) + signal.signal(signal.SIGINT, signal.SIG_IGN) + + +def _process_file(job: tuple[Path, Path, float]) -> list[dict]: + source, destination, full_height = job + records = process_exercise_pdf( + source, destination, None, full_height, dpi=200, padding_mm=6 + ) + changed = sum(record["status"] == "cropped" for record in records) + if changed: + print( + f"{source.parent.name}/{source.name} : {changed}/{len(records)} page(s) rognée(s)", + flush=True, + ) + return records + + +def process_files( + workspace: EvaluationWorkspace, + files: list[Path], + staging: Path, + workers: int, +) -> list[dict]: + heights: dict[str, float] = {} + for copy_name in sorted({source.parent.name for source in files}): + copy_pdf = workspace.copies_dir / f"{copy_name}.pdf" + if not copy_pdf.is_file(): + raise CliError(f"PDF source introuvable : {copy_pdf}") + heights[copy_name] = _full_page_height(copy_pdf) + jobs = [ + ( + source, + staging / source.relative_to(workspace.copies_dir), + heights[source.parent.name], + ) + for source in files + ] + count = min(workers, len(jobs)) + print( + f"Analyse de {len(files)} PDF de réponses avec {count} traitement(s) en parallèle.", + flush=True, + ) + if count == 1: + previous_threads = cv2.getNumThreads() + cv2.setNumThreads(1) + try: + batches = [_process_file(job) for job in jobs] + finally: + cv2.setNumThreads(previous_threads) + else: + with multiprocessing.get_context("spawn").Pool( + count, _initialize_worker + ) as pool: + batches = list(pool.imap_unordered(_process_file, jobs)) + order = {source.as_posix(): index for index, source in enumerate(files)} + return sorted( + (record for batch in batches for record in batch), + key=lambda record: (order[record["file"]], record["page"]), + ) + + +def _publish( + workspace: EvaluationWorkspace, + changed_files: list[Path], + staging: Path, + records: list[dict], +) -> Path: + workspace.runs_dir.mkdir(parents=True, exist_ok=True) + run_dir = Path( + tempfile.mkdtemp(prefix="crop-exercise-bottoms-", dir=workspace.runs_dir) + ) + backup_root = run_dir / "Copies" + replaced: list[Path] = [] + try: + for source in changed_files: + backup = backup_root / source.relative_to(workspace.copies_dir) + backup.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source, backup) + (run_dir / "report.json").write_text( + json.dumps(records, ensure_ascii=False, indent=2) + "\n", encoding="utf-8" + ) + for source in changed_files: + prepared = staging / source.relative_to(workspace.copies_dir) + prepared.replace(source) + replaced.append(source) + except BaseException: + for source in replaced: + backup = backup_root / source.relative_to(workspace.copies_dir) + if backup.is_file(): + shutil.copy2(backup, source) + shutil.rmtree(run_dir, ignore_errors=True) + raise + return backup_root + + +def run(workspace: EvaluationWorkspace, target: Path, *, workers: int = 5) -> ExitCode: + if workers < 1: + raise CliError( + "Le nombre de traitements parallèles doit être positif.", + ExitCode.INVALID_ARGUMENTS, + ) + files = selected_files(workspace, target) + if not files: + raise CliError( + "Aucun PDF de réponse trouvé dans Copies/CopieXX/.", + ExitCode.INVALID_ARGUMENTS, + ) + with tempfile.TemporaryDirectory( + prefix=".crop-exercise-bottoms-", dir=workspace.root + ) as directory: + staging = Path(directory) + records = process_files(workspace, files, staging, workers) + changed_names = { + record["file"] for record in records if record["status"] == "cropped" + } + changed_files = [source for source in files if source.as_posix() in changed_names] + digests = {record["file"]: record["source_sha256"] for record in records} + for source in files: + if hashlib.sha256(source.read_bytes()).hexdigest() != digests[source.as_posix()]: + raise CliError( + f"{source} a changé pendant l’analyse. Aucun PDF remplacé." + ) + if not changed_files: + print("Terminé : aucun PDF ne remplit les critères de rognage.", flush=True) + return ExitCode.SUCCESS + backup = _publish(workspace, changed_files, staging, records) + cropped_pages = sum(record["status"] == "cropped" for record in records) + print(f"Sauvegarde des PDF non rognés : {backup}", flush=True) + print( + f"Terminé : {cropped_pages} page(s) rognée(s) dans " + f"{len(changed_files)} PDF remplacé(s).", + flush=True, + ) + return ExitCode.SUCCESS + + +def main(argv: Sequence[str] | None = None) -> int: + parser = target_parser( + "Rogner les grands espaces vides au bas des réponses déjà découpées" + ) + parser.add_argument( + "--workers", + type=int, + default=5, + help="Nombre de PDF traités en parallèle (défaut : 5)", + ) + + def handle(arguments: argparse.Namespace) -> ExitCode: + workspace, target = workspace_from_target(arguments) + return run(workspace, target, workers=arguments.workers) + + return execute(parser, argv, handle) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/copienator/crop_exercise_bottoms.py b/copienator/crop_exercise_bottoms.py new file mode 100644 index 0000000..0d005eb --- /dev/null +++ b/copienator/crop_exercise_bottoms.py @@ -0,0 +1,298 @@ +"""Propose large bottom-only crops for PDFs already split into exercises.""" + +from __future__ import annotations + +import argparse +import csv +import hashlib +import html +import json +import multiprocessing +import signal +from pathlib import Path +from urllib.parse import quote + +import cv2 +import numpy as np +import pymupdf +from PIL import Image, ImageDraw + +from copienator.crop_blank_margins import apply_bounds +from copienator.filesystem import staged_directory +from copienator.ink_detection import detect_bounds + + +LINES_PER_PAGE = 36 +MINIMUM_HEIGHT_LINES = 10 +IGNORED_BOTTOM_LINES = 0.75 +MINIMUM_CROP_LINES = 4 + + +def _displayed_media_height(page: pymupdf.Page) -> float: + """Return the uncropped sheet height in the page's displayed orientation.""" + return page.mediabox.width if page.rotation % 180 else page.mediabox.height + + +def _full_page_height(copy_pdf: Path) -> float: + with pymupdf.open(copy_pdf) as document: + if not len(document): + raise ValueError(f"PDF sans page : {copy_pdf}") + return max(_displayed_media_height(page) for page in document) + + +def _save_review(rgb: np.ndarray, bottom: int, destination: Path) -> None: + preview = Image.fromarray(rgb) + preview.thumbnail((500, 700)) + overlay = Image.new("RGBA", preview.size) + draw = ImageDraw.Draw(overlay) + y = bottom / rgb.shape[0] * preview.height + draw.rectangle((0, y, preview.width, preview.height), fill=(255, 40, 40, 85)) + draw.line((0, y, preview.width, y), fill=(230, 0, 0, 255), width=2) + destination.parent.mkdir(parents=True, exist_ok=True) + Image.alpha_composite(preview.convert("RGBA"), overlay).convert("RGB").save( + destination, quality=88 + ) + + +def process_exercise_pdf( + source: Path, + destination: Path, + review_dir: Path | None, + full_page_height: float, + *, + dpi: int = 200, + padding_mm: float = 6, +) -> list[dict]: + """Crop qualifying pages and save the PDF only when at least one changes.""" + digest = hashlib.sha256(source.read_bytes()).hexdigest() + line_points = full_page_height / LINES_PER_PAGE + records: list[dict] = [] + changed = False + with pymupdf.open(source) as document: + page_count = len(document) + for index, page in enumerate(document): + visible_height = page.rect.height + record = { + "file": source.as_posix(), + "page": index + 1, + "page_count": page_count, + "source_sha256": digest, + "height_lines": round(visible_height / line_points, 2), + "bottom_removed_lines": 0.0, + "bottom_removed_mm": 0.0, + "status": "skipped-short", + } + if visible_height + 1e-6 < MINIMUM_HEIGHT_LINES * line_points: + records.append(record) + continue + + pixmap = page.get_pixmap( + dpi=dpi, colorspace=pymupdf.csRGB, alpha=False + ) + rgb = np.frombuffer(pixmap.samples, np.uint8).reshape( + pixmap.height, pixmap.width, 3 + ) + pixels_per_point = pixmap.height / visible_height + ignored_pixels = min( + pixmap.height - 1, + round(IGNORED_BOTTOM_LINES * line_points * pixels_per_point), + ) + analysis_bottom = pixmap.height - ignored_pixels + detection = detect_bounds( + rgb[:analysis_bottom], dpi=dpi, padding_mm=padding_mm, min_crop_mm=0 + ) + proposed_bottom = detection["bottom_px"] + removed_points = (pixmap.height - proposed_bottom) / pixels_per_point + removed_lines = removed_points / line_points + record["detector_status"] = detection["status"] + record["proposed_bottom_removed_lines"] = round(removed_lines, 2) + if removed_lines + 1e-6 < MINIMUM_CROP_LINES: + record["status"] = "unchanged-small-crop" + records.append(record) + continue + + apply_bounds(page, 0, proposed_bottom / pixmap.height) + record.update( + bottom_removed_lines=round(removed_lines, 2), + bottom_removed_mm=round(removed_points * 25.4 / 72, 2), + status="cropped", + output_height_points=round(page.rect.height, 3), + ) + if review_dir is not None: + record["output"] = destination.relative_to(review_dir.parent).as_posix() + review_path = review_dir / source.parent.name / ( + f"{source.stem}-p{index + 1:02}.jpg" + ) + _save_review(rgb, proposed_bottom, review_path) + record["review"] = review_path.relative_to(review_dir.parent).as_posix() + changed = True + records.append(record) + + if changed: + destination.parent.mkdir(parents=True, exist_ok=True) + document.save(destination, garbage=3, deflate=True) + + if changed: + with pymupdf.open(destination) as check: + if len(check) != page_count: + raise RuntimeError(f"Nombre de pages modifié : {source}") + for page in check: + if page.rect.is_empty: + raise RuntimeError(f"Page vide produite : {destination}") + page.get_pixmap(matrix=pymupdf.Matrix(0.25, 0.25)) + if hashlib.sha256(source.read_bytes()).hexdigest() != digest: + raise RuntimeError(f"PDF source modifié pendant le rognage : {source}") + return records + + +def _initialize_worker() -> None: + cv2.setNumThreads(1) + signal.signal(signal.SIGINT, signal.SIG_IGN) + + +def _process_job(job: tuple[Path, Path, Path, float, int, float]) -> list[dict]: + source, destination, review_dir, full_height, dpi, padding = job + records = process_exercise_pdf( + source, destination, review_dir, full_height, dpi=dpi, padding_mm=padding + ) + changed = sum(record["status"] == "cropped" for record in records) + print(f"{source.parent.name}/{source.name} : {changed}/{len(records)} page(s) rognée(s)", + flush=True) + return records + + +def _write_index(output: Path, records: list[dict]) -> None: + changed = [record for record in records if record["status"] == "cropped"] + changed_files: dict[str, list[int]] = {} + for record in changed: + changed_files.setdefault(record["file"], []).append(record["page"]) + (output / "cropped-files.txt").write_text( + "".join( + f"{path} - page(s) {', '.join(map(str, pages))}\n" + for path, pages in changed_files.items() + ), + encoding="utf-8", + ) + cards = [] + for record in changed: + relative = Path(record["file"]) + output_pdf = quote(record["output"]) + preview = quote(record["review"]) + name = html.escape(f"{relative.parent.name}/{relative.name} - page {record['page']}") + cards.append( + f'
{name}' + f'

{record["bottom_removed_lines"]:.2f} lignes ' + f'({record["bottom_removed_mm"]:.1f} mm) retirées

' + f'
' + ) + (output / "index.html").write_text( + 'Rognage bas des exercices' + '' + f'

{len(changed)} pages rognées

' + '

Le rouge montre la zone retirée. Seuls les PDF modifiés sont présents dans cropped/.

' + '
' + ''.join(cards) + '
', + encoding="utf-8", + ) + + +def run( + input_path: Path, + output: Path, + *, + dpi: int = 200, + padding_mm: float = 6, + workers: int = 5, +) -> list[dict]: + copies = input_path / "Copies" if (input_path / "Copies").is_dir() else input_path + if not copies.is_dir(): + raise ValueError(f"Dossier Copies introuvable : {input_path}") + sources = sorted( + copies.glob("Copie*/*.pdf"), + key=lambda path: (path.parent.name.casefold(), path.name.casefold()), + ) + if not sources: + raise ValueError(f"Aucun PDF d'exercice trouvé dans {copies}") + if workers < 1: + raise ValueError("Le nombre de traitements parallèles doit être positif") + + heights: dict[str, float] = {} + for copy_name in sorted({source.parent.name for source in sources}): + copy_pdf = copies / f"{copy_name}.pdf" + if not copy_pdf.is_file(): + raise ValueError(f"PDF source introuvable : {copy_pdf}") + heights[copy_name] = _full_page_height(copy_pdf) + + with staged_directory(output) as staging: + cropped_dir = staging / "cropped" + review_dir = staging / "review" + jobs = [ + ( + source, + cropped_dir / source.relative_to(copies), + review_dir, + heights[source.parent.name], + dpi, + padding_mm, + ) + for source in sources + ] + count = min(workers, len(jobs)) + if count == 1: + previous_threads = cv2.getNumThreads() + cv2.setNumThreads(1) + try: + batches = [_process_job(job) for job in jobs] + finally: + cv2.setNumThreads(previous_threads) + else: + with multiprocessing.get_context("spawn").Pool( + count, _initialize_worker + ) as pool: + batches = list(pool.imap_unordered(_process_job, jobs)) + order = {source.as_posix(): i for i, source in enumerate(sources)} + records = sorted( + (record for batch in batches for record in batch), + key=lambda record: (order[record["file"]], record["page"]), + ) + (staging / "report.json").write_text( + json.dumps(records, ensure_ascii=False, indent=2) + "\n", encoding="utf-8" + ) + with (staging / "report.csv").open("w", encoding="utf-8", newline="") as stream: + fields = sorted({key for record in records for key in record}) + writer = csv.DictWriter(stream, fieldnames=fields) + writer.writeheader() + writer.writerows(records) + _write_index(staging, records) + return records + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("input", type=Path, help="Évaluation ou dossier Copies") + parser.add_argument("output", type=Path) + parser.add_argument("--dpi", type=int, default=200) + parser.add_argument("--padding-mm", type=float, default=6) + parser.add_argument("--workers", type=int, default=5) + arguments = parser.parse_args() + if arguments.dpi < 100 or arguments.padding_mm < 0: + parser.error("Utilisez dpi >= 100 et une marge positive ou nulle") + try: + records = run( + arguments.input, + arguments.output, + dpi=arguments.dpi, + padding_mm=arguments.padding_mm, + workers=arguments.workers, + ) + except ValueError as error: + parser.error(str(error)) + cropped = sum(record["status"] == "cropped" for record in records) + files = len({record["file"] for record in records if record["status"] == "cropped"}) + print(f"Terminé : {cropped} page(s) dans {files} PDF rognée(s). Revue : " + f"{arguments.output / 'index.html'}") + + +if __name__ == "__main__": + main() diff --git a/copienator/dispatcher.py b/copienator/dispatcher.py index a7decc5..1328838 100644 --- a/copienator/dispatcher.py +++ b/copienator/dispatcher.py @@ -18,6 +18,9 @@ COMMANDS: dict[str, Command] = { "copies": Command("copies_tools", "Rotate or rename scanned copies"), "page-split": Command("page_splitter", "Split and reorder scanned PDF pages"), "crop-margins": Command("crop_margins", "Trim blank top and bottom margins in Copies"), + "crop-answer-bottoms": Command( + "crop_exercise_bottoms", "Trim large blank bottoms from split answers" + ), "crop-labels": Command("cutleft", "Crop the label margin from copies"), "labels": Command("gemini_for_labels", "Detect question labels with Gemini"), "review-labels": Command("plotting", "Review detected labels interactively"), diff --git a/copienator_gui/workflow.py b/copienator_gui/workflow.py index 116d28c..9c9d677 100644 --- a/copienator_gui/workflow.py +++ b/copienator_gui/workflow.py @@ -237,6 +237,27 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]: artifacts=("Copies/Copie*/*",), auto_start_first_visit=True, ), + StepDefinition( + "crop_exercise_bottoms", + "Labels et regroupement", + "Rogner le bas des réponses", + "Facultatif après le découpage par labels : rogne uniquement les grands espaces " + "vides au bas des réponses. Les PDF modifiés sont remplacés, les originaux sont " + "sauvegardés et plusieurs fichiers sont analysés en parallèle.", + (python("default", "Rognage du bas", "crop-answer-bottoms"),), + arguments=( + arg_target("Dossier de l’évaluation"), + ArgumentSpec( + "workers", + "PDF traités en parallèle", + "int", + "--workers", + default=5, + ), + ), + optional=True, + requires=("Copies/Copie*/*.pdf",), + ), StepDefinition( "grouping", "Labels et regroupement", diff --git a/tests/test_crop_exercise_bottoms.py b/tests/test_crop_exercise_bottoms.py new file mode 100644 index 0000000..2088fa7 --- /dev/null +++ b/tests/test_crop_exercise_bottoms.py @@ -0,0 +1,59 @@ +import tempfile +import unittest +from pathlib import Path + +import pymupdf + +from copienator.crop_exercise_bottoms import process_exercise_pdf + + +class CropExerciseBottomsTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.root = Path(self.temp.name) + self.review = self.root / "review" + + def make_pdf(self, name: str, height: float, text_y: float, footer=True) -> Path: + path = self.root / name + with pymupdf.open() as document: + page = document.new_page(width=600, height=height) + page.insert_text((80, text_y), "student answer", fontsize=16) + if footer: + page.insert_text((20, height - 3), "Ex 2", fontsize=10) + document.save(path) + return path + + def test_short_page_is_skipped(self): + source = self.make_pdf("short.pdf", 200, 100) + destination = self.root / "out" / source.name + records = process_exercise_pdf( + source, destination, self.review, 800, dpi=150 + ) + self.assertEqual(records[0]["status"], "skipped-short") + self.assertFalse(destination.exists()) + + def test_footer_fragment_is_ignored_for_a_large_bottom_crop(self): + source = self.make_pdf("large.pdf", 400, 100) + destination = self.root / "out" / source.name + records = process_exercise_pdf( + source, destination, self.review, 800, dpi=150 + ) + self.assertEqual(records[0]["status"], "cropped") + self.assertGreaterEqual(records[0]["bottom_removed_lines"], 4) + with pymupdf.open(destination) as result: + self.assertLess(result[0].rect.height, 250) + self.assertAlmostEqual(result[0].rect.width, 600) + + def test_crop_smaller_than_four_lines_is_not_written(self): + source = self.make_pdf("small.pdf", 400, 330, footer=False) + destination = self.root / "out" / source.name + records = process_exercise_pdf( + source, destination, self.review, 800, dpi=150 + ) + self.assertEqual(records[0]["status"], "unchanged-small-crop") + self.assertFalse(destination.exists()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_crop_exercise_bottoms_command.py b/tests/test_crop_exercise_bottoms_command.py new file mode 100644 index 0000000..de33a7c --- /dev/null +++ b/tests/test_crop_exercise_bottoms_command.py @@ -0,0 +1,123 @@ +import contextlib +import io +import tempfile +import unittest +from pathlib import Path + +import pymupdf + +from copienator.cli import CliError +from copienator.commands import crop_exercise_bottoms +from copienator.commands.clean import apply_cleanup, build_cleanup_plan +from copienator.dispatcher import main +from copienator.workspace import EvaluationWorkspace +from copienator_gui.workflow import build_workflow + + +class CropExerciseBottomsCommandTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.workspace = EvaluationWorkspace(Path(self.temp.name) / "Évaluation") + self.workspace.copies_dir.mkdir(parents=True) + self.copy_pdf = self.workspace.copies_dir / "Copie01.pdf" + with pymupdf.open() as document: + document.new_page(width=600, height=800) + document.save(self.copy_pdf) + self.answers = self.workspace.copies_dir / "Copie01" + self.answers.mkdir() + self.large = self.answers / "Ex 1.pdf" + self.short = self.answers / "Ex 2.pdf" + self._make_answer(self.large, 400, 100, footer=True) + self._make_answer(self.short, 200, 100, footer=False) + self.large_original = self.large.read_bytes() + self.short_original = self.short.read_bytes() + + @staticmethod + def _make_answer(path: Path, height: float, text_y: float, *, footer: bool) -> None: + with pymupdf.open() as document: + page = document.new_page(width=600, height=height) + page.insert_text((80, text_y), "student answer", fontsize=16) + if footer: + page.insert_text((20, height - 3), "Ex 2", fontsize=10) + document.save(path) + + def test_dispatcher_replaces_only_changed_answers_and_backs_them_up(self): + with contextlib.redirect_stdout(io.StringIO()) as log: + self.assertEqual( + main(["crop-answer-bottoms", str(self.workspace.root), "--workers", "2"]), + 0, + ) + self.assertIn("1 PDF remplacé", log.getvalue()) + with pymupdf.open(self.large) as cropped: + self.assertLess(cropped[0].rect.height, 250) + self.assertEqual(self.short.read_bytes(), self.short_original) + backups = list( + self.workspace.runs_dir.glob( + "crop-exercise-bottoms-*/Copies/Copie01/Ex 1.pdf" + ) + ) + self.assertEqual(len(backups), 1) + self.assertEqual(backups[0].read_bytes(), self.large_original) + self.assertFalse( + list( + self.workspace.runs_dir.glob( + "crop-exercise-bottoms-*/Copies/Copie01/Ex 2.pdf" + ) + ) + ) + self.assertTrue((backups[0].parents[2] / "report.json").is_file()) + + def test_detection_failure_does_not_publish_an_earlier_result(self): + broken = self.answers / "Ex 3.pdf" + broken.write_bytes(b"not a PDF") + with contextlib.redirect_stdout(io.StringIO()), self.assertRaises(Exception): + crop_exercise_bottoms.run( + self.workspace, self.workspace.root, workers=1 + ) + self.assertEqual(self.large.read_bytes(), self.large_original) + self.assertEqual(self.short.read_bytes(), self.short_original) + self.assertEqual(broken.read_bytes(), b"not a PDF") + self.assertFalse(self.workspace.runs_dir.exists()) + + def test_invalid_target_and_worker_count_are_rejected(self): + with self.assertRaises(CliError): + crop_exercise_bottoms.run(self.workspace, self.large, workers=1) + with self.assertRaises(CliError): + crop_exercise_bottoms.run(self.workspace, self.workspace.root, workers=0) + + def test_archiving_removes_the_retained_originals_and_report(self): + with contextlib.redirect_stdout(io.StringIO()): + crop_exercise_bottoms.run( + self.workspace, self.workspace.root, workers=1 + ) + run_dirs = list( + self.workspace.runs_dir.glob("crop-exercise-bottoms-*") + ) + self.assertEqual(len(run_dirs), 1) + self.workspace.correction_file.write_text("{}") + student = self.workspace.return_dir / "Student" + student.mkdir(parents=True) + (student / "answer.jpg").write_bytes(b"return image") + (student / "score.json").write_text("{}") + plan = build_cleanup_plan(self.workspace) + self.assertTrue( + any(path.name == "report.json" for path in plan.deleted_files) + ) + with contextlib.redirect_stdout(io.StringIO()): + apply_cleanup(self.workspace, plan) + self.assertFalse(self.workspace.runs_dir.exists()) + + def test_optional_step_follows_splitting_and_precedes_grouping(self): + steps = build_workflow(False) + index = next( + i for i, step in enumerate(steps) if step.id == "crop_exercise_bottoms" + ) + self.assertEqual(steps[index - 1].id, "splitting") + self.assertEqual(steps[index + 1].id, "grouping") + self.assertTrue(steps[index].optional) + self.assertFalse(steps[index].auto_start_first_visit) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_gui_convenience.py b/tests/test_gui_convenience.py index 32e6fb0..e69fc91 100644 --- a/tests/test_gui_convenience.py +++ b/tests/test_gui_convenience.py @@ -68,6 +68,24 @@ class GuiConvenienceTests(unittest.TestCase): self.app._skip_step() self.assertEqual(self.app.state_store.step("crop_blank_margins")["status"], "skipped") + def test_optional_answer_bottom_crop_uses_parallel_workers_and_can_be_skipped(self): + answers = self.evaluation / "Copies" / "Copie01" + answers.mkdir(parents=True) + (answers / "Ex 1.pdf").touch() + self.app.tree.selection_set("crop_exercise_bottoms") + self.app.update() + self.assertEqual(self.app.current_step.id, "crop_exercise_bottoms") + self.assertEqual(str(self.app.skip_button.cget("state")), "normal") + command = self.app._make_command() + self.assertEqual( + command[-4:], + ["crop-answer-bottoms", self.app._evaluation_arg(), "--workers", "5"], + ) + self.app._skip_step() + self.assertEqual( + self.app.state_store.step("crop_exercise_bottoms")["status"], "skipped" + ) + def test_redo_targets_selected_copy_and_copies_runnable_command(self): for folder in ("Copies", "Copies Originales"): (self.evaluation / folder).mkdir()