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

Standardisation 4

Browse Source
This commit is contained in:
sebastien
2026-08-20 14:45:53 +02:00
parent 8b087bb3e4
commit bcba5facc8
6 changed files with 838 additions and 732 deletions
Download Patch File Download Diff File Expand all files Collapse all files
reading_annotations.py
+234 -328
View File
@@ -1,380 +1,286 @@
import sys
import os
import json
import numpy as np
import shutil
from __future__ import annotations
import argparse
from collections.abc import Sequence
from pathlib import Path
from PIL import Image, ImageChops, ImageFilter
Image.MAX_IMAGE_PIXELS = None
from typing import Any
import numpy as np
from pdf2image import convert_from_path
import annotating # Reuse rendering logic
from PIL import Image, ImageChops, ImageDraw, ImageFilter
DPI = 100
import annotating
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
def detect_checks_and_notes(output_dir):
"""
Returns:
actions: List of dicts {type, label, ...} for checked boxes
notes_img: RGBA image of manual notes (checks masked out)
"""
Image.MAX_IMAGE_PIXELS = None
names = ["Concat_annotated.pdf"]
for name in names:
pdf_path = os.path.join(output_dir, name)
if os.path.exists(pdf_path):
break
# ref_path = os.path.join(output_dir, "Reference.png")
ref_path = os.path.join(output_dir, "Reference.jpg")
json_path = os.path.join(output_dir, "checkboxes.json")
if not (os.path.exists(pdf_path) and os.path.exists(ref_path)):
print(f"\tMissing annotated file in {output_dir}")
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
# Load Coordinates
with open(json_path, 'r') as f:
boxes = json.load(f)
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()
# Load Reference
ref_img = Image.open(ref_path).convert("RGB")
# Load User PDF (First page only, assuming it's one long strip)
# Warning: If the PDF is huge, pdf2image might split pages or OOM.
# Assuming user didn't change page dimensions/order.
try:
# user_pages = convert_from_path(pdf_path, dpi=DPI)
# La version suivante évite les size mismatch
# Mais donne plus de bruit
user_pages = convert_from_path(pdf_path, dpi=72)
except Exception as e:
print(f"Error reading PDF: {e}")
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
# Concatenate PDF pages back to one image if user saved as multiple pages
total_h = sum(p.height for p in user_pages)
user_img = Image.new("RGB", (user_pages[0].width, total_h))
y = 0
for p in user_pages:
user_img.paste(p, (0, y))
y += p.height
# Resize user_img to match ref_img if slight mismatch (DPI export diffs)
if user_img.size != ref_img.size:
print("Debug : size mismatch : ", user_img.size, ref_img.size)
user_img = user_img.resize(ref_img.size, Image.Resampling.LANCZOS)
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)
# --- Detection Phase ---
actions = []
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]] = []
# Convert to numpy for analysis
ref_arr = np.array(ref_img)
user_arr = np.array(user_img)
# Diff for analysis
# Simple absolute difference
diff = np.abs(ref_arr.astype(int) - user_arr.astype(int)).astype(np.uint8)
# Convert to grayscale for thresholding
diff_gray = np.mean(diff, axis=2)
# Threshold for "Checked"
CHECK_THRESHOLD = 30 # intensity diff
DENSITY_THRESHOLD = 0.05 # 5% of pixels darkened
# Mask to hide checkmarks from the "Notes" extraction
mask_img = Image.new("L", ref_img.size, 255) # White (255) = keep, Black (0) = hide
mask_draw = ImageDraw.Draw(mask_img)
for box in boxes:
# global_box: [x1, y1, x2, y2]
b = box['global_box']
x1, y1, x2, y2 = map(int, b)
# Ensure bounds
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(ref_img.width, x2), min(ref_img.height, y2)
# Analyze ROI
roi = diff_gray[y1+5:y2-5, x1+5:x2-5]
if roi.size == 0: continue
changed_pixels = np.sum(roi > CHECK_THRESHOLD)
density = changed_pixels / roi.size
if density > DENSITY_THRESHOLD:
# print("A checked box !", density, b)
actions.append(box)
# It's checked, so we mask this area out for manual notes
# Expand mask slightly to catch sloppy ticks
mask_draw.rectangle([x1-15, y1-15, x2+15, y2+15], fill=0)
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 box["type"] == "score" and box["value"] == 0.0:
# Mask the whole line
mask_draw.rectangle([0, y1-10, ref_img.width, y2+10], fill=0)
# --- Extraction Phase ---
# 150 + no blur is alright, with some lines at the end
# 100 + 2 px blur is too clean : tes annotations sont morcelées
# 50 + 2 px blur seems good
ref_blur = ref_img.filter(ImageFilter.GaussianBlur(2))
user_blur = user_img.filter(ImageFilter.GaussianBlur(2))
# 1. Get difference image
# diff_img = ImageChops.difference(ref_img, user_img).convert("L")
diff_img = ImageChops.difference(ref_blur, user_blur).convert("L")
diff_data = np.array(diff_img)
alpha = np.where(diff_data > 50, 255, 0).astype(np.uint8)
notes = user_img.convert("RGBA")
r, g, b, a = notes.split()
# Combine the diff-based alpha with the box-mask
mask_arr = np.array(mask_img)
final_alpha = np.minimum(alpha, mask_arr)
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))
# notes.show()
return actions, notes
from PIL import ImageDraw
from utils import natural_key
from annotating import MARGIN_LEFT, ANNOT_WIDTH
def has_significant_notes(note_img, threshold=20):
"""Checks if the note layer has visible content (non-transparent pixels)."""
# Assuming note_img is RGBA.
# We check alpha channel for non-zero values (or low transparency)
# Since we generated notes with variable alpha based on diff, checking alpha sum is good.
if note_img.mode != 'RGBA':
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]
# Count pixels with significant opacity
visible_pixels = np.sum(alpha > 50)
# visible_pixels_bis = np.sum(alpha > 200)
# if visible_pixels > 0:
# print(f"Debug : visible pixels is {visible_pixels}")
return visible_pixels > threshold
return bool(np.sum(alpha > 50) > threshold)
def apply_actions_and_regenerate(root_dir, data, student_id, actions, notes_layer,
all_labels, update_score=False):
"""
Modifies data based on actions, reads bnote.json, cuts notes,
regenerates all label images for consistency, saves dirty ones,
and generates Concat.jpg.
"""
output_dir = os.path.join(root_dir, "Bnot", f"Copie{student_id}")
bnote_path = os.path.join(output_dir, "bnote.json")
score_path = os.path.join(output_dir, "score.json")
if not os.path.exists(bnote_path):
print(f"Error: bnote.json not found in {output_dir}")
return
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
with open(bnote_path, 'r') as f:
bnote_data = json.load(f)
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)
# --- 1. Apply Actions to Data (Update scores / Flags for deletion) ---
actions_by_label = {}
for a in actions:
actions_by_label.setdefault(a['label'], []).append(a)
dirty_labels = set() # Labels that logic says changed
for label, acts in actions_by_label.items():
if label not in labels_data: continue
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']
feedbacks = result.get('feedback', [])
result = content["result"]
scores[label] = str(result.get("score", 0))
# Helpers to find objects by index (references match those in feedbacks list)
global_fb = [f for f in feedbacks if not f.get('box_2d')]
local_fb = [f for f in feedbacks if f.get('box_2d')]
local_fb.sort(key=lambda x: x['box_2d'][0])
for act in acts:
if act['type'] == 'score':
result['score'] = act['value']
dirty_labels.add(label)
print(f" > Updated score for {label} to {act['value']}")
elif act['type'] == 'clear_all':
for fb in feedbacks:
fb["to_delete"] = True
if fb.get("box_2d"):
fb["norectangle"] = True
dirty_labels.add(label)
print(f" > Cleared all feedbacks in {label}")
elif act['type'] == 'del_global':
if act['index'] < len(global_fb):
global_fb[act['index']]["to_delete"] = True
dirty_labels.add(label)
print(f" > Deleted global feedback in {label}")
elif act['type'] in ('del_local', 'del_local_rect'):
if act['index'] < len(local_fb):
target = local_fb[act['index']]
if act['type'] == 'del_local':
target["to_delete"] = True
print(f" > Deleted local feedback in {label}")
else:
target["norectangle"] = True
print(f" > Deleted rect in {label}")
dirty_labels.add(label)
# --- 1.5 Override with existing score.json if requested ---
if update_score and os.path.exists(score_path):
try:
with open(score_path, "r") as f:
existing_scores = json.load(f)
for label, existing_score in existing_scores.items():
if label in labels_data:
current_score = str(labels_data[label]['result'].get('score', 0))
# If manually modified, override the result and mark dirty
if current_score != str(existing_score):
labels_data[label]['result']['score'] = existing_score
dirty_labels.add(label)
print(f" > Overrode score for {label} to {existing_score} from existing score.json")
except json.JSONDecodeError:
print(f" > Warning: Could not read existing {score_path}")
# --- 2. Process Images (Cut notes, Regenerate, Concatenate) ---
concat_list = []
concat_list_F = []
d_notes = dict.fromkeys(all_labels, "")
# Iterate over images defined in bnote.json to maintain order/geometry
for img_info in bnote_data.get("images", []):
label = img_info["label"]
if label not in labels_data: continue
# Update scores dict
content = labels_data[label]
result = content['result']
d_notes[label] = str(result.get('score', 0))
# A. Cut Manual Notes
hmin, hmax = img_info["hmin"], img_info["hmax"]
sub_note = None
if notes_layer:
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)
# B. Regenerate Label Image
# We always regenerate to ensure Concat.jpg is consistent with any modifications
# pdf_path = Path(root_dir) / "Copies" / f"Copie{student_id}" / f"{label}.pdf"
pdf_path = content.get('pdf_path') # Contient le suffixe _new si nécessaire
if not os.path.exists(pdf_path): continue
(base_img, _, _) = annotating.make_base_image(pdf_path)
# Compose uses the result object we modified in step 1
final_img, new_header_h = annotating.compose_label_image(
base_img, label, content['result'], content['coordinates'][0],
with_error=False
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_img==None:
if final_image is None:
incomplete = True
continue
# Overlay manual notes
if has_notes:
old_header_h = int(img_info.get("header_height", 0))
w, h = sub_note.size
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)
# 1. Paste header ink at the top
if old_header_h > 0:
header_crop = sub_note.crop((0, 0, w, min(h, old_header_h)))
final_img.paste(header_crop, (0, 0), mask=header_crop)
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)
# 2. Paste student-content ink at the new header height
if h > old_header_h:
body_crop = sub_note.crop((0, old_header_h, w, h))
final_img.paste(body_crop, (0, new_header_h), mask=body_crop)
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")
# C. Save individual file if Modified (Dirty logic or visual notes)
if (label in dirty_labels) or has_notes:
save_path = os.path.join(output_dir, f"{label}.jpg")
final_img.save(save_path)
print(f" Saved dirty image: {label}.jpg")
print(f" Saved regenerated files in {output_dir}")
return ExitCode.PARTIAL if incomplete else ExitCode.SUCCESS
concat_list.append(final_img)
perfect_no_comment = True
if float(d_notes[label]) != 4.0:
perfect_no_comment = False
if len(result.get('feedback', [])) != 0:
perfect_no_comment = False
if not perfect_no_comment:
concat_list_F.append(final_img)
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
# --- 3. Save Final Outputs ---
with open(score_path, "w") as f:
json.dump(d_notes, f, indent=4)
print(f" Saved {score_path}")
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
if concat_list:
max_w = max(i.width for i in concat_list)
total_h = sum(i.height for i in concat_list)
full_img = Image.new("RGB", (max_w, total_h), "white")
y = 0
for img in concat_list:
full_img.paste(img, (0, y))
y += img.height
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
full_img.save(os.path.join(output_dir, "Concat.jpg"))
print(f" Saved regenerated Concat.jpg")
if concat_list_F:
max_w = max(i.width for i in concat_list_F)
total_h = sum(i.height for i in concat_list_F)
full_img = Image.new("RGB", (max_w, total_h), "white")
y = 0
for img in concat_list_F:
full_img.paste(img, (0, y))
y += img.height
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)
full_img.save(os.path.join(output_dir, "Concat_F.jpg"))
print(f" Saved regenerated Concat_F.jpg")
from utils import read_all_labels
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description="Read annotations and compile PDFs")
parser.add_argument("input_path", help="Directory path")
parser.add_argument("--update-score", action="store_true", help="Override scores with values from existing score.json")
args = parser.parse_args()
root_dir = args.input_path
try:
all_labels = read_all_labels(Path(root_dir))
except FileNotFoundError:
all_labels = []
# Load original data
original_data = annotating.make_dictionary(root_dir)
# Process each Bnot folder
for student_id in original_data.keys():
bnot_dir = os.path.join(root_dir, "Bnot", f"Copie{student_id}")
if os.path.exists(bnot_dir):
print(f"Processing annotations for: {student_id}")
actions, notes = detect_checks_and_notes(bnot_dir)
if actions or notes or args.update_score:
apply_actions_and_regenerate(root_dir, original_data, student_id,
actions, notes, all_labels,
update_score=args.update_score)
else:
print(" No changes detected or missing files.")
raise SystemExit(main())