440 lines
15 KiB
Python
440 lines
15 KiB
Python
from __future__ import annotations
|
||
|
||
import argparse
|
||
import concurrent.futures
|
||
import re
|
||
from collections.abc import Sequence
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import matplotlib
|
||
|
||
matplotlib.use("Agg")
|
||
|
||
from PIL import Image, ImageFont
|
||
from reportlab.pdfgen import canvas
|
||
|
||
from copienator import (
|
||
CliError,
|
||
EvaluationWorkspace,
|
||
ExitCode,
|
||
atomic_write_json,
|
||
execute,
|
||
read_json,
|
||
target_parser,
|
||
utils,
|
||
workspace_from_target,
|
||
)
|
||
from copienator.annotation_data import load_annotation_data
|
||
from copienator.annotation_documents import annotation_limits, copy_documents, document_fits
|
||
from copienator.commands import annotating
|
||
from copienator.commands.import_annotations import clear_import_directory
|
||
from copienator.filesystem import staged_directory
|
||
from copienator.utils import natural_key
|
||
|
||
BOX_SIZE = 30
|
||
SCORE_BOX_SIZE = 40
|
||
SCORES = [value * 0.5 for value in range(10)]
|
||
EXPECTED_OUTPUTS = ("bnote.json", "checkboxes.json", "Reference.jpg", "Concat.pdf")
|
||
|
||
try:
|
||
CHECKBOX_FONT = ImageFont.truetype("DejaVuSans.ttf", 20)
|
||
except OSError:
|
||
try:
|
||
CHECKBOX_FONT = ImageFont.truetype("arial.ttf", 20)
|
||
except OSError:
|
||
CHECKBOX_FONT = ImageFont.load_default()
|
||
|
||
|
||
def draw_checkbox(draw, x, y, size=BOX_SIZE, label=None, fill="white"):
|
||
if label:
|
||
draw.text(
|
||
(x - BOX_SIZE - 5, y + 2), str(label), fill="black", font=CHECKBOX_FONT
|
||
)
|
||
draw.rectangle([x, y, x + size, y + size], fill=fill, outline="black", width=2)
|
||
return [x, y, x + size, y + size]
|
||
|
||
|
||
class CheckboxRenderer:
|
||
def __init__(self, label_name):
|
||
self.label = label_name
|
||
self.checkboxes = []
|
||
|
||
def callback(self, kind, draw, pos, meta):
|
||
if kind == "header_item":
|
||
if meta.get("type") == "score":
|
||
start_x = pos["w"] + 20
|
||
for value in SCORES:
|
||
box = draw_checkbox(
|
||
draw,
|
||
start_x,
|
||
pos["y"] + 25,
|
||
SCORE_BOX_SIZE,
|
||
str(value),
|
||
)
|
||
self.checkboxes.append(
|
||
{
|
||
"type": "score",
|
||
"label": self.label,
|
||
"value": value,
|
||
"rel_box": box,
|
||
}
|
||
)
|
||
start_x += SCORE_BOX_SIZE + 45
|
||
start_x += SCORE_BOX_SIZE + 60
|
||
box = draw_checkbox(
|
||
draw,
|
||
start_x,
|
||
pos["y"] + 25,
|
||
SCORE_BOX_SIZE,
|
||
"clr",
|
||
)
|
||
self.checkboxes.append(
|
||
{"type": "clear_all", "label": self.label, "rel_box": box}
|
||
)
|
||
elif meta.get("type") == "global_fb":
|
||
box = draw_checkbox(
|
||
draw,
|
||
pos["w"] - BOX_SIZE - 5,
|
||
pos["y"] + 5,
|
||
BOX_SIZE,
|
||
)
|
||
self.checkboxes.append(
|
||
{
|
||
"type": "del_global",
|
||
"label": self.label,
|
||
"index": meta["index"],
|
||
"rel_box": box,
|
||
"text_preview": meta["data"]["text"][:20],
|
||
}
|
||
)
|
||
elif kind == "local_rect":
|
||
rectangle = pos["box"]
|
||
box = draw_checkbox(
|
||
draw,
|
||
rectangle[2] - BOX_SIZE,
|
||
rectangle[1],
|
||
BOX_SIZE,
|
||
)
|
||
self.checkboxes.append(
|
||
{
|
||
"type": "del_local_rect",
|
||
"label": self.label,
|
||
"index": meta["index"],
|
||
"final_box": box,
|
||
"text_preview": meta["data"]["text"][:20],
|
||
}
|
||
)
|
||
elif kind == "local_text":
|
||
box = draw_checkbox(
|
||
draw,
|
||
pos["x"] + pos["w"] - BOX_SIZE,
|
||
pos["y"],
|
||
BOX_SIZE,
|
||
)
|
||
self.checkboxes.append(
|
||
{
|
||
"type": "del_local",
|
||
"label": self.label,
|
||
"index": meta["index"],
|
||
"final_box": box,
|
||
"text_preview": meta["data"]["text"][:20],
|
||
}
|
||
)
|
||
|
||
|
||
def _output_complete(output_dir: Path) -> bool:
|
||
return all(
|
||
(document / name).is_file()
|
||
for document in copy_documents(output_dir)
|
||
for name in EXPECTED_OUTPUTS
|
||
)
|
||
|
||
|
||
def _render_student(
|
||
workspace: EvaluationWorkspace,
|
||
student_id: str,
|
||
labels: dict[str, dict[str, Any]],
|
||
*,
|
||
overwrite: bool,
|
||
output_mode: str,
|
||
output_root: Path | None = None,
|
||
) -> str:
|
||
max_height_px, max_pixels = annotation_limits()
|
||
output_dir = (
|
||
output_root
|
||
if output_root is not None
|
||
else workspace.annotation_dir(output_mode)
|
||
) / f"Copie{student_id}"
|
||
if _output_complete(output_dir) and not overwrite:
|
||
print(f"Skipping {student_id}: output is complete.")
|
||
return "skipped"
|
||
|
||
print(f"Generating checkable PDF for: {student_id}")
|
||
label_images: list[Image.Image] = []
|
||
checkbox_groups: list[list[dict[str, Any]]] = []
|
||
bnote_entries: list[dict[str, Any]] = []
|
||
problems = False
|
||
|
||
for label, content in sorted(labels.items(), key=lambda item: natural_key(item[0])):
|
||
pdf_path = Path(content["pdf_path"])
|
||
if not pdf_path.exists():
|
||
print(f"Warning: answer PDF not found: {pdf_path}")
|
||
problems = True
|
||
continue
|
||
base_image, _, _ = annotating.make_base_image(pdf_path)
|
||
checkbox_renderer = CheckboxRenderer(label)
|
||
final_image, header_height = annotating.compose_label_image(
|
||
base_image,
|
||
label,
|
||
content["result"],
|
||
content["coordinates"][0],
|
||
draw_callback=checkbox_renderer.callback,
|
||
source_pdf=pdf_path,
|
||
)
|
||
if final_image is None:
|
||
problems = True
|
||
continue
|
||
label_images.append(final_image)
|
||
checkbox_groups.append(checkbox_renderer.checkboxes)
|
||
bnote_entries.append(
|
||
{
|
||
"id": student_id,
|
||
"label": label,
|
||
"header_height": header_height,
|
||
"layout": final_image.info.get('annotation_layout', annotating.layout_for(None)),
|
||
}
|
||
)
|
||
|
||
if not label_images:
|
||
print(f"Warning: no annotations could be rendered for Copie{student_id}")
|
||
return "partial"
|
||
|
||
batches = []
|
||
start = 0
|
||
height = 0
|
||
width = 0
|
||
for index, image in enumerate(label_images):
|
||
if not document_fits(image.width, image.height, max_height_px, max_pixels):
|
||
raise CliError(
|
||
f"Copie{student_id} ({bnote_entries[index]['label']}) is "
|
||
f"{image.width} × {image.height} px ({image.width * image.height} pixels); "
|
||
f"a single answer cannot be split. Document limits are "
|
||
f"{max_height_px} px high and {max_pixels} pixels."
|
||
)
|
||
if not document_fits(max(width, image.width), height + image.height,
|
||
max_height_px, max_pixels):
|
||
batches.append((start, index))
|
||
start = index
|
||
height = 0
|
||
width = 0
|
||
height += image.height
|
||
width = max(width, image.width)
|
||
batches.append((start, len(label_images)))
|
||
|
||
with staged_directory(output_dir) as staging:
|
||
parts = []
|
||
for index, (start, end) in enumerate(batches, start=1):
|
||
destination = staging
|
||
if len(batches) > 1:
|
||
name = f"Part{index}"
|
||
parts.append(name)
|
||
destination /= name
|
||
_save_document(destination, label_images[start:end],
|
||
checkbox_groups[start:end], bnote_entries[start:end])
|
||
if parts:
|
||
atomic_write_json(staging / "bnote.json", {
|
||
"parts": parts, "images": bnote_entries,
|
||
})
|
||
return "partial" if problems else "success"
|
||
|
||
|
||
def _save_document(
|
||
output_dir: Path,
|
||
label_images: list[Image.Image],
|
||
checkbox_groups: list[list[dict[str, Any]]],
|
||
bnote_entries: list[dict[str, Any]],
|
||
) -> None:
|
||
max_height_px, max_pixels = annotation_limits()
|
||
max_width = max(image.width for image in label_images)
|
||
total_height = sum(image.height for image in label_images)
|
||
if not document_fits(max_width, total_height, max_height_px, max_pixels):
|
||
raise CliError(
|
||
f"Document would be {max_width} × {total_height} px "
|
||
f"({max_width * total_height} pixels); document limits are "
|
||
f"{max_height_px} px high and {max_pixels} pixels."
|
||
)
|
||
concatenated = Image.new("RGB", (max_width, total_height), "white")
|
||
checkbox_map: list[dict[str, Any]] = []
|
||
current_y = 0
|
||
for index, (image, checkboxes) in enumerate(
|
||
zip(label_images, checkbox_groups, strict=True)
|
||
):
|
||
concatenated.paste(image, (0, current_y))
|
||
bnote_entries[index]["hmin"] = current_y
|
||
bnote_entries[index]["hmax"] = current_y + image.height
|
||
for item in checkboxes:
|
||
box = item.get("final_box") or item.get("rel_box")
|
||
item["global_box"] = [
|
||
box[0],
|
||
box[1] + current_y,
|
||
box[2],
|
||
box[3] + current_y,
|
||
]
|
||
checkbox_map.append(item)
|
||
current_y += image.height
|
||
|
||
output_dir.mkdir(parents=True, exist_ok=True)
|
||
atomic_write_json(
|
||
output_dir / "bnote.json",
|
||
{"width": max_width, "height": total_height, "images": bnote_entries},
|
||
)
|
||
atomic_write_json(output_dir / "checkboxes.json", checkbox_map)
|
||
reference = output_dir / "Reference.jpg"
|
||
concatenated.save(reference, quality=90)
|
||
pdf_path = output_dir / "Concat.pdf"
|
||
pdf_canvas = canvas.Canvas(str(pdf_path), pagesize=(max_width, total_height))
|
||
pdf_canvas.drawImage(
|
||
str(reference),
|
||
0,
|
||
0,
|
||
width=max_width,
|
||
height=total_height,
|
||
)
|
||
pdf_canvas.save()
|
||
|
||
|
||
def _copy_id_from_target(workspace: EvaluationWorkspace, target: Path) -> str | None:
|
||
if target == workspace.root:
|
||
return None
|
||
match = re.search(r"Copie(\d+)", target.name)
|
||
if match is None:
|
||
raise CliError(f"Could not extract a copy id from target: {target}")
|
||
return match.group(1)
|
||
|
||
|
||
def _load_refaire(workspace: EvaluationWorkspace):
|
||
workspace.require_files("refaire.json")
|
||
loaded = read_json(workspace.refaire_file)
|
||
if not isinstance(loaded, list):
|
||
raise CliError("refaire.json must contain a JSON array")
|
||
return loaded
|
||
|
||
|
||
def run(
|
||
workspace: EvaluationWorkspace,
|
||
target: Path,
|
||
*,
|
||
overwrite: bool = False,
|
||
refaire: bool = False,
|
||
clear_import: bool = False,
|
||
) -> ExitCode:
|
||
workspace.require_files("labels", "correction.json")
|
||
workspace.require_directories("Copies", "Par label")
|
||
utils.read_all_labels(workspace.root)
|
||
copy_id = _copy_id_from_target(workspace, target)
|
||
refaire_list = _load_refaire(workspace) if refaire else None
|
||
loaded = load_annotation_data(
|
||
workspace,
|
||
refaire_list=refaire_list,
|
||
copy_id=None if refaire else copy_id,
|
||
)
|
||
for warning in loaded.warnings:
|
||
print(f"Warning: {warning}")
|
||
if not loaded.data:
|
||
print("Warning: no annotation data was found.")
|
||
return ExitCode.PARTIAL
|
||
|
||
output_mode = "refaire" if refaire else "checks"
|
||
tasks = sorted(loaded.data.items(), key=lambda item: natural_key(item[0]))
|
||
if refaire:
|
||
output_root = workspace.annotation_dir("refaire")
|
||
if output_root.exists() and not overwrite:
|
||
raise CliError(
|
||
"BRnot already exists; use --overwrite to replace the previous redo."
|
||
)
|
||
|
||
class IncompleteRedo(Exception):
|
||
pass
|
||
|
||
try:
|
||
with staged_directory(output_root) as staging:
|
||
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
|
||
futures = [
|
||
executor.submit(
|
||
_render_student,
|
||
workspace,
|
||
student_id,
|
||
labels,
|
||
overwrite=True,
|
||
output_mode="refaire",
|
||
output_root=staging,
|
||
)
|
||
for student_id, labels in tasks
|
||
]
|
||
statuses = [future.result() for future in futures]
|
||
if loaded.warnings or any(status != "success" for status in statuses):
|
||
raise IncompleteRedo
|
||
except IncompleteRedo:
|
||
print("Warning: incomplete redo generation; previous BRnot was preserved.")
|
||
return ExitCode.PARTIAL
|
||
if clear_import:
|
||
clear_import_directory(workspace)
|
||
return ExitCode.SUCCESS
|
||
|
||
statuses: list[str] = []
|
||
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
|
||
futures = [
|
||
executor.submit(
|
||
_render_student,
|
||
workspace,
|
||
student_id,
|
||
labels,
|
||
overwrite=overwrite,
|
||
output_mode=output_mode,
|
||
)
|
||
for student_id, labels in tasks
|
||
]
|
||
for future in futures:
|
||
statuses.append(future.result())
|
||
if loaded.warnings or "partial" in statuses:
|
||
return ExitCode.PARTIAL
|
||
if clear_import:
|
||
clear_import_directory(workspace)
|
||
return ExitCode.SUCCESS
|
||
|
||
|
||
def build_parser() -> argparse.ArgumentParser:
|
||
parser = target_parser("Generate annotated PDFs with checkboxes.")
|
||
parser.add_argument("--clear-import", action="store_true",
|
||
help="Empty IMPORT_DIR after successful annotation generation")
|
||
parser.add_argument(
|
||
"--overwrite", action="store_true", help="Replace existing outputs"
|
||
)
|
||
parser.add_argument(
|
||
"--refaire",
|
||
action="store_true",
|
||
help="Process only entries from refaire.json",
|
||
)
|
||
return parser
|
||
|
||
|
||
def main(argv: Sequence[str] | None = None) -> int:
|
||
parser = build_parser()
|
||
|
||
def handler(args: argparse.Namespace) -> ExitCode:
|
||
workspace, target = workspace_from_target(args)
|
||
return run(
|
||
workspace,
|
||
target,
|
||
overwrite=args.overwrite,
|
||
refaire=args.refaire,
|
||
clear_import=args.clear_import,
|
||
)
|
||
|
||
return execute(parser, argv, handler)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|