374 lines
12 KiB
Python
374 lines
12 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.commands import annotating
|
|
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((output_dir / name).is_file() 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:
|
|
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,
|
|
)
|
|
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,
|
|
"img_h": final_image.height,
|
|
}
|
|
)
|
|
|
|
if not label_images:
|
|
print(f"Warning: no annotations could be rendered for Copie{student_id}")
|
|
return "partial"
|
|
|
|
max_width = max(image.width for image in label_images)
|
|
total_height = sum(image.height for image in label_images)
|
|
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
|
|
del bnote_entries[index]["img_h"]
|
|
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
|
|
|
|
with staged_directory(output_dir) as staging:
|
|
atomic_write_json(
|
|
staging / "bnote.json",
|
|
{"width": max_width, "height": total_height, "images": bnote_entries},
|
|
)
|
|
atomic_write_json(staging / "checkboxes.json", checkbox_map)
|
|
reference = staging / "Reference.jpg"
|
|
concatenated.save(reference, quality=90)
|
|
pdf_path = staging / "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()
|
|
return "partial" if problems else "success"
|
|
|
|
|
|
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,
|
|
) -> 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
|
|
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
|
|
return ExitCode.SUCCESS
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = target_parser("Generate annotated PDFs with checkboxes.")
|
|
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,
|
|
)
|
|
|
|
return execute(parser, argv, handler)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|