sebastien/Copies
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Specific request to detect names.

Browse Source
This commit is contained in:
sebastien committed 2026-10-26 08:09:53 +01:00
1 parent d2aa4faa1b
commit 3abd7873e6
8 files changed
+395 -9
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

copienator/commands/cutleft.py
+5
View File
@@ -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:
copienator/commands/gemini_for_labels.py
+56 -7
View File
@@ -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
copienator/configuration.py
+2
View File
@@ -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
copienator/name_grids.py
+155
View File
@@ -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