From 3abd7873e606b3bf7b486b1d32696fad6281b5a0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Miquel?= Date: Mon, 26 Oct 2026 08:09:53 +0100 Subject: [PATCH] Specific request to detect names. --- copienator/commands/cutleft.py | 5 + copienator/commands/gemini_for_labels.py | 63 ++++++++- copienator/configuration.py | 2 + copienator/name_grids.py | 155 +++++++++++++++++++++++ default_config.py | 5 + docs/guide_utilisateur_gui.md | 19 +++ tests/test_marked_copies.py | 10 +- tests/test_name_grids.py | 145 +++++++++++++++++++++ 8 files changed, 395 insertions(+), 9 deletions(-) create mode 100644 copienator/name_grids.py create mode 100644 tests/test_name_grids.py diff --git a/copienator/commands/cutleft.py b/copienator/commands/cutleft.py index 69543f2..b9cc5dd 100644 --- a/copienator/commands/cutleft.py +++ b/copienator/commands/cutleft.py @@ -224,6 +224,11 @@ def save_results( for index in range(1, len(splits) + 1): print(f"Saved: {base_name}_{index:02d}.jpg") print(f"Saved schema: {base_name}_schema.json") + if pdf_path.is_file(): + from copienator.name_grids import save_crop, build_grids + pages = get_pdf_pages(pdf_path) + save_crop(pages[0], pdf_path, output_dir / "Names", RENDER_DPI) + build_grids(output_dir / "Names", pdf_path.parent) def regenerate_full_page_image(workspace: EvaluationWorkspace, image_path: Path) -> None: diff --git a/copienator/commands/gemini_for_labels.py b/copienator/commands/gemini_for_labels.py index e722384..6c17e5b 100644 --- a/copienator/commands/gemini_for_labels.py +++ b/copienator/commands/gemini_for_labels.py @@ -16,6 +16,7 @@ from pydantic import BaseModel, Field, model_validator from copienator import configuration as config from copienator.gemini_config import fresh_seed, thinking_config +from copienator.name_grids import identify_names, MISSING from copienator.copy_errors import marked_copy_paths from copienator.orientation import record_detection, reports as orientation_reports from copienator import ( @@ -145,11 +146,27 @@ be missing. Since this copy isn't the first part of a sequence, simply set the name to `\"Continued\"`.""" + +def labels_only_prompt(template: str) -> str: + """Keep label instructions/context, removing every name-related instruction.""" + start = template.index("You also need to give me the student name.") + end = template.index("You may find the same label", start) + template = template[:start] + """Answer with a JSON object containing a `list` of bounding boxes and labels. +The box_2d is [ymin, xmin, ymax, xmax] normalized to 0-1000. +Example: {"list": [{"box_2d": [10, 20, 30, 40], "label": "Ex 1 : 1)"}]} +Order boxes column by column, top to bottom within each column. + +""" + template[end:] + for ending in ("Here's a list of the names", "Since this copy isn't the first part"): + template = template.split(ending)[0] + return template + + class BoxItem(BaseModel): box_2d: list[int] = Field(description="Bounding box coordinates (e.g., [ymin, xmin, ymax, xmax])") label: str = Field(description="The label associated with the specific box") -class AnnotationData(BaseModel): +class LabelData(BaseModel): upside_down_pages: typing.List[typing.Annotated[int, Field(strict=True, ge=1)]] = Field( default_factory=list, description='1-based column numbers in this image that appear upside down (180 degrees)', @@ -158,11 +175,20 @@ class AnnotationData(BaseModel): 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_labels(self): + if self.error is None and not self.upside_down_pages and "list" not in self.model_fields_set: + raise ValueError("Successful detection must include list") + return self + + +class AnnotationData(LabelData): + name: str = Field(default="Unknown", description="The name identifier") + @model_validator(mode="after") def require_success_payload(self): if self.error is None and not self.upside_down_pages and not {"name", "list"} <= self.model_fields_set: @@ -232,6 +258,7 @@ def generate_request( *, retry: bool = False, full_pages: bool = False, + known_name: str | None = None, ): """Generates request for Gemini with context.""" @@ -241,10 +268,10 @@ def generate_request( context_str = ", ".join([f'"{l}"' for l in context_labels]) if context_labels else "No previous context" if context_labels == []: - text = my_prompt.replace("##labels##", labels)\ + text = (labels_only_prompt(my_prompt) if known_name else my_prompt).replace("##labels##", labels)\ .replace("##names##", names) else: - text = my_prompt2.replace("##labels##", labels)\ + text = (labels_only_prompt(my_prompt2) if known_name else my_prompt2).replace("##labels##", labels)\ .replace("##prev_context##", context_str) if wrong_labels: formatted_wrong_labels = "\n".join(f'- "{label}"' for label in wrong_labels) @@ -297,6 +324,9 @@ students may skip questions. On success, return error=null. 'you may return name="Unknown", list=[] and error=null. Orientation ' 'reporting takes priority over insufficient_crop.') + if known_name: + text = text.replace('name="Unknown", list=[]', 'list=[]') + contents = [ types.Content( role="user", @@ -320,7 +350,7 @@ students may skip questions. On success, return error=null. seed=fresh_seed() if seed is None else seed, max_output_tokens=65535, response_mime_type= "application/json", - response_json_schema= AnnotationData.model_json_schema(), + response_json_schema=(LabelData if known_name else AnnotationData).model_json_schema(), ) return (contents, generate_content_config) @@ -461,11 +491,13 @@ def process_copy_group( valid_labels: set[str], valid_names: set[str], overwrite: bool, + known_name: str | None = None, sleep: Sleep = time.sleep, target_interval: float = TARGET_INTERVAL, progress: Callable[[str, str, float], None] | None = None, ) -> int: """Process one student's image parts sequentially to preserve context.""" + known_name = known_name if known_name in valid_names and known_name not in {"Missing", "Unknown", "Continued"} else None accumulated_labels: list[str] = [] generated = 0 for image_index, image_file in enumerate(files): @@ -518,13 +550,21 @@ def process_copy_group( seed=request_seed, retry=attempt > 0 or full_page_fallback, full_pages=full_page_fallback, + known_name=known_name, ) response = client.models.generate_content( model=MODEL_ID, contents=contents, config=request_config, ) - annotation = AnnotationData.model_validate_json(response.text) + if known_name: + detected = LabelData.model_validate_json(response.text) + # Preserve the established downstream first-part/continuation contract. + part = re.search(r"_(\d+)$", image_file.stem) + saved_name = "Continued" if part and int(part.group(1)) > 1 else known_name + annotation = AnnotationData(**detected.model_dump(), name=saved_name) + else: + annotation = AnnotationData.model_validate_json(response.text) if record_detection(workspace, image_file, annotation.upside_down_pages): output_json.unlink(missing_ok=True) print(f'[{group_key}] Orientation à vérifier avant de valider les labels.') @@ -575,7 +615,7 @@ def process_copy_group( label_retry_count += 1 attempt += 1 continue - if annotation.name not in valid_names: + if not known_name and annotation.name not in valid_names: print( f"Error: {image_file.name} returned unknown name: " f"{annotation.name}" @@ -653,6 +693,14 @@ def run( client = genai.Client(api_key=api_key) groups = group_images(images) + pending_groups = {key: files for key, files in groups.items() + if overwrite or any(not (workspace.copies_dir / f"{image.stem}.json").exists() + for image in files)} + identified_names = identify_names( + workspace, pending_groups, client=client, model=config.MODEL_FOR_NAME_ID, + names_text=names_text, overwrite=overwrite, + thinking_level=config.THINKING_LEVEL_FOR_NAME, + ) total_groups = len(groups) active: dict[str, tuple[str, float]] = {} progress_lock = Lock() @@ -682,6 +730,7 @@ def run( valid_labels=set(all_labels), valid_names=valid_names, overwrite=overwrite, + known_name=(identified_names.get(group_key) if identified_names.get(group_key) != MISSING else None), sleep=sleep, progress=report_progress, ): group_key diff --git a/copienator/configuration.py b/copienator/configuration.py index 8b3dede..d822b32 100644 --- a/copienator/configuration.py +++ b/copienator/configuration.py @@ -34,6 +34,8 @@ else: # Keep new optional settings compatible with older personal configuration files. THINKING_LEVEL_LITE = "low" THINKING_LEVEL_FOR_LABEL = "medium" +MODEL_FOR_NAME_ID = getattr(_configuration, "MODEL_FOR_LABEL_ID", "gemini-3.8-flash") +THINKING_LEVEL_FOR_NAME = "low" THINKING_LEVEL_FLASH = "medium" THINKING_LEVEL_PRO = "high" ALWAYS_CROP = False diff --git a/copienator/name_grids.py b/copienator/name_grids.py new file mode 100644 index 0000000..8e3e9c1 --- /dev/null +++ b/copienator/name_grids.py @@ -0,0 +1,155 @@ +"""First-page name crops, numbered contact sheets, and roster-checked recognition.""" +from __future__ import annotations + +from pathlib import Path + +from google.genai import types +from PIL import Image, ImageDraw +from pydantic import BaseModel, Field + +from copienator import atomic_write_json, read_json +from copienator.annotation_layout import fingerprint +from copienator.filesystem import staged_files +from copienator.gemini_config import fresh_seed, thinking_config +from copienator.utils import natural_key + +COLUMNS, ROWS = 6, 4 +MARGIN_MM = 28 +MISSING = "Missing" + + +def save_crop(page: Image.Image, pdf: Path, directory: Path, dpi: int) -> None: + """Crop independently of the adjustable label strip, from the true top left.""" + width = min(page.width, max(1, round(MARGIN_MM * dpi / 25.4))) + crop = page.crop((0, 0, width, max(1, page.height // 4))) + with staged_files(directory) as staging: + crop.save(staging / f"{pdf.stem}.jpg", "JPEG", quality=95) + atomic_write_json(staging / f"{pdf.stem}.json", { + "source_sha256": fingerprint(pdf), "width_mm": MARGIN_MM, + "height_fraction": 0.25, + }) + + +def build_grids(directory: Path, copies_dir: Path) -> list[dict]: + """Rebuild deterministic 6-by-4 sheets, excluding deleted or changed copies.""" + crops = [] + for path in sorted(directory.glob("*.jpg"), key=natural_key): + pdf = copies_dir / f"{path.stem}.pdf" + try: + if pdf.is_file() and read_json(path.with_suffix('.json'))['source_sha256'] == fingerprint(pdf): + crops.append(path) + except (OSError, ValueError, KeyError, TypeError): + continue + grids_dir = directory / "Grids" + grids = [] + with staged_files(grids_dir, remove=[p.name for p in grids_dir.glob('*') if p.is_file()]) as staging: + for offset in range(0, len(crops), COLUMNS * ROWS): + batch = crops[offset:offset + COLUMNS * ROWS] + images = [] + for path in batch: + with Image.open(path) as image: + images.append(image.convert('RGB')) + width = max(image.width for image in images) + height = max(image.height for image in images) + border, header = 6, 32 + cell_width, cell_height = width + border, height + header + border + sheet = Image.new('RGB', (COLUMNS * cell_width + border, ROWS * cell_height + border), 'black') + draw = ImageDraw.Draw(sheet) + cells = [] + for index in range(COLUMNS * ROWS): + row, column = divmod(index, COLUMNS) + x, y = border + column * cell_width, border + row * cell_height + draw.rectangle((x, y, x + width - 1, y + height + header - 1), fill='white') + if index < len(images): + sheet.paste(images[index], (x, y + header)) + draw.text((x + 5, y + 5), f"R{row + 1} C{column + 1} | {index + 1}", fill='black') + cells.append({'cell': index + 1, 'row': row + 1, 'column': column + 1, + 'copy': batch[index].stem}) + else: + draw.text((x + 5, y + 5), 'EMPTY', fill='black') + filename = f"grid_{len(grids) + 1:03d}.jpg" + sheet.save(staging / filename, 'JPEG', quality=95) + grids.append({'image': filename, 'columns': COLUMNS, 'rows': ROWS, 'cells': cells}) + atomic_write_json(staging / 'manifest.json', grids) + return grids + + +class NameCell(BaseModel): + cell: int = Field(strict=True, ge=1, le=24) + name: str + + +class NameResponse(BaseModel): + cells: list[NameCell] + + +def identify_names(workspace, groups, *, client, model: str, names_text: str, overwrite: bool, + thinking_level: str | None = "low") -> dict[str, str]: + """Identify pending copies before label requests; safely fall back on any failure.""" + directory = workspace.cutleft_dir / 'Names' + result = {stem: MISSING for stem in groups} + if not groups or not directory.is_dir(): + return result # Older workspaces retain their existing label/name workflow. + grids = build_grids(directory, workspace.copies_dir) + roster = {line.strip() for line in names_text.splitlines() if line.strip()} - {MISSING, 'Unknown', 'Continued'} + for grid in grids: + selected = {cell['copy']: cell for cell in grid['cells'] if cell['copy'] in groups} + pending = {} + for stem, cell in selected.items(): + crop_hash = fingerprint(directory / f'{stem}.jpg') + cache_path = directory / f'{stem}_result.json' + try: + cached = read_json(cache_path) + if (not overwrite and cached['crop_sha256'] == crop_hash + and cached['names_text'] == names_text + and cached['name'] in roster | {MISSING}): + result[stem] = cached['name'] + continue + except (OSError, ValueError, KeyError, TypeError): + pass + pending[stem] = (cell, crop_hash, cache_path) + if not pending: + continue + positions = '\n'.join(f"Cell {c['cell']}: row {c['row']}, column {c['column']}" for c in grid['cells']) + prompt = f"""Identify student names from the top-left corners of exam first pages. +The image is a grid with exactly 6 columns and 4 rows. Read rows left to right, +then top to bottom. Thick black borders separate cells; headers show row, column, +and cell number. Each cell belongs to a DIFFERENT student. Never carry a name +from one cell into another. Ignore class names such as MPSI 3. +Occupied cells are exactly: +{positions} +All other cells are EMPTY padding: do not return them. +Return JSON {{"cells": [{{"cell": 1, "name": "exact authorized name"}}]}}. +Return one entry per occupied cell, using its cell number. Match only confidently +readable handwriting to an exact authorized name below. Do not guess or assign +names by elimination. Return "Missing" for an unreadable, absent, or ambiguous name. +Authorized names: +{names_text}""" + print(f"Identifying student names: {grid['image']} ({len(grid['cells'])} occupied cells, model={model}, thinking={thinking_level or 'default'}).", flush=True) + try: + response = client.models.generate_content( + model=model, + contents=[types.Content(role='user', parts=[ + types.Part.from_bytes(data=(directory / 'Grids' / grid['image']).read_bytes(), mime_type='image/jpeg'), + types.Part.from_text(text=prompt), + ])], + config=types.GenerateContentConfig( + http_options=types.HttpOptions(timeout=80_000, retry_options=types.HttpRetryOptions(attempts=1)), + automatic_function_calling=types.AutomaticFunctionCallingConfig(disable=True), + thinking_config=thinking_config(thinking_level), seed=fresh_seed(), + response_mime_type='application/json', response_json_schema=NameResponse.model_json_schema(), + ), + ) + entries = NameResponse.model_validate_json(response.text).cells + by_cell = {} + for entry in entries: + by_cell.setdefault(entry.cell, []).append(entry.name) + for stem, (cell, crop_hash, cache_path) in pending.items(): + candidates = by_cell.get(cell['cell'], []) + name = candidates[0] if len(candidates) == 1 and candidates[0] in roster else MISSING + result[stem] = name + atomic_write_json(cache_path, {'name': name, 'crop_sha256': crop_hash, 'names_text': names_text}) + except Exception as exc: # Remote failures must preserve the existing fallback. + print(f"Name recognition failed for {grid['image']}: {type(exc).__name__}: {exc}; using label/name fallback.") + # Do not cache transport/schema failures: a resumed run should retry. + return result diff --git a/default_config.py b/default_config.py index 8812f7a..e39913d 100644 --- a/default_config.py +++ b/default_config.py @@ -50,6 +50,11 @@ THINKING_LEVEL_LITE = "low" MODEL_FOR_LABEL_ID = "gemini-3.8-flash" THINKING_LEVEL_FOR_LABEL = "low" +# Modèle pour reconnaître les noms sur les grilles de premières pages. +MODEL_FOR_NAME_ID = "gemini-3.8-flash" +THINKING_LEVEL_FOR_NAME = "low" + + # Modèle pour des choses normales MODEL_FLASH_ID = "gemini-3.8-flash" THINKING_LEVEL_FLASH = "medium" diff --git a/docs/guide_utilisateur_gui.md b/docs/guide_utilisateur_gui.md index 9b94586..c8cbcab 100644 --- a/docs/guide_utilisateur_gui.md +++ b/docs/guide_utilisateur_gui.md @@ -134,6 +134,25 @@ son propre signalement ; un signalement de vérification des labels reste actif. ## Labels et regroupement +La validation enregistre aussi le coin supérieur gauche de la première page : +28 mm de large et un quart de la hauteur. Ces images sont conservées dans +`Cutleft/Names`, puis assemblées en grilles de 6 colonnes et 4 lignes dans +`Cutleft/Names/Grids`. Des bordures noires et des numéros identifient chaque case ; +les cases inutilisées portent la mention `EMPTY`. + +La détection des labels commence par reconnaître les noms sur ces grilles avec +Gemini, configuré par `MODEL_FOR_NAME_ID` et `THINKING_LEVEL_FOR_NAME` dans +`config.py` (réflexion `low` par défaut), indépendamment du modèle de détection +des labels. Le modèle reçoit les positions des cases occupées et +la liste des noms autorisés. Un nom absent, illisible, ambigu ou hors liste est +conservé comme `Missing` dans le résultat de cette première passe. Dans ce cas, +la détection habituelle des labels recherche aussi le nom. Si le nom a été +reconnu, les requêtes de labels ne demandent plus de nom et ne transmettent plus +la liste des élèves. Les résultats de noms sont réutilisés à la reprise tant +que les images et la liste des noms restent identiques ; une relance avec +écrasement les recalcule. Pour les anciennes découpes sans grilles, le +fonctionnement antérieur reste disponible ; relancer la découpe crée les grilles. + ### Détecter les labels avec Gemini **Effet :** reconnaît les identifiants de questions et leur position dans les copies. diff --git a/tests/test_marked_copies.py b/tests/test_marked_copies.py index 5dbb540..ae90a6e 100644 --- a/tests/test_marked_copies.py +++ b/tests/test_marked_copies.py @@ -39,9 +39,14 @@ class MarkedCopiesTests(unittest.TestCase): review.current_width_offset = 0 review.default_max_per_file = 5 review.current_max_per_file = 1 + review.review_settings = {} review.root = Mock() review.stop_prefetch = threading.Event() - review.load_current_image = Mock() + def load_current_image(): + review.current_shift, review.current_width_offset, review.current_max_per_file = ( + review.review_settings.get(review.index, (0, 0, review.default_max_per_file)) + ) + review.load_current_image = Mock(side_effect=load_current_image) review.update_display = Mock() image = Image.new("RGB", (8, 8), "white") review.current_result = (image, [image], {"total_pages": 1, "columns_per_file": [1]}) @@ -79,7 +84,8 @@ class MarkedCopiesTests(unittest.TestCase): for path in self.files: mark_copy_error(self.workspace, path, "Review needed") review = self.reviewer() - review.on_next(None) + with patch.object(cutleft, "get_pdf_pages", return_value=[review.current_result[0]]): + review.on_next(None) self.assertTrue((review.output_dir / "Copie01_01.jpg").is_file()) self.assertEqual(set(copy_errors(self.workspace)), {"Copie02.pdf"}) self.assertEqual(review.index, 1) diff --git a/tests/test_name_grids.py b/tests/test_name_grids.py new file mode 100644 index 0000000..74440e8 --- /dev/null +++ b/tests/test_name_grids.py @@ -0,0 +1,145 @@ +import json +import tempfile +import unittest +from pathlib import Path +from unittest.mock import Mock, patch + +from PIL import Image + +from copienator import EvaluationWorkspace, ExitCode, read_json +from copienator import name_grids as names +from copienator.commands import cutleft, gemini_for_labels as labels + + +class NameGridTests(unittest.TestCase): + def setUp(self): + temporary = tempfile.TemporaryDirectory() + self.addCleanup(temporary.cleanup) + self.workspace = EvaluationWorkspace(Path(temporary.name)) + self.workspace.copies_dir.mkdir() + self.workspace.cutleft_dir.mkdir() + self.directory = self.workspace.cutleft_dir / 'Names' + (self.workspace.root / 'names').write_text('Alice\nBob\n') + (self.workspace.root / 'labels').write_text('Ex 1\n') + self.client = Mock() + + def crop(self, number): + pdf = self.workspace.copies_dir / f'Copie{number:02d}.pdf' + pdf.write_bytes(b'pdf') + page = Image.new('RGB', (1654, 2338), 'white') + page.putpixel((0, 0), (0, 0, 0)) + names.save_crop(page, pdf, self.directory, 200) + (self.workspace.cutleft_dir / f'{pdf.stem}_01.jpg').write_bytes(b'image') + return pdf + + def identify(self, groups, overwrite=False): + return names.identify_names(self.workspace, groups, client=self.client, model='test', + names_text='Alice\nBob\n', overwrite=overwrite) + + def test_crop_geometry_and_partial_grid_mapping(self): + for number in range(1, 26): + self.crop(number) + with Image.open(self.directory / 'Copie01.jpg') as crop: + self.assertEqual(crop.size, (220, 584)) + self.assertLess(crop.getpixel((0, 0))[0], 80) + grids = names.build_grids(self.directory, self.workspace.copies_dir) + self.assertEqual([len(grid['cells']) for grid in grids], [24, 1]) + self.assertEqual(grids[0]['cells'][6], {'cell': 7, 'row': 2, 'column': 1, 'copy': 'Copie07'}) + self.assertEqual(grids[1]['cells'][0]['copy'], 'Copie25') + with Image.open(self.directory / 'Grids' / grids[1]['image']) as grid: + self.assertEqual(grid.size, (1362, 2494)) + self.assertLess(grid.getpixel((228, 500))[0], 40) + # Deleted/stale PDFs cannot be assigned names from old crops. + (self.workspace.copies_dir / 'Copie25.pdf').unlink() + (self.workspace.copies_dir / 'Copie24.pdf').write_bytes(b'changed') + grids = names.build_grids(self.directory, self.workspace.copies_dir) + self.assertEqual(len(grids), 1) + self.assertEqual(len(grids[0]['cells']), 23) + self.assertFalse((self.directory / 'Grids' / 'grid_002.jpg').exists()) + + def test_validation_cache_and_roster_changes(self): + for number in range(1, 5): + self.crop(number) + self.client.models.generate_content.return_value = Mock(text=json.dumps({'cells': [ + {'cell': 1, 'name': 'Alice'}, {'cell': 2, 'name': 'Outsider'}, + {'cell': 3, 'name': 'Bob'}, {'cell': 3, 'name': 'Alice'}, + {'cell': 24, 'name': 'Bob'}, + ]})) + groups = {f'Copie{i:02d}': [] for i in range(1, 5)} + expected = {'Copie01': 'Alice', **{f'Copie{i:02d}': 'Missing' for i in range(2, 5)}} + self.assertEqual(self.identify(groups), expected) + request = self.client.models.generate_content.call_args.kwargs + self.assertEqual(request['config'].thinking_config.thinking_level.value, 'LOW') + text = request['contents'][0].parts[1].text + self.assertIn('6 columns and 4 rows', text) + self.assertIn('Cell 4: row 1, column 4', text) + self.assertNotIn('Cell 5:', text) + self.assertEqual(self.identify(groups), expected) + self.assertEqual(self.client.models.generate_content.call_count, 1) + self.identify(groups, overwrite=True) + self.assertEqual(self.client.models.generate_content.call_count, 2) + names.identify_names(self.workspace, groups, client=self.client, model='test', + names_text='Bob\n', overwrite=False) + self.assertEqual(read_json(self.directory / 'Copie01_result.json')['name'], 'Missing') + + def test_api_failure_is_retried_on_resume(self): + self.crop(1) + self.client.models.generate_content.side_effect = TimeoutError() + self.assertEqual(self.identify({'Copie01': []}), {'Copie01': 'Missing'}) + self.assertFalse((self.directory / 'Copie01_result.json').exists()) + self.identify({'Copie01': []}) + self.assertEqual(self.client.models.generate_content.call_count, 2) + + def test_known_and_missing_names_choose_different_label_requests(self): + for number in (1, 2): + self.crop(number) + (self.workspace.cutleft_dir / 'Copie01_02.jpg').write_bytes(b'image') + self.client.models.generate_content.side_effect = [ + Mock(text='{"cells":[{"cell":1,"name":"Alice"},{"cell":2,"name":"Missing"}]}'), + Mock(text='{"list":[{"box_2d":[1,2,3,4],"label":"Ex 1"}]}'), Mock(text='{"list":[]}'), + Mock(text='{"name":"Bob","list":[]}'), + ] + with patch.object(labels.config, 'MODEL_FOR_NAME_ID', 'custom-name-model'), \ + patch.object(labels.config, 'THINKING_LEVEL_FOR_NAME', 'medium'): + status = labels.run(self.workspace, [self.workspace.root], client=self.client, + sleep=lambda _: None, max_workers=1) + self.assertEqual(status, ExitCode.SUCCESS) + requests = self.client.models.generate_content.call_args_list + self.assertEqual(len(requests), 4) + self.assertEqual(requests[0].kwargs['model'], 'custom-name-model') + self.assertEqual(requests[0].kwargs['config'].thinking_config.thinking_level.value, 'MEDIUM') + self.assertEqual(requests[1].kwargs['model'], labels.MODEL_ID) + for request in requests[1:3]: + text = request.kwargs['contents'][0].parts[1].text + self.assertNotIn('Alice', text) + self.assertNotIn('Bob', text) + self.assertNotIn('student name', text) + self.assertNotIn('name', request.kwargs['config'].response_json_schema['properties']) + self.assertIn('Alice', requests[3].kwargs['contents'][0].parts[1].text) + for stem, expected in [('Copie01_01', 'Alice'), ('Copie01_02', 'Continued'), ('Copie02_01', 'Bob')]: + self.assertEqual(read_json(self.workspace.copies_dir / f'{stem}.json')['name'], expected) + self.client.models.generate_content.reset_mock() + labels.run(self.workspace, [self.workspace.root], client=self.client, sleep=lambda _: None) + self.client.models.generate_content.assert_not_called() + + def test_save_label_crops_also_saves_name_grid(self): + pdf = self.workspace.copies_dir / 'Copie01.pdf' + pdf.write_bytes(b'pdf') + page = Image.new('RGB', (1654, 2338), 'white') + with patch.object(cutleft, 'get_pdf_pages', return_value=[page]): + result = cutleft.process_single_pdf(pdf, automatic_margin=False) + cutleft.save_results(result, pdf, self.workspace.cutleft_dir) + self.assertTrue((self.directory / 'Copie01.jpg').is_file()) + self.assertEqual(read_json(self.directory / 'Grids' / 'manifest.json')[0]['cells'][0]['copy'], 'Copie01') + + def test_known_name_survives_label_retry(self): + self.crop(1) + image = self.workspace.cutleft_dir / 'Copie01_01.jpg' + self.client.models.generate_content.side_effect = [TimeoutError(), Mock(text='{"list":[]}')] + labels.process_copy_group(self.workspace, 'Copie01', [image], client=self.client, + labels_text='Ex 1', names_text='Alice\nBob', valid_labels={'Ex 1'}, + valid_names={'Alice', 'Bob'}, overwrite=False, known_name='Alice', + sleep=lambda _: None, target_interval=0) + self.assertEqual(read_json(self.workspace.copies_dir / 'Copie01_01.json')['name'], 'Alice') + for call in self.client.models.generate_content.call_args_list: + self.assertNotIn('Alice', call.kwargs['contents'][0].parts[1].text)