288 lines
10 KiB
Python
288 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
from collections.abc import Sequence
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import numpy as np
|
|
from pdf2image import convert_from_path
|
|
from PIL import Image, ImageChops, ImageDraw, ImageFilter
|
|
|
|
from copienator.commands import annotating
|
|
from copienator import utils
|
|
from copienator import (
|
|
EvaluationWorkspace,
|
|
ExitCode,
|
|
atomic_write_json,
|
|
evaluation_parser,
|
|
execute,
|
|
read_json,
|
|
workspace_from_args,
|
|
)
|
|
from copienator.annotation_actions import apply_checkbox_actions, apply_score_overrides
|
|
from copienator.annotation_data import AnnotationData, load_annotation_data
|
|
from copienator.filesystem import staged_files
|
|
|
|
Image.MAX_IMAGE_PIXELS = None
|
|
|
|
|
|
def detect_checks_and_notes(
|
|
output_dir: str | Path,
|
|
) -> tuple[list[dict[str, Any]], Image.Image | None]:
|
|
"""Detect checked boxes and extract handwritten notes from an annotated PDF."""
|
|
directory = Path(output_dir)
|
|
pdf_path = directory / "Concat_annotated.pdf"
|
|
reference_path = directory / "Reference.jpg"
|
|
boxes_path = directory / "checkboxes.json"
|
|
missing = [
|
|
path.name
|
|
for path in (pdf_path, reference_path, boxes_path)
|
|
if not path.is_file()
|
|
]
|
|
if missing:
|
|
print(f"\tMissing annotation input in {directory}: {', '.join(missing)}")
|
|
return [], None
|
|
|
|
boxes = read_json(boxes_path)
|
|
if not isinstance(boxes, list):
|
|
raise TypeError(f"Expected a JSON array in {boxes_path}")
|
|
with Image.open(reference_path) as opened_reference:
|
|
reference = opened_reference.convert("RGB").copy()
|
|
|
|
try:
|
|
pages = convert_from_path(pdf_path, dpi=72)
|
|
except Exception as exc: # noqa: BLE001 - PDF backends expose many errors
|
|
print(f"Error reading PDF {pdf_path}: {exc}")
|
|
return [], None
|
|
if not pages:
|
|
print(f"Error reading PDF {pdf_path}: no page found")
|
|
return [], None
|
|
|
|
user_image = Image.new("RGB", (pages[0].width, sum(page.height for page in pages)))
|
|
current_y = 0
|
|
for page in pages:
|
|
user_image.paste(page.convert("RGB"), (0, current_y))
|
|
current_y += page.height
|
|
if user_image.size != reference.size:
|
|
print(f" Resizing annotated PDF from {user_image.size} to {reference.size}")
|
|
user_image = user_image.resize(reference.size, Image.Resampling.LANCZOS)
|
|
|
|
difference = np.abs(
|
|
np.array(reference).astype(int) - np.array(user_image).astype(int)
|
|
).astype(np.uint8)
|
|
difference_gray = np.mean(difference, axis=2)
|
|
keep_mask = Image.new("L", reference.size, 255)
|
|
mask_draw = ImageDraw.Draw(keep_mask)
|
|
actions: list[dict[str, Any]] = []
|
|
|
|
for raw_box in boxes:
|
|
if not isinstance(raw_box, dict) or "global_box" not in raw_box:
|
|
continue
|
|
x1, y1, x2, y2 = map(int, raw_box["global_box"])
|
|
x1, y1 = max(0, x1), max(0, y1)
|
|
x2, y2 = min(reference.width, x2), min(reference.height, y2)
|
|
region = difference_gray[y1 + 5 : y2 - 5, x1 + 5 : x2 - 5]
|
|
if region.size == 0:
|
|
continue
|
|
density = np.sum(region > 30) / region.size
|
|
if density > 0.05:
|
|
actions.append(raw_box)
|
|
mask_draw.rectangle([x1 - 15, y1 - 15, x2 + 15, y2 + 15], fill=0)
|
|
else:
|
|
mask_draw.rectangle([x1 - 2, y1 - 2, x2 + 2, y2 + 2], fill=0)
|
|
if raw_box.get("type") == "score" and raw_box.get("value") == 0.0:
|
|
mask_draw.rectangle([0, y1 - 10, reference.width, y2 + 10], fill=0)
|
|
|
|
reference_blur = reference.filter(ImageFilter.GaussianBlur(2))
|
|
user_blur = user_image.filter(ImageFilter.GaussianBlur(2))
|
|
diff_image = ImageChops.difference(reference_blur, user_blur).convert("L")
|
|
alpha = np.where(np.array(diff_image) > 50, 255, 0).astype(np.uint8)
|
|
final_alpha = np.minimum(alpha, np.array(keep_mask))
|
|
notes = user_image.convert("RGBA")
|
|
notes.putalpha(Image.fromarray(final_alpha))
|
|
return actions, notes
|
|
|
|
|
|
def has_significant_notes(note_img: Image.Image | None, threshold: int = 20) -> bool:
|
|
"""Return whether an RGBA note layer contains enough visible pixels."""
|
|
if note_img is None or note_img.mode != "RGBA":
|
|
return False
|
|
alpha = np.array(note_img)[:, :, 3]
|
|
return bool(np.sum(alpha > 50) > threshold)
|
|
|
|
|
|
def concatenate(images: list[Image.Image]) -> Image.Image | None:
|
|
if not images:
|
|
return None
|
|
result = Image.new(
|
|
"RGB",
|
|
(max(image.width for image in images), sum(image.height for image in images)),
|
|
"white",
|
|
)
|
|
current_y = 0
|
|
for image in images:
|
|
result.paste(image, (0, current_y))
|
|
current_y += image.height
|
|
return result
|
|
|
|
|
|
def apply_actions_and_regenerate(
|
|
workspace: EvaluationWorkspace,
|
|
data: AnnotationData,
|
|
student_id: str,
|
|
actions: list[dict[str, Any]],
|
|
notes_layer: Image.Image | None,
|
|
all_labels: list[str],
|
|
*,
|
|
update_score: bool = False,
|
|
) -> ExitCode:
|
|
"""Apply annotations and atomically merge the regenerated student files."""
|
|
output_dir = workspace.annotation_dir("checks") / f"Copie{student_id}"
|
|
bnote_path = output_dir / "bnote.json"
|
|
if not bnote_path.is_file():
|
|
print(f" Missing {bnote_path}")
|
|
return ExitCode.PARTIAL
|
|
bnote_data = read_json(bnote_path)
|
|
if not isinstance(bnote_data, dict):
|
|
raise TypeError(f"Expected a JSON object in {bnote_path}")
|
|
|
|
labels_data = data[student_id]
|
|
dirty_labels = apply_checkbox_actions(labels_data, actions, print)
|
|
if update_score:
|
|
dirty_labels |= apply_score_overrides(labels_data, output_dir / "score.json", print)
|
|
|
|
scores = dict.fromkeys(all_labels, "")
|
|
dirty_images: dict[str, Image.Image] = {}
|
|
concatenated: list[Image.Image] = []
|
|
filtered: list[Image.Image] = []
|
|
incomplete = False
|
|
|
|
for image_info in bnote_data.get("images", []):
|
|
if not isinstance(image_info, dict):
|
|
incomplete = True
|
|
continue
|
|
label = str(image_info.get("label", ""))
|
|
if label not in labels_data:
|
|
incomplete = True
|
|
continue
|
|
content = labels_data[label]
|
|
result = content["result"]
|
|
scores[label] = str(result.get("score", 0))
|
|
|
|
sub_note = None
|
|
if notes_layer is not None:
|
|
hmin = int(image_info.get("hmin", 0))
|
|
hmax = int(image_info.get("hmax", 0))
|
|
sub_note = notes_layer.crop((0, hmin, notes_layer.width, hmax))
|
|
has_notes = has_significant_notes(sub_note)
|
|
|
|
pdf_path = Path(content["pdf_path"])
|
|
if not pdf_path.is_file():
|
|
print(f" Missing answer PDF: {pdf_path}")
|
|
incomplete = True
|
|
continue
|
|
base_image, _, _ = annotating.make_base_image(pdf_path)
|
|
final_image, new_header_height = annotating.compose_label_image(
|
|
base_image,
|
|
label,
|
|
result,
|
|
content["coordinates"][0],
|
|
with_error=False,
|
|
)
|
|
if final_image is None:
|
|
incomplete = True
|
|
continue
|
|
|
|
if has_notes and sub_note is not None:
|
|
old_header_height = int(image_info.get("header_height", 0))
|
|
width, height = sub_note.size
|
|
if old_header_height > 0:
|
|
header = sub_note.crop((0, 0, width, min(height, old_header_height)))
|
|
final_image.paste(header, (0, 0), mask=header)
|
|
if height > old_header_height:
|
|
body = sub_note.crop((0, old_header_height, width, height))
|
|
final_image.paste(body, (0, new_header_height), mask=body)
|
|
|
|
if label in dirty_labels or has_notes:
|
|
dirty_images[label] = final_image
|
|
concatenated.append(final_image)
|
|
if float(scores[label]) != 4.0 or result.get("feedback", []):
|
|
filtered.append(final_image)
|
|
|
|
concat_image = concatenate(concatenated)
|
|
filtered_image = concatenate(filtered)
|
|
with staged_files(output_dir) as staging:
|
|
for label, image in dirty_images.items():
|
|
image.save(staging / f"{label}.jpg")
|
|
atomic_write_json(staging / "score.json", scores)
|
|
if concat_image is not None:
|
|
concat_image.save(staging / "Concat.jpg")
|
|
if filtered_image is not None:
|
|
filtered_image.save(staging / "Concat_F.jpg")
|
|
|
|
print(f" Saved regenerated files in {output_dir}")
|
|
return ExitCode.PARTIAL if incomplete else ExitCode.SUCCESS
|
|
|
|
|
|
def run(workspace: EvaluationWorkspace, *, update_score: bool = False) -> ExitCode:
|
|
workspace.require_files("labels", "correction.json")
|
|
workspace.require_directories("Copies", "Par label", "Bnot")
|
|
all_labels = utils.read_all_labels(workspace.root)
|
|
loaded = load_annotation_data(workspace)
|
|
for warning in loaded.warnings:
|
|
print(f"Warning: {warning}")
|
|
if not loaded.data:
|
|
print("No annotation data found.")
|
|
return ExitCode.PARTIAL
|
|
|
|
status = ExitCode.PARTIAL if loaded.warnings else ExitCode.SUCCESS
|
|
for student_id in sorted(loaded.data, key=utils.natural_key):
|
|
output_dir = workspace.annotation_dir("checks") / f"Copie{student_id}"
|
|
if not output_dir.is_dir():
|
|
print(f"Warning: missing annotation directory {output_dir}")
|
|
status = ExitCode.PARTIAL
|
|
continue
|
|
print(f"Processing annotations for: {student_id}")
|
|
actions, notes = detect_checks_and_notes(output_dir)
|
|
if notes is None and not actions and not update_score:
|
|
print(" No readable annotation input found.")
|
|
status = ExitCode.PARTIAL
|
|
continue
|
|
result = apply_actions_and_regenerate(
|
|
workspace,
|
|
loaded.data,
|
|
student_id,
|
|
actions,
|
|
notes,
|
|
all_labels,
|
|
update_score=update_score,
|
|
)
|
|
if result != ExitCode.SUCCESS:
|
|
status = ExitCode.PARTIAL
|
|
return status
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = evaluation_parser("Read checked annotations and regenerate copies")
|
|
parser.add_argument(
|
|
"--update-score",
|
|
action="store_true",
|
|
help="Override generated scores with values from existing score.json files",
|
|
)
|
|
return parser
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
|
parser = build_parser()
|
|
|
|
def handle(args: argparse.Namespace) -> ExitCode:
|
|
return run(workspace_from_args(args), update_score=args.update_score)
|
|
|
|
return execute(parser, argv, handle)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|
|
|