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

Standardisation 3

Browse Source
This commit is contained in:
sebastien
2026-08-20 14:35:04 +02:00
parent 63f690b353
commit 8b087bb3e4
9 changed files with 1187 additions and 683 deletions
Download Patch File Download Diff File Expand all files Collapse all files
annotating.py
+137 -239
View File
@@ -1,140 +1,52 @@
import sys
from __future__ import annotations
import argparse
import io
import os
import json
import glob
from pathlib import Path
import re
import subprocess
from PIL import Image
import tempfile
from collections.abc import Sequence
from pathlib import Path
import matplotlib
import matplotlib.colors as mcolors
import PIL.ImageOps
matplotlib.use("Agg")
from matplotlib import pyplot as plt
from pdf2image import convert_from_path
from PIL import Image, ImageDraw, ImageFont
import utils
from config import LATEX_ANOT_AFTER, LATEX_ANOT_BEFORE
from copienator import (
EvaluationWorkspace,
ExitCode,
atomic_write_json,
evaluation_parser,
execute,
workspace_from_args,
)
from copienator.annotation_data import load_annotation_data
from copienator.filesystem import staged_directory
from utils import natural_key
MARGIN_LEFT = 300
ANNOT_WIDTH = 600
# Results is : Copie id -> label -> {pdf_path, gemini_result, coordinates}
# Coordinates are the real coordinates (hmin, hmax) of the image in the Group
# The gemini_result coordinates should be un-normalized !
def make_dictionary(root_dir, refaire=False, refaire_list=[]):
correction_path = os.path.join(root_dir, "correction.json")
# Load correction data
try:
with open(correction_path, 'r', encoding='utf-8') as f:
corrections = json.load(f)
except FileNotFoundError:
print(f"Error: {correction_path} not found.")
sys.exit(1)
# Dictionary: keys are IDs
result_data = {}
def make_dictionary(root_dir, refaire=False, refaire_list=None):
"""Compatibility wrapper used by the annotation-reading scripts."""
workspace = EvaluationWorkspace(Path(root_dir))
loaded = load_annotation_data(
workspace,
refaire_list=(refaire_list or []) if refaire else None,
)
return loaded.data
# Iterate through labels and items in correction.json
for label, items in corrections.items():
items = sum(items, []) # Flatten
for item in items:
# print(item)
student_id = item['id']
result_obj = item['result']
if result_obj.get("suffix") == "_old":
continue
# Find coordinates
coordinates = None
height,width= None, None
label_dir = Path(root_dir) / "Par label" / label
# Search all json files in Dir/label
json_files = glob.glob(os.path.join(label_dir, "*.json"))
for jf in json_files:
try:
with open(jf, 'r', encoding='utf-8') as f:
coord_list = json.load(f)
# Format: [["id", x, y, width_r, "label"], ...]
for entry in coord_list:
if entry[0] == student_id:
coordinates = (entry[1], entry[2])
img_path = os.path.splitext(jf)[0] + ".jpg"
with Image.open(img_path) as img:
width, height = img.size
break
except json.JSONDecodeError:
continue
if coordinates:
break
suffix = result_obj.get("suffix", "")
if suffix == "_new":
pdf_path = Path(root_dir) / "Copies" / f"Copie{student_id}" / f"{label}_new.pdf"
else:
pdf_path = Path(root_dir) / "Copies" / f"Copie{student_id}" / f"{label}.pdf"
# Initialize dictionary structure for this ID if missing
if student_id not in result_data:
result_data[student_id] = {}
fb = result_obj.get("feedback", [])
for i in range(len(fb)):
el = fb[i]
if height == None or width == None:
print("?? height or width is None, for ", student_id, label)
if "box_2d" in el and el["box_2d"]:
el["box_2d"][0] = (el["box_2d"][0] * height)//1000
el["box_2d"][2] = (el["box_2d"][2] * height)//1000
el["box_2d"][1] = (el["box_2d"][1] * width)//1000
el["box_2d"][3] = (el["box_2d"][3] * width)//1000
# Populate the object
result_data[student_id][label] = {
"pdf_path": pdf_path,
"result": result_obj,
"coordinates": coordinates
}
if refaire:
for copie_name, labels_to_redo in refaire_list:
sid = copie_name.replace("Copie", "") # Extract "01" from "Copie01"
if sid in result_data:
# Si des labels à refaire ne sont pas présent dans la correction
# On ajoute des dummies
if labels_to_redo: # Si la liste est non vide
for lbl in labels_to_redo:
pdf_path = Path(root_dir) / "Copies" / f"Copie{sid}" / f"{lbl}.pdf"
if not Path(pdf_path).exists():
pdf_path_new = Path(root_dir) / "Copies" / f"Copie{sid}" / f"{lbl}_new.pdf"
if pdf_path_new.exists():
pdf_path = pdf_path_new
else:
print("Debug : asked to refaire", sid, lbl, "but pdf absent")
continue
# result_data[sid][lbl] = {
# "pdf_path": pdf_path,
# "result": {
# "score": 0.0,
# "feedback": [],
# "error": "non traité"
# },
# "coordinates": (0,0)
# }
else: # Ce student id n'a jamais été corrigé
result_data[sid] = {}
for lbl in labels_to_redo:
pdf_path = Path(root_dir) / "Copies" / f"Copie{sid}" / f"{lbl}.pdf"
if not pdf_path.exists():
pdf_path_new = Path(root_dir) / "Copies" / f"Copie{sid}" / f"{lbl}_new.pdf"
if pdf_path_new.exists():
pdf_path = pdf_path_new
else:
print("Debug : asked to refaire", sid, lbl, "but pdf absent")
continue
result_data[sid][lbl] = {
"pdf_path": pdf_path,
"result": {
"score": 0.0,
"feedback": [],
"error": "non traité"
},
"coordinates": (0,0)
}
return result_data
def make_base_image(pdf_path):
pages = convert_from_path(pdf_path)
@@ -152,20 +64,6 @@ def make_base_image(pdf_path):
current_y += page.height
return (base_img, total_h, max_w)
import io
import shutil
from pdf2image import convert_from_path
from PIL import Image, ImageDraw, ImageFont
import matplotlib
matplotlib.use('Agg') # Force headless rendering
import matplotlib.pyplot as plt
# plt.rcParams.update({ "text.usetex": True,
# "text.latex.preamble": r"\usepackage{bbold}"})
import re
import textwrap
def normalize_mathtext(text):
"""
Replaces LaTeX shortcuts not supported by Matplotlib's mathtext parser.
@@ -274,16 +172,6 @@ def render_latex_text(text, width_px, bg_color=(255, 255, 255, 255), max_lines=N
final_img.alpha_composite(img)
return final_img
import os
import tempfile
import subprocess
import PIL.ImageOps
from config import LATEX_ANOT_AFTER, LATEX_ANOT_BEFORE
from string import Template
def render_real_latex_text(text, width_px, bg_color=(255, 255, 255, 255), max_lines=None, fontsize=19):
dpi = 100
width_in = width_px / dpi
@@ -303,11 +191,12 @@ def render_real_latex_text(text, width_px, bg_color=(255, 255, 255, 255), max_li
f.write(latex_template)
# Compile to PDF
result = subprocess.run(
subprocess.run(
['pdflatex', '-interaction=nonstopmode', 'text.tex'],
cwd=temp_dir,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL
stderr=subprocess.DEVNULL,
check=False,
)
if not os.path.exists(pdf_path):
@@ -336,12 +225,6 @@ def render_real_latex_text(text, width_px, bg_color=(255, 255, 255, 255), max_li
return final_img
import io
from PIL import Image
import matplotlib.pyplot as plt
import matplotlib.colors as mcolors
from highlight_text import ax_text
def color(score):
t = max(0.0, min(1.0, float(score) / 4.0))
t = t*1.5 - 0.25
@@ -350,8 +233,6 @@ def color(score):
green = 150 * t
return mcolors.to_hex((red/255, green/255, 0))
from PIL import Image, ImageDraw, ImageFont
def render_score_text(label, score, error, width_px, fontsize=30,
bg_color=(255, 255, 255, 255),
with_error=True, id=None):
@@ -380,7 +261,7 @@ def render_score_text(label, score, error, width_px, fontsize=30,
try:
font_regular = ImageFont.truetype("DejaVuSans.ttf", fontsize)
font_bold = ImageFont.truetype("DejaVuSans-Bold.ttf", fontsize)
except IOError:
except OSError:
# Fallback for systems without specific TTFs readily available
print("here")
try:
@@ -429,7 +310,6 @@ def compose_label_image(base_img, label, result, hmin,
if base_img.width < TARGET_MIN_WIDTH:
total_missing = TARGET_MIN_WIDTH - base_img.width
left_pad = min(total_missing, MARGIN_LEFT)
right_pad = total_missing - left_pad
new_base = Image.new("RGB", (TARGET_MIN_WIDTH, base_img.height), "white")
new_base.paste(base_img, (left_pad, 0))
@@ -549,100 +429,118 @@ def compose_label_image(base_img, label, result, hmin,
return final_img, header_height
from utils import natural_key
import concurrent.futures
def process_student(student_id, labels_data, root_dir, all_labels, overwrite):
"""Helper function to process a single student."""
# Prepare output directory: Dir/Anot_CopieID
output_dir = os.path.join(root_dir, "Anot", f"Copie{student_id}")
output_dir = Path(root_dir) / "Anot" / f"Copie{student_id}"
# Check if already processed (Concat.jpg exists)
concat_path = os.path.join(output_dir, "Concat.jpg")
if os.path.exists(concat_path) and not overwrite:
concat_path = output_dir / "Concat.jpg"
if concat_path.exists() and not overwrite:
print(f"Skipping Copie {student_id} (Concat.jpg exists)")
return
return "skipped"
print("Processing :", student_id)
problems = False
with staged_directory(output_dir) as staging:
d_notes = dict.fromkeys(all_labels, "")
label_images = []
sorted_labels = sorted(labels_data.items(), key=lambda item: natural_key(item[0]))
# Clean folder if re-processing
if os.path.exists(output_dir):
shutil.rmtree(output_dir)
os.makedirs(output_dir)
for label, content in sorted_labels:
pdf_full_path = content.get('pdf_path')
if not pdf_full_path or not Path(pdf_full_path).exists():
print(f"File not found: {pdf_full_path}")
problems = True
continue
try:
base_img, _, _ = make_base_image(pdf_full_path)
except Exception as exc: # noqa: BLE001 - PDF/LaTeX backends vary
print(f"Error converting {pdf_full_path}: {exc}")
problems = True
continue
d_notes = dict.fromkeys(all_labels, "")
label_images = []
result = content.get('result', {})
coordinates = content.get('coordinates', (0, 0))
d_notes[label] = str(result.get('score', 0))
final_img, _ = compose_label_image(
base_img,
label,
result,
coordinates[0],
with_empty=True,
render_fn=render_real_latex_text,
)
final_img.save(staging / f"{label}.jpg")
if result.get('error', "") != "empty-answer":
label_images.append(final_img)
# !! Trier par l'ordre des labels plutôt
sorted_labels = sorted(list(labels_data.items()), key=natural_key)
for label, content in sorted_labels:
# 1. Find PDF path
copie_folder = f"Copie{student_id}"
pdf_full_path = content.get('pdf_path')
if not pdf_full_path or not os.path.exists(pdf_full_path):
print(f"File not found: {pdf_full_path}")
continue
# 2. Convert PDF to Image
try:
(base_img, _, _) = make_base_image(pdf_full_path)
except Exception as e:
print(f"Error converting {pdf_full_path}: {e}")
continue
result = content.get('result', {})
coordinates = content.get('coordinates', (0, 0)) # (hmin, hmax)
score = result.get('score', 0)
d_notes[label] = str(score)
final_img, _ = compose_label_image(base_img, label, result, coordinates[0],
with_empty=True,
render_fn=render_real_latex_text)
# 7. Save Image
save_path = os.path.join(output_dir, f"{label}.jpg")
final_img.save(save_path)
if result.get('error', "") != "empty-answer":
label_images.append(final_img)
# Save scores
with open(os.path.join(output_dir, "score.json"), "w") as f:
json.dump(d_notes, f, indent=4)
# Concatenate
if label_images:
max_w = max(i.width for i in label_images)
total_h = sum(i.height for i in label_images)
canvas = Image.new('RGB', (max_w, total_h))
cy = 0
for img in label_images:
canvas.paste(img, (0, cy))
cy += img.height
canvas.save(concat_path)
atomic_write_json(staging / "score.json", d_notes)
if label_images:
max_w = max(image.width for image in label_images)
total_h = sum(image.height for image in label_images)
canvas = Image.new('RGB', (max_w, total_h))
current_y = 0
for image in label_images:
canvas.paste(image, (0, current_y))
current_y += image.height
canvas.save(staging / "Concat.jpg")
elif labels_data:
problems = True
return "partial" if problems else "success"
def process_correction(root_dir, data, all_labels, overwrite=False):
# Ne pas thread cette application
# 1. Il faut protéger les appels à matplotlib
# 2. tu vas perdre les erreurs
for student_id, labels in sorted(data.items()):
statuses = [
process_student(student_id, labels, root_dir, all_labels, overwrite)
for student_id, labels in sorted(data.items())
]
return ExitCode.PARTIAL if "partial" in statuses else ExitCode.SUCCESS
def build_parser() -> argparse.ArgumentParser:
parser = evaluation_parser("Generate simple annotated copies.")
parser.add_argument(
"--overwrite",
action="store_true",
help="Replace existing student output directories",
)
return parser
def run(workspace: EvaluationWorkspace, *, overwrite: bool = False) -> ExitCode:
workspace.require_files("labels", "correction.json")
workspace.require_directories("Copies", "Par label")
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("Warning: no annotation data was found.")
return ExitCode.PARTIAL
result = process_correction(
workspace.root,
loaded.data,
labels,
overwrite=overwrite,
)
if loaded.warnings and result == ExitCode.SUCCESS:
return ExitCode.PARTIAL
return result
def main(argv: Sequence[str] | None = None) -> int:
parser = build_parser()
return execute(
parser,
argv,
lambda args: run(workspace_from_args(args), overwrite=args.overwrite),
)
import argparse
import utils
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Annotate copies")
parser.add_argument("root_dir", help="Directory containing the copies")
parser.add_argument("--overwrite", action="store_true", help="Reprocess even if Concat.jpg exists")
args = parser.parse_args()
root_dir = args.root_dir
labels = utils.read_all_labels(root_dir)
results = make_dictionary(root_dir)
# Results is : Copie id -> label -> {pdf_path, gemini_result, coordinates}
# Coordinates are the real coordinates (hmin, hmax) of the image in the Group
# print(results,"\n\n\n")
process_correction(root_dir, results, labels,overwrite=args.overwrite)
raise SystemExit(main())