Initial cropping support
This commit is contained in:
@@ -0,0 +1,30 @@
|
||||
import unittest
|
||||
|
||||
import numpy as np
|
||||
import pymupdf
|
||||
|
||||
from copienator.crop_blank_margins import apply_bounds
|
||||
|
||||
|
||||
class CropBlankMarginsTests(unittest.TestCase):
|
||||
def test_cropbox_coordinates_with_rotation_and_existing_crop(self):
|
||||
for rotation in (0, 90, 180, 270):
|
||||
with self.subTest(rotation=rotation), pymupdf.open() as doc:
|
||||
page = doc.new_page(width=600, height=800)
|
||||
page.set_cropbox(pymupdf.Rect(30, 40, 570, 760))
|
||||
page.set_rotation(rotation)
|
||||
before = page.rect
|
||||
page.insert_text((80, 220), 'Visible content', fontsize=20)
|
||||
pix = page.get_pixmap()
|
||||
original = np.frombuffer(pix.samples, np.uint8).reshape(pix.height, pix.width, 3)
|
||||
apply_bounds(page, .2, .8)
|
||||
self.assertAlmostEqual(page.rect.width, before.width)
|
||||
self.assertAlmostEqual(page.rect.height, before.height*.6, places=3)
|
||||
self.assertEqual(page.rotation, rotation)
|
||||
pix = page.get_pixmap()
|
||||
cropped = np.frombuffer(pix.samples, np.uint8).reshape(pix.height, pix.width, 3)
|
||||
np.testing.assert_array_equal(cropped, original[int(before.height*.2):int(before.height*.8)])
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -0,0 +1,189 @@
|
||||
import contextlib
|
||||
import io
|
||||
import os
|
||||
import signal
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import pymupdf
|
||||
|
||||
from copienator.cli import CliError
|
||||
from copienator.commands import crop_margins
|
||||
from copienator.commands.clean import build_cleanup_plan, apply_cleanup
|
||||
from copienator.dispatcher import main
|
||||
from copienator.workspace import EvaluationWorkspace
|
||||
from copienator_gui.workflow import build_workflow
|
||||
|
||||
|
||||
class CropMarginsCommandTests(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.source = self.workspace.copies_dir/"Copie01.pdf"
|
||||
with pymupdf.open() as doc:
|
||||
page = doc.new_page(width=300, height=420)
|
||||
page.insert_text((50, 180), "answer = 42", fontsize=15)
|
||||
doc.new_page(width=300, height=420)
|
||||
doc.save(self.source)
|
||||
self.original = self.source.read_bytes()
|
||||
|
||||
def test_dispatcher_replaces_copies_and_keeps_recoverable_original(self):
|
||||
with contextlib.redirect_stdout(io.StringIO()) as log:
|
||||
self.assertEqual(main(["crop-margins", str(self.workspace.root)]), 0)
|
||||
self.assertIn("Page 2/2", log.getvalue())
|
||||
with pymupdf.open(self.source) as result:
|
||||
self.assertEqual(len(result), 2)
|
||||
self.assertLess(result[0].rect.height, 150)
|
||||
self.assertIn("answer = 42", result[0].get_text())
|
||||
self.assertEqual(result[1].rect.height, 420)
|
||||
for page in result:
|
||||
self.assertEqual(page.rect.width, 300)
|
||||
page.get_pixmap()
|
||||
backups = list(self.workspace.runs_dir.glob("crop-margins-*/Copies/Copie01.pdf"))
|
||||
self.assertEqual(len(backups), 1)
|
||||
self.assertEqual(backups[0].read_bytes(), self.original)
|
||||
self.assertTrue((backups[0].parent.parent/"report.json").is_file())
|
||||
|
||||
def test_failure_or_interruption_never_publishes_partial_batch(self):
|
||||
second = self.workspace.copies_dir/"Copie02.pdf"
|
||||
second.write_bytes(self.original)
|
||||
real = crop_margins.process_pdf
|
||||
for error in (OSError("bad scan"), KeyboardInterrupt()):
|
||||
with self.subTest(error=type(error).__name__):
|
||||
def process(source, *args, **kwargs):
|
||||
if source == second:
|
||||
raise error
|
||||
return real(source, *args, **kwargs)
|
||||
with patch.object(crop_margins, "process_pdf", side_effect=process):
|
||||
with contextlib.redirect_stdout(io.StringIO()):
|
||||
with self.assertRaises(type(error)):
|
||||
crop_margins.run(self.workspace, self.workspace.root, workers=1)
|
||||
self.assertEqual(self.source.read_bytes(), self.original)
|
||||
self.assertEqual(second.read_bytes(), self.original)
|
||||
|
||||
def test_existing_coordinates_are_not_silently_invalidated(self):
|
||||
self.source.with_suffix(".json").write_text('{"list": []}')
|
||||
with self.assertRaisesRegex(CliError, "coordonnées"):
|
||||
crop_margins.run(self.workspace, self.workspace.root)
|
||||
self.assertEqual(self.source.read_bytes(), self.original)
|
||||
|
||||
def test_archiving_removes_crop_backups_but_keeps_processed_pdf_and_logs(self):
|
||||
with contextlib.redirect_stdout(io.StringIO()):
|
||||
crop_margins.run(self.workspace, self.workspace.root)
|
||||
processed = self.source.read_bytes()
|
||||
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('{}')
|
||||
self.workspace.logs_dir.mkdir(parents=True)
|
||||
log = self.workspace.logs_dir/"crop_blank_margins.log"
|
||||
log.write_text("Completed crop")
|
||||
backups = list(self.workspace.runs_dir.glob("crop-margins-*/Copies/*.pdf"))
|
||||
reports = list(self.workspace.runs_dir.glob("crop-margins-*/report.json"))
|
||||
self.assertTrue(backups)
|
||||
self.assertTrue(reports)
|
||||
plan = build_cleanup_plan(self.workspace)
|
||||
for path in backups+reports:
|
||||
self.assertIn(path, plan.deleted_files)
|
||||
self.assertIn(self.source, plan.kept_files)
|
||||
self.assertIn(log, plan.kept_files)
|
||||
with contextlib.redirect_stdout(io.StringIO()):
|
||||
apply_cleanup(self.workspace, plan)
|
||||
self.assertFalse(self.workspace.runs_dir.exists())
|
||||
self.assertEqual(self.source.read_bytes(), processed)
|
||||
self.assertEqual(log.read_text(), "Completed crop")
|
||||
self.assertTrue((student/"answer.jpg").exists())
|
||||
self.assertTrue((student/"score.json").exists())
|
||||
|
||||
def test_single_copy_selection_cannot_target_original_scans(self):
|
||||
self.assertEqual(crop_margins.selected_files(self.workspace, self.source), [self.source])
|
||||
original = self.workspace.root/"Original.pdf"
|
||||
original.write_bytes(self.original)
|
||||
with self.assertRaises(CliError):
|
||||
crop_margins.selected_files(self.workspace, original)
|
||||
|
||||
def test_parallel_workers_match_serial_results_in_copy_and_page_order(self):
|
||||
second = self.workspace.copies_dir/"Copie02.pdf"
|
||||
second.write_bytes(self.original)
|
||||
serial, parallel = self.workspace.root/"serial", self.workspace.root/"parallel"
|
||||
serial.mkdir()
|
||||
parallel.mkdir()
|
||||
files = [self.source, second]
|
||||
with contextlib.redirect_stdout(io.StringIO()):
|
||||
expected = crop_margins.process_copies(files, serial, 1)
|
||||
actual = crop_margins.process_copies(files, parallel, 2)
|
||||
self.assertEqual(actual, expected)
|
||||
self.assertEqual([(r['file'],r['page']) for r in actual],
|
||||
[(p.name,i) for p in files for i in (1,2)])
|
||||
for source in files:
|
||||
with pymupdf.open(serial/source.name) as a, pymupdf.open(parallel/source.name) as b:
|
||||
self.assertEqual([list(p.cropbox) for p in a], [list(p.cropbox) for p in b])
|
||||
|
||||
def test_worker_failure_does_not_replace_any_copy(self):
|
||||
broken = self.workspace.copies_dir/"Copie02.pdf"
|
||||
broken.write_bytes(b"not a PDF")
|
||||
with contextlib.redirect_stdout(io.StringIO()):
|
||||
with self.assertRaises(Exception):
|
||||
crop_margins.run(self.workspace, self.workspace.root, workers=2)
|
||||
self.assertEqual(self.source.read_bytes(), self.original)
|
||||
self.assertEqual(broken.read_bytes(), b"not a PDF")
|
||||
self.assertFalse(list(self.workspace.root.glob(".Copies.*.files.tmp")))
|
||||
|
||||
def test_invalid_worker_count_is_rejected_before_processing(self):
|
||||
with self.assertRaises(CliError):
|
||||
crop_margins.run(self.workspace, self.workspace.root, workers=0)
|
||||
self.assertEqual(self.source.read_bytes(), self.original)
|
||||
|
||||
@unittest.skipIf(os.name == "nt", "SIGINT subprocess check uses Unix signals")
|
||||
def test_interrupt_stops_parallel_workers_before_removing_staging(self):
|
||||
import select
|
||||
with pymupdf.open() as doc:
|
||||
for _ in range(30):
|
||||
page = doc.new_page(width=595, height=842)
|
||||
page.insert_text((100,400), "answer = 42", fontsize=20)
|
||||
data = doc.tobytes()
|
||||
self.source.write_bytes(data)
|
||||
second = self.workspace.copies_dir/"Copie02.pdf"
|
||||
second.write_bytes(data)
|
||||
proc = subprocess.Popen(
|
||||
[sys.executable,"-u","-m","copienator","crop-margins",
|
||||
str(self.workspace.root),"--workers","2"],
|
||||
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
|
||||
cwd=Path(__file__).resolve().parents[1])
|
||||
try:
|
||||
while True:
|
||||
ready, _, _ = select.select([proc.stdout], [], [], 20)
|
||||
self.assertTrue(ready, "No progress from parallel crop command")
|
||||
line = proc.stdout.readline()
|
||||
self.assertTrue(line, "Crop command exited before processing a page")
|
||||
if "Page " in line:
|
||||
break
|
||||
proc.send_signal(signal.SIGINT)
|
||||
output, _ = proc.communicate(timeout=20)
|
||||
self.assertEqual(proc.returncode, 130, output)
|
||||
finally:
|
||||
if proc.poll() is None:
|
||||
proc.kill()
|
||||
proc.communicate()
|
||||
self.assertEqual(self.source.read_bytes(), data)
|
||||
self.assertEqual(second.read_bytes(), data)
|
||||
self.assertFalse(list(self.workspace.root.glob(".Copies.*.files.tmp")))
|
||||
|
||||
def test_optional_step_sits_between_page_split_and_label_crop(self):
|
||||
steps = build_workflow(False)
|
||||
index = next(i for i, step in enumerate(steps) if step.id == "crop_blank_margins")
|
||||
self.assertEqual(steps[index-1].id, "page_splitter")
|
||||
self.assertEqual(steps[index+1].id, "cutleft")
|
||||
self.assertTrue(steps[index].optional)
|
||||
self.assertFalse(steps[index].auto_start_first_visit)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -52,6 +52,22 @@ class GuiConvenienceTests(unittest.TestCase):
|
||||
self.app.open_evaluation_button.invoke()
|
||||
opened.assert_called_once_with(self.evaluation)
|
||||
|
||||
def test_optional_blank_crop_can_target_a_copy_or_be_skipped(self):
|
||||
copies = self.evaluation/"Copies"
|
||||
copies.mkdir()
|
||||
source = copies/"Copie01.pdf"
|
||||
source.touch()
|
||||
self.app.tree.selection_set("crop_blank_margins")
|
||||
self.app.update()
|
||||
self.assertEqual(self.app.current_step.id, "crop_blank_margins")
|
||||
self.assertEqual(str(self.app.skip_button.cget("state")), "normal")
|
||||
self.app.copy_var.set(source.name)
|
||||
self.app._target_selected_copy()
|
||||
command = self.app._make_command()
|
||||
self.assertEqual(command[-4:], ["crop-margins", str(source), "--workers", "5"])
|
||||
self.app._skip_step()
|
||||
self.assertEqual(self.app.state_store.step("crop_blank_margins")["status"], "skipped")
|
||||
|
||||
def test_redo_targets_selected_copy_and_copies_runnable_command(self):
|
||||
for folder in ("Copies", "Copies Originales"):
|
||||
(self.evaluation / folder).mkdir()
|
||||
|
||||
@@ -0,0 +1,163 @@
|
||||
import unittest
|
||||
|
||||
import cv2
|
||||
import numpy as np
|
||||
|
||||
from copienator.ink_detection import detect_bounds, _paper_blur
|
||||
|
||||
|
||||
class InkDetectionTests(unittest.TestCase):
|
||||
def test_fast_background_blur_matches_opencv_pixel_for_pixel(self):
|
||||
rng = np.random.default_rng(42)
|
||||
for shape in ((97,131), (241,319)):
|
||||
gray = rng.integers(0,256,shape,dtype=np.uint8)
|
||||
for dpi in (100,150,200,300):
|
||||
with self.subTest(shape=shape,dpi=dpi):
|
||||
expected = cv2.GaussianBlur(gray,(0,0),dpi/8)
|
||||
np.testing.assert_array_equal(_paper_blur(gray,dpi/8),expected)
|
||||
|
||||
def scan(self):
|
||||
image = np.full((1200, 850, 3), 255, np.uint8)
|
||||
for y in range(20, 1200, 20):
|
||||
cv2.line(image, (0, y), (849, y), (160, 155, 205), 1)
|
||||
for x in range(10, 850, 20):
|
||||
cv2.line(image, (x, 0), (x, 1199), (160, 155, 205), 1)
|
||||
for y in (100, 180, 400, 480, 800, 880, 1050, 1130):
|
||||
cv2.ellipse(image, (25, y), (12, 20), 0, 0, 300, (155,155,155), 2)
|
||||
cv2.putText(image, 'x + y = 2', (110, 400), cv2.FONT_HERSHEY_SIMPLEX,
|
||||
1, (20, 90, 190), 2)
|
||||
cv2.putText(image, 'answer = 42', (110, 700), cv2.FONT_HERSHEY_SIMPLEX,
|
||||
1, (20, 90, 190), 2)
|
||||
return image
|
||||
|
||||
def test_crops_past_holes_and_grid(self):
|
||||
r = detect_bounds(self.scan(), dpi=150)
|
||||
self.assertGreater(r['top_px'], 250)
|
||||
self.assertLess(r['top_px'], 370)
|
||||
self.assertGreater(r['bottom_px'], 700)
|
||||
self.assertLess(r['bottom_px'], 800)
|
||||
|
||||
def test_isolated_margin_note_survives(self):
|
||||
image = self.scan()
|
||||
cv2.putText(image, '1', (4, 1110), cv2.FONT_HERSHEY_SIMPLEX,
|
||||
.65, (20, 90, 190), 2)
|
||||
self.assertGreater(detect_bounds(image, dpi=150)['bottom_px'], 1110)
|
||||
|
||||
def test_long_fraction_bar_on_grid_survives(self):
|
||||
for colour in ((20,90,190), (20,20,20), (190,20,20)):
|
||||
with self.subTest(colour=colour):
|
||||
image = self.scan()
|
||||
cv2.line(image, (100, 1020), (700, 1020), colour, 3)
|
||||
self.assertGreater(detect_bounds(image, dpi=150)['bottom_px'], 1020)
|
||||
|
||||
def test_black_annotation_survives(self):
|
||||
image = self.scan()
|
||||
cv2.putText(image, 'note', (600, 1100), cv2.FONT_HERSHEY_SIMPLEX,
|
||||
.7, (30,30,30), 2)
|
||||
self.assertGreater(detect_bounds(image, dpi=150)['bottom_px'], 1100)
|
||||
|
||||
def test_blank_and_pencil_only_pages_are_retained(self):
|
||||
for pencil in (False, True):
|
||||
image = np.full((1200,850,3),255,np.uint8)
|
||||
if pencil:
|
||||
cv2.putText(image, 'pencil', (100,600), cv2.FONT_HERSHEY_SIMPLEX,
|
||||
1, (195,195,195), 2)
|
||||
r = detect_bounds(image, dpi=150)
|
||||
self.assertEqual((r['top_px'],r['bottom_px']), (0,1200))
|
||||
self.assertEqual(r['status'], 'review-no-ink-seeds')
|
||||
|
||||
def test_skewed_colour_scan(self):
|
||||
matrix = cv2.getRotationMatrix2D((425,600),2,1)
|
||||
image = cv2.warpAffine(self.scan(),matrix,(850,1200),borderValue=(255,255,255))
|
||||
r = detect_bounds(image,dpi=150)
|
||||
self.assertGreater(r['top_px'],250)
|
||||
self.assertLess(r['top_px'],370)
|
||||
self.assertGreater(r['bottom_px'],715)
|
||||
self.assertLess(r['bottom_px'],820)
|
||||
|
||||
def test_weak_stroke_attached_to_ink_is_recovered(self):
|
||||
image = np.full((1200,850,3),255,np.uint8)
|
||||
cv2.rectangle(image,(400,500),(410,530),(20,90,190),-1)
|
||||
cv2.rectangle(image,(400,531),(410,540),(130,170,205),-1)
|
||||
r = detect_bounds(image,dpi=150,padding_mm=0,min_crop_mm=0)
|
||||
self.assertGreaterEqual(r['bottom_px'],541)
|
||||
|
||||
def dark_grid(self):
|
||||
image = np.full((1200,850,3),255,np.uint8)
|
||||
for y in range(20,1200,20):
|
||||
cv2.line(image,(0,y),(849,y),(65,65,65),1)
|
||||
for x in range(10,850,20):
|
||||
cv2.line(image,(x,0),(x,1199),(65,65,65),1)
|
||||
cv2.putText(image,'x + y = 2',(100,400),cv2.FONT_HERSHEY_SIMPLEX,
|
||||
1,(20,20,20),2)
|
||||
cv2.putText(image,'answer = 42',(100,700),cv2.FONT_HERSHEY_SIMPLEX,
|
||||
1,(20,20,20),2)
|
||||
return image
|
||||
|
||||
def test_dark_grid_does_not_seed_entire_page(self):
|
||||
r = detect_bounds(self.dark_grid(),dpi=150)
|
||||
self.assertTrue(r['paper_cleanup'])
|
||||
self.assertGreater(r['top_px'],250)
|
||||
self.assertLess(r['top_px'],370)
|
||||
self.assertGreater(r['bottom_px'],700)
|
||||
self.assertLess(r['bottom_px'],820)
|
||||
|
||||
def test_faint_isolated_note_on_dark_grid_survives(self):
|
||||
image = self.dark_grid()
|
||||
cv2.putText(image,'pencil',(100,1100),cv2.FONT_HERSHEY_SIMPLEX,
|
||||
.7,(190,190,190),2)
|
||||
self.assertGreater(detect_bounds(image,dpi=150)['bottom_px'],1100)
|
||||
|
||||
def test_large_unruled_diagram_is_not_paper(self):
|
||||
image = np.full((1200,850,3),255,np.uint8)
|
||||
cv2.rectangle(image,(100,100),(750,1100),(20,20,20),3)
|
||||
r = detect_bounds(image,dpi=150)
|
||||
self.assertFalse(r['paper_cleanup'])
|
||||
self.assertLess(r['top_px'],100)
|
||||
self.assertGreater(r['bottom_px'],1100)
|
||||
|
||||
def test_dark_grid_preserves_black_fraction_bar(self):
|
||||
image = self.dark_grid()
|
||||
cv2.line(image,(150,1020),(700,1020),(0,0,0),3)
|
||||
self.assertGreater(detect_bounds(image,dpi=150)['bottom_px'],1020)
|
||||
|
||||
def test_disconnected_dark_grid_is_still_recognized(self):
|
||||
image = self.dark_grid()
|
||||
for x in range(170,850,170):
|
||||
image[:,x:x+5] = 255
|
||||
for y in range(200,1200,200):
|
||||
image[y:y+5,:] = 255
|
||||
cv2.putText(image,'x + y = 2',(100,400),cv2.FONT_HERSHEY_SIMPLEX,
|
||||
1,(20,20,20),2)
|
||||
r = detect_bounds(image,dpi=150)
|
||||
self.assertTrue(r['paper_cleanup'])
|
||||
self.assertGreater(r['top_px'],250)
|
||||
self.assertLess(r['top_px'],370)
|
||||
self.assertGreater(r['bottom_px'],700)
|
||||
self.assertLess(r['bottom_px'],850)
|
||||
|
||||
def test_repeated_dark_holes_on_either_side(self):
|
||||
for right in (False, True):
|
||||
with self.subTest(right=right):
|
||||
image = self.dark_grid()
|
||||
x = 820 if right else 25
|
||||
for y in (110, 370, 630, 890, 1130):
|
||||
cv2.ellipse(image, (x,y), (12,20), 0, 0, 300, (35,35,35), 2)
|
||||
r = detect_bounds(image,dpi=150)
|
||||
self.assertLess(r['bottom_px'],850)
|
||||
# The same column can contain handwriting as well as holes.
|
||||
cv2.putText(image,'7',(x-5,1060),cv2.FONT_HERSHEY_SIMPLEX,
|
||||
.7,(20,20,20),2)
|
||||
r = detect_bounds(image,dpi=150)
|
||||
self.assertGreater(r['bottom_px'],1060)
|
||||
|
||||
def test_faded_neutral_grid(self):
|
||||
image = self.dark_grid()
|
||||
image[np.all(image == 65,axis=2)] = 145
|
||||
r = detect_bounds(image,dpi=150)
|
||||
self.assertTrue(r['paper_cleanup'])
|
||||
self.assertLess(r['bottom_px'],850)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user