diff --git a/copienator/commands/cutleft.py b/copienator/commands/cutleft.py index e087640..65ad166 100644 --- a/copienator/commands/cutleft.py +++ b/copienator/commands/cutleft.py @@ -1,6 +1,7 @@ from __future__ import annotations import argparse +import re import threading import tkinter as tk from tkinter import messagebox @@ -19,6 +20,7 @@ from copienator import ( EvaluationWorkspace, ExitCode, atomic_write_json, + read_json, execute, target_parser, workspace_from_target, @@ -221,6 +223,42 @@ def save_results( print(f"Saved schema: {base_name}_schema.json") +def regenerate_full_page_image(workspace: EvaluationWorkspace, image_path: Path) -> None: + """Replace one strip montage with the same pages at their full width.""" + match = re.fullmatch(r"(.+)_(\d+)", image_path.stem) + if not match or image_path.parent != workspace.cutleft_dir: + raise ValueError(f"Cannot map Cutleft image to PDF pages: {image_path}") + stem, part = match.group(1), int(match.group(2)) - 1 + schema_path = workspace.cutleft_dir / f"{stem}_schema.json" + schema = read_json(schema_path) + counts = schema["columns_per_file"] + if (not isinstance(counts, list) or not counts + or any(type(count) is not int or count < 1 for count in counts) + or not 0 <= part < len(counts)): + raise ValueError("Invalid Cutleft page distribution") + first = sum(counts[:part]) + last = first + counts[part] + page_numbers = set(range(first + 1, last + 1)) + if page_numbers <= set(schema.get("full_width_pages", [])): + raise ValueError("Cutleft image already contains full pages") + pages = get_pdf_pages(workspace.copies_dir / f"{stem}.pdf") + if len(pages) != sum(counts): + raise ValueError("Cutleft schema does not match the PDF page count") + combined = stitch_images(pages[first:last]) + if combined is None: + raise ValueError("No pages available for full-page fallback") + crops = {crop["page"]: crop for crop in schema.get("page_crops", [])} + for index in range(first, last): + width, height = pages[index].size + crops[index + 1] = {"page": index + 1, "left_px": 0, "right_px": width, + "width_px": width, "height_px": height} + schema["page_crops"] = [crops[key] for key in sorted(crops)] + schema["full_width_pages"] = sorted(set(schema.get("full_width_pages", [])) | page_numbers) + with staged_files(workspace.cutleft_dir) as staging: + combined.save(staging / image_path.name, "JPEG", quality=95) + atomic_write_json(staging / schema_path.name, schema) + + class ImageReviewer: def __init__( self, diff --git a/copienator/commands/gemini_for_labels.py b/copienator/commands/gemini_for_labels.py index 9801294..2c68381 100644 --- a/copienator/commands/gemini_for_labels.py +++ b/copienator/commands/gemini_for_labels.py @@ -12,7 +12,7 @@ from pathlib import Path from google import genai from google.genai import types -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, model_validator from copienator import configuration as config from copienator.gemini_config import fresh_seed, thinking_config @@ -149,10 +149,20 @@ class BoxItem(BaseModel): label: str = Field(description="The label associated with the specific box") class AnnotationData(BaseModel): - name: str = Field(description="The name identifier") - list: typing.List[BoxItem] = Field( # noqa: UP006 - field name shadows list - description="List of bounding box items" + error: typing.Literal["insufficient_crop"] | None = Field( + default=None, description="Report insufficient_crop when the crop hides information needed to identify labels" ) + error_reason: str | None = Field(default=None, description="Explanation of the crop error") + name: str = Field(default="Unknown", description="The name identifier") + list: typing.List[BoxItem] = Field( # noqa: UP006 - field name shadows list + default_factory=list, description="List of bounding box items" + ) + + @model_validator(mode="after") + def require_success_payload(self): + if self.error is None and not {"name", "list"} <= self.model_fields_set: + raise ValueError("Successful detection must include name and list") + return self TEXT_CONTEXT_MAX_CHARS = 4000 @@ -216,6 +226,7 @@ def generate_request( seed: int | None = None, *, retry: bool = False, + full_pages: bool = False, ): """Generates request for Gemini with context.""" @@ -263,6 +274,18 @@ def generate_request( text = text.replace("##wrong_label_text_context##", "") + text += """ + +If the cropped margins hide labels or exercise headings needed to identify +questions reliably, return error="insufficient_crop", a short error_reason, +and an empty list. Do not guess labels from incomplete cropped information. +Missing answers or an empty page alone are not evidence of insufficient crop: +students may skip questions. On success, return error=null. +""" + if full_pages: + text = text.replace("image of the left columns", "image of the full pages") + text += "\nThis image now contains full pages, in the same reading order. Identify the labels using this expanded view." + contents = [ types.Content( role="user", @@ -399,6 +422,23 @@ def _existing_context(output_json: Path) -> list[str]: return [] +def image_contains_full_pages(workspace: EvaluationWorkspace, image_file: Path) -> bool: + """Recognize expanded images on subsequent detection runs.""" + match = re.fullmatch(r"(.+)_(\d+)", image_file.stem) + if not match: + return False + try: + schema = read_json(workspace.cutleft_dir / f"{match.group(1)}_schema.json") + counts = schema["columns_per_file"] + part = int(match.group(2)) - 1 + if not 0 <= part < len(counts) or any(type(n) is not int or n < 1 for n in counts): + return False + first = sum(counts[:part]) + 1 + return set(range(first, first + counts[part])) <= set(schema.get("full_width_pages", [])) + except (OSError, KeyError, TypeError, ValueError): + return False + + def process_copy_group( workspace: EvaluationWorkspace, group_key: str, @@ -433,13 +473,14 @@ def process_copy_group( label_retry_count = 0 wrong_labels: list[str] = [] request_seed = None + full_page_fallback = image_contains_full_pages(workspace, image_file) while attempt < MAX_ATTEMPTS: if progress: progress(group_key, f"{image_file.name}, attempt {attempt + 1}/{MAX_ATTEMPTS}", started) if attempt > 0: print(f"[{group_key}] Retrying {image_file.name} in {10 * attempt}s...") sleep(10 * attempt) - level = "low" if attempt > 0 else config.THINKING_LEVEL_FOR_LABEL + level = "low" if attempt > 0 or full_page_fallback else config.THINKING_LEVEL_FOR_LABEL request_seed = fresh_seed(request_seed) print( f"[{group_key}] {image_file.name}: attempt {attempt + 1}/{MAX_ATTEMPTS} " @@ -464,7 +505,8 @@ def process_copy_group( text_context, text_context_file, seed=request_seed, - retry=attempt > 0, + retry=attempt > 0 or full_page_fallback, + full_pages=full_page_fallback, ) response = client.models.generate_content( model=MODEL_ID, @@ -472,6 +514,25 @@ def process_copy_group( config=request_config, ) annotation = AnnotationData.model_validate_json(response.text) + if annotation.error == "insufficient_crop": + print(f"[{group_key}] {image_file.name}: insufficient crop: " + f"{annotation.error_reason or 'labels require a wider view'}") + if full_page_fallback: + raise IncompleteCopyError(files[image_index:]) + # Keep numbering and page grouping stable for downstream consumers. + from copienator.commands.cutleft import regenerate_full_page_image + try: + regenerate_full_page_image(workspace, image_file) + except Exception as exc: + print(f"[{group_key}] Full-page regeneration failed: {exc}") + raise IncompleteCopyError(files[image_index:]) from exc + output_json.unlink(missing_ok=True) + full_page_fallback = True + attempt = 0 + label_retry_count = 0 + wrong_labels = [] + print(f"[{group_key}] Regenerated {image_file.name} with full pages; resending to Gemini.") + continue unknown = [ item.label for item in annotation.list @@ -512,11 +573,13 @@ def process_copy_group( annotation.list = sort_boxes_for_image( workspace, image_file, annotation.list ) - atomic_write_json(output_json, annotation.model_dump()) + atomic_write_json(output_json, annotation.model_dump(exclude={"error", "error_reason"})) accumulated_labels.extend(box.label for box in annotation.list) generated += 1 print(f"[{group_key}] Saved {image_file.name} after {time.monotonic() - started:.0f}s.") break + except IncompleteCopyError: + raise except KeyboardInterrupt: raise except Exception as exc: # noqa: BLE001 - remote API retry boundary diff --git a/tests/test_label_copy_flags.py b/tests/test_label_copy_flags.py index 0daf0fd..8251ec0 100644 --- a/tests/test_label_copy_flags.py +++ b/tests/test_label_copy_flags.py @@ -1,6 +1,4 @@ -import os import tempfile -import tkinter as tk import unittest from pathlib import Path from queue import Queue @@ -159,29 +157,6 @@ class LabelCopyFlagTests(unittest.TestCase): self.assertEqual(copy_errors(self.workspace), {self.pdf.name: 'plotting'}) self.assertFalse(self.pdf.with_suffix('.json').exists()) - @unittest.skipUnless(os.environ.get('DISPLAY'), 'Tk requires a display') - def test_exclamation_key_flags_current_copy_in_real_viewer(self): - root = tk.Tk() - try: - image_queue = Queue() - image_queue.put((Image.new('RGB', (100, 100), 'white'), - self.pdf.with_name('Copie01_01.json'), - {'copie': 'Copie01', 'part': 1, 'name': 'Student', - 'schema': {'columns_per_file': [1, 1]}})) - viewer = plotting.ImageViewer(root, self.workspace, set(), image_queue) - root.update() - root.focus_force() - root.update() - root.event_generate('', keysym='exclam') - root.update() - self.assertEqual(viewer.skipped_copies, {'Copie01'}) - self.assertFalse(viewer.is_viewing) - self.assertIn(self.pdf.name, copy_errors(self.workspace)) - finally: - for callback in root.tk.splitlist(root.tk.call('after', 'info')): - root.after_cancel(callback) - root.destroy() - def test_empty_marked_selection_does_not_process_whole_evaluation(self): with patch.object(self.workspace.__class__, 'require_files'), \ patch.object(gemini_for_labels, 'selected_images') as select: diff --git a/tests/test_label_full_pages.py b/tests/test_label_full_pages.py new file mode 100644 index 0000000..21b3c96 --- /dev/null +++ b/tests/test_label_full_pages.py @@ -0,0 +1,107 @@ +import json +import tempfile +import unittest +from pathlib import Path +from unittest.mock import Mock, patch + +from PIL import Image + +from copienator import EvaluationWorkspace, read_json +from copienator.commands import cutleft, gemini_for_labels as labels + + +class FullPageFallbackTests(unittest.TestCase): + def setUp(self): + temporary = tempfile.TemporaryDirectory() + self.addCleanup(temporary.cleanup) + self.workspace = EvaluationWorkspace(Path(temporary.name)) + self.workspace.cutleft_dir.mkdir() + self.workspace.copies_dir.mkdir() + self.images = [self.workspace.cutleft_dir / f'Copie01_{i:02d}.jpg' for i in (1, 2, 3)] + for image in self.images: + image.write_bytes(b'original crop') + self.schema_path = self.workspace.cutleft_dir / 'Copie01_schema.json' + self.schema_path.write_text(json.dumps({'columns_per_file': [1, 2, 1], + 'full_width_pages': [], 'total_pages': 4})) + self.pages = [Image.new('RGB', (100, 200), color) for color in ('red', 'green', 'blue', 'white')] + self.client = Mock() + self.error = Mock(text='{"error":"insufficient_crop","error_reason":"Exercise heading clipped"}') + self.success = Mock(text='{"name":"Student","list":[{"box_2d":[1,2,3,4],"label":"Ex 1"}]}') + + def process(self): + return labels.process_copy_group( + self.workspace, 'Copie01', self.images, client=self.client, + labels_text='Ex 1', names_text='Student', valid_labels={'Ex 1'}, + valid_names={'Student'}, overwrite=False, sleep=lambda _: None, target_interval=0, + ) + + def test_regenerates_only_affected_pages_and_updates_metadata(self): + with patch.object(cutleft, 'get_pdf_pages', return_value=self.pages): + cutleft.regenerate_full_page_image(self.workspace, self.images[1]) + self.assertEqual(self.images[0].read_bytes(), b'original crop') + self.assertEqual(self.images[2].read_bytes(), b'original crop') + with Image.open(self.images[1]) as result: + self.assertEqual(result.size, (205, 200)) + self.assertGreater(result.getpixel((40, 50))[1], 100) + self.assertGreater(result.getpixel((150, 50))[2], 200) + schema = read_json(self.schema_path) + self.assertEqual(schema['columns_per_file'], [1, 2, 1]) + self.assertEqual(schema['full_width_pages'], [2, 3]) + self.assertEqual([crop['right_px'] for crop in schema['page_crops']], [100, 100]) + with self.assertRaisesRegex(ValueError, 'already contains full pages'): + cutleft.regenerate_full_page_image(self.workspace, self.images[1]) + + def test_error_regenerates_and_resends_with_context(self): + self.client.models.generate_content.side_effect = [self.success, self.error, self.success, self.success] + with patch.object(cutleft, 'get_pdf_pages', return_value=self.pages): + self.assertEqual(self.process(), 3) + calls = self.client.models.generate_content.call_args_list + retry_parts = calls[2].kwargs['contents'][0].parts + self.assertEqual(retry_parts[0].inline_data.data, self.images[1].read_bytes()) + self.assertIn('now contains full pages', retry_parts[1].text) + self.assertIn('"Ex 1"', retry_parts[1].text) + result = read_json(self.workspace.copies_dir / 'Copie01_02.json') + self.assertEqual(set(result), {'name', 'list'}) + self.assertEqual(result['list'][0]['label'], 'Ex 1') + + def test_second_crop_error_stops_and_defers_following_images(self): + self.client.models.generate_content.side_effect = [self.error, self.error] + with patch.object(cutleft, 'get_pdf_pages', return_value=self.pages), \ + self.assertRaises(labels.IncompleteCopyError) as caught: + self.process() + self.assertEqual(caught.exception.images, self.images) + self.assertEqual(self.client.models.generate_content.call_count, 2) + self.assertFalse(list(self.workspace.copies_dir.glob('*.json'))) + + def test_bad_mapping_leaves_original_image_and_schema_intact(self): + original_schema = self.schema_path.read_bytes() + self.client.models.generate_content.return_value = self.error + with patch.object(cutleft, 'get_pdf_pages', return_value=self.pages[:1]), \ + self.assertRaises(labels.IncompleteCopyError): + self.process() + self.assertEqual(self.client.models.generate_content.call_count, 1) + self.assertEqual(self.schema_path.read_bytes(), original_schema) + self.assertEqual(self.images[0].read_bytes(), b'original crop') + + def test_full_page_retry_gets_fresh_bounded_attempt_budget(self): + self.client.models.generate_content.side_effect = [TimeoutError(), TimeoutError(), self.error, + TimeoutError(), TimeoutError(), self.success, + self.success, self.success] + with patch.object(cutleft, 'get_pdf_pages', return_value=self.pages): + self.assertEqual(self.process(), 3) + self.assertEqual(self.client.models.generate_content.call_count, 8) + + def test_resumed_full_pages_use_correct_prompt_without_regenerating(self): + schema = read_json(self.schema_path) + schema['full_width_pages'] = [1] + self.schema_path.write_text(json.dumps(schema)) + self.client.models.generate_content.return_value = self.error + with patch.object(cutleft, 'get_pdf_pages') as render, self.assertRaises(labels.IncompleteCopyError): + self.process() + render.assert_not_called() + prompt = self.client.models.generate_content.call_args.kwargs['contents'][0].parts[1].text + self.assertIn('now contains full pages', prompt) + + def test_missing_success_payload_is_not_accepted(self): + with self.assertRaises(ValueError): + labels.AnnotationData.model_validate_json('{}')