363 lines
13 KiB
Python
363 lines
13 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 configuration, 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.answer_info import build_answer_info
|
|
from copienator.annotation_data import AnnotationData, load_annotation_data
|
|
from copienator.filesystem import staged_files
|
|
from copienator.return_answers import save_return_answer_options
|
|
|
|
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)
|
|
|
|
# Keep the full-size difference in uint8. Converting both tall group images
|
|
# to the platform ``int`` dtype used several gigabytes per scan worker.
|
|
difference = np.asarray(
|
|
ImageChops.difference(reference, user_image), dtype=np.uint8
|
|
)
|
|
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[y1 + 5 : y2 - 5, x1 + 5 : x2 - 5]
|
|
if region.size == 0:
|
|
continue
|
|
# Preserve the previous mean-across-RGB threshold, but allocate its
|
|
# temporary float array only for the small checkbox region.
|
|
density = np.count_nonzero(np.mean(region, axis=2) > 30) / (
|
|
region.shape[0] * region.shape[1]
|
|
)
|
|
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)
|
|
|
|
del difference
|
|
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")
|
|
del reference_blur, user_blur
|
|
alpha = np.asarray(diff_image, dtype=np.uint8).copy()
|
|
np.greater(alpha, 50, out=alpha)
|
|
alpha *= np.uint8(255)
|
|
np.minimum(alpha, np.asarray(keep_mask, dtype=np.uint8), out=alpha)
|
|
notes = user_image.convert("RGBA")
|
|
notes.putalpha(Image.fromarray(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]
|
|
apply_checkbox_actions(labels_data, actions, print)
|
|
score_path = output_dir / "score.json"
|
|
preserve_score_file = update_score and score_path.is_file()
|
|
if update_score:
|
|
apply_score_overrides(labels_data, score_path, print)
|
|
|
|
scores = dict.fromkeys(all_labels, "")
|
|
answer_labels: list[str] = []
|
|
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))
|
|
if result.get("error") == "empty-answer":
|
|
continue
|
|
|
|
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)
|
|
|
|
dirty_images[label] = final_image
|
|
answer_labels.append(label)
|
|
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, remove=("Concat.jpg", "Concat_F.jpg", "Concat_F.pdf",
|
|
"touched.json", "answer_labels.json")) as staging:
|
|
for label, image in dirty_images.items():
|
|
image.save(staging / f"{label}.jpg")
|
|
if not preserve_score_file:
|
|
atomic_write_json(staging / "score.json", scores)
|
|
atomic_write_json(staging / "info.json", build_answer_info(
|
|
scores, labels_data, answer_labels
|
|
))
|
|
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")
|
|
|
|
if preserve_score_file:
|
|
print(f" Preserved existing score.json in {output_dir}")
|
|
print(f" Saved regenerated files in {output_dir}")
|
|
return ExitCode.PARTIAL if incomplete else ExitCode.SUCCESS
|
|
|
|
|
|
def run(
|
|
workspace: EvaluationWorkspace,
|
|
*,
|
|
update_score: bool = False,
|
|
return_answers_context: bool | None = None,
|
|
return_answers_question: bool | None = None,
|
|
return_answers_solution: bool | None = None,
|
|
) -> ExitCode:
|
|
workspace.require_files("labels", "correction.json")
|
|
workspace.require_directories("Copies", "Par label", "Bnot")
|
|
if configuration.RETURN_ANSWERS_ENABLED:
|
|
save_return_answer_options(
|
|
workspace.root,
|
|
context=(
|
|
configuration.RETURN_ANSWERS_CONTEXT
|
|
if return_answers_context is None
|
|
else return_answers_context
|
|
),
|
|
question=(
|
|
configuration.RETURN_ANSWERS_QUESTION
|
|
if return_answers_question is None
|
|
else return_answers_question
|
|
),
|
|
solution=(
|
|
configuration.RETURN_ANSWERS_SOLUTION
|
|
if return_answers_solution is None
|
|
else return_answers_solution
|
|
),
|
|
)
|
|
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=(
|
|
"Regenerate images with current statement/solution PDFs while "
|
|
"preserving and applying existing score.json values"
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--return-answers-context",
|
|
action=argparse.BooleanOptionalAction,
|
|
default=configuration.RETURN_ANSWERS_CONTEXT,
|
|
help="Include applicable context pages in individual answer exports",
|
|
)
|
|
parser.add_argument(
|
|
"--return-answers-question",
|
|
action=argparse.BooleanOptionalAction,
|
|
default=configuration.RETURN_ANSWERS_QUESTION,
|
|
help="Include the current question PDF in individual answer exports",
|
|
)
|
|
parser.add_argument(
|
|
"--return-answers-solution",
|
|
action=argparse.BooleanOptionalAction,
|
|
default=configuration.RETURN_ANSWERS_SOLUTION,
|
|
help="Include the current solution PDF in individual answer exports",
|
|
)
|
|
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_answers_context=args.return_answers_context,
|
|
return_answers_question=args.return_answers_question,
|
|
return_answers_solution=args.return_answers_solution,
|
|
)
|
|
|
|
return execute(parser, argv, handle)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|