Compare commits

...
2 Commits
Author SHA1 Message Date
sebastien 55d5cf16c9 Bnot fixes ; Missing énoncés fixes. 2026-10-23 11:15:05 +02:00
sebastien f39868207d dismissed wrong label triggers recorrection 2026-10-23 08:51:28 +02:00
21 changed files with 684 additions and 58 deletions

No files matched your search

+9
View File
@@ -170,6 +170,15 @@ Optional : Set proxy with ~export HTTPS_PROXY="http://10.0.0.1:3128"~
Fais les requêtes de correction à Gemini. Fais les requêtes de correction à Gemini.
L'énoncé et le corrigé sont recherchés dans =Text= et =Sol= : label
exact, fichier groupé, puis section parente la plus précise disponible.
Par exemple, =Ex 3 : I)3)b)= peut utiliser =Text/Ex 3 : I)= et
=Sol/Ex 3 : I)3)=. À la fin de la préparation de l'énoncé (personnelle
ou Gemini), une vérification signale les labels dont l'énoncé ou le
corrigé est absent ou vide. Ces avertissements n'empêchent pas de
poursuivre la correction avec les documents disponibles.
Les fichiers =Persp= restent facultatifs.
=copienator correct= peut être relancé sans supprimer son état. Les =copienator correct= peut être relancé sans supprimer son état. Les
fichiers =correction.json= et =correction_progress.json= sont mis à fichiers =correction.json= et =correction_progress.json= sont mis à
jour atomiquement. Avec =--overwrite=, leur version précédente reste jour atomiquement. Avec =--overwrite=, leur version précédente reste
+4 -1
View File
@@ -164,6 +164,7 @@ def rebuild(root: Path, mode: str, directory: Path, staging: Path, meta: dict,
from copienator.commands import annotating from copienator.commands import annotating
from copienator.commands.reading_grouped_annotations import ( from copienator.commands.reading_grouped_annotations import (
get_extra_pdfs_as_images, get_extra_pdfs_as_images,
save_checked_pdf,
save_paginated_pdf, save_paginated_pdf,
) )
@@ -179,7 +180,7 @@ def rebuild(root: Path, mode: str, directory: Path, staging: Path, meta: dict,
label for label, entry in info.items() if entry["not_empty"] label for label, entry in info.items() if entry["not_empty"]
} }
all_labels = utils.read_all_labels(root) all_labels = utils.read_all_labels(root)
images, groups = [], [] images, groups, checked_images = [], [], []
seen_contexts = set() seen_contexts = set()
order = sorted(bodies, key=utils.natural_key) order = sorted(bodies, key=utils.natural_key)
for label in order: for label in order:
@@ -198,6 +199,7 @@ def rebuild(root: Path, mode: str, directory: Path, staging: Path, meta: dict,
if info[label]["not_empty"]: if info[label]["not_empty"]:
images.append(image) images.append(image)
if label in filtered: if label in filtered:
checked_images.append((label, image))
extras = get_extra_pdfs_as_images(root, label, annotating, all_labels, extras = get_extra_pdfs_as_images(root, label, annotating, all_labels,
seen_contexts=seen_contexts) if mode == "BGnot" else [] seen_contexts=seen_contexts) if mode == "BGnot" else []
groups.append([*extras, image]) groups.append([*extras, image])
@@ -212,6 +214,7 @@ def rebuild(root: Path, mode: str, directory: Path, staging: Path, meta: dict,
save_paginated_pdf(groups, staging / "Concat_F.pdf") save_paginated_pdf(groups, staging / "Concat_F.pdf")
else: else:
stack([image for group in groups for image in group]).save(staging / "Concat_F.jpg") stack([image for group in groups for image in group]).save(staging / "Concat_F.jpg")
save_checked_pdf(root, checked_images, all_labels, staging / "Concat_F.pdf")
def stage_baseline(root: Path, mode: str, output: Path, staging: Path, def stage_baseline(root: Path, mode: str, output: Path, staging: Path,
+16 -10
View File
@@ -59,16 +59,20 @@ def copy_return_artifacts(source_dir: Path, destination_dir: Path) -> None:
shutil.copytree(answers_source, answers_destination) shutil.copytree(answers_source, answers_destination)
def copy_histogram(output_dir: Path) -> None: def copy_histogram(output_dir: Path, *, copy_merged_histogramme: bool = False) -> None:
"""Copy the score histogram beside the per-student output folders.""" """Copy the score histogram beside the per-student output folders."""
if not HISTOGRAM_PATH.is_file(): sources = [(HISTOGRAM_PATH, "histogramme.pdf")]
print(f"Warning: Missing histogram '{HISTOGRAM_PATH}'.") if copy_merged_histogramme:
return sources.append((HISTOGRAM_PATH.with_name("merged_histogramme.pdf"), "merged_histogramme.pdf"))
destination = output_dir / "histogramme.pdf" for source, filename in sources:
shutil.copy2(HISTOGRAM_PATH, destination) if not source.is_file():
print(f"Copied histogram: {destination}") print(f"Warning: Missing histogram '{source}'.")
continue
destination = output_dir / filename
shutil.copy2(source, destination)
print(f"Copied histogram: {destination}")
def process_images(base_dir, output_dir): def process_images(base_dir, output_dir, *, copy_merged_histogramme=False):
# 1. Load Data # 1. Load Data
try: try:
# header=None assumes the file starts directly with data. # header=None assumes the file starts directly with data.
@@ -148,18 +152,20 @@ def process_images(base_dir, output_dir):
for pdf_path in pdf_paths: for pdf_path in pdf_paths:
shutil.copy2(pdf_path, student_output / pdf_path.name) shutil.copy2(pdf_path, student_output / pdf_path.name)
copy_histogram(output_dir) copy_histogram(output_dir, copy_merged_histogramme=copy_merged_histogramme)
def main(argv=None): def main(argv=None):
parser = argparse.ArgumentParser(description="Stamp scores on exam copies.") parser = argparse.ArgumentParser(description="Stamp scores on exam copies.")
parser.add_argument("dir", type=Path, help="Root directory containing 'A Rendre' folder") parser.add_argument("dir", type=Path, help="Root directory containing 'A Rendre' folder")
parser.add_argument("--copy-merged-histogramme", action="store_true",
help="Also copy merged_histogramme.pdf from the histogram staging directory")
args = parser.parse_args(argv) args = parser.parse_args(argv)
base_dir = args.dir.expanduser().resolve() base_dir = args.dir.expanduser().resolve()
output_dir = OUTPUT_DIR / base_dir.name output_dir = OUTPUT_DIR / base_dir.name
output_dir.mkdir(parents=True, exist_ok=True) output_dir.mkdir(parents=True, exist_ok=True)
process_images(base_dir, output_dir) process_images(base_dir, output_dir, copy_merged_histogramme=args.copy_merged_histogramme)
return 0 return 0
+38 -11
View File
@@ -392,7 +392,7 @@ def reserve_next_group_idx(label: str) -> int:
reserved_group_indices[label] = idx + 1 reserved_group_indices[label] = idx + 1
return idx return idx
def handle_label_errors(pid, label, res, pdf_path): def handle_label_errors(pid, label, res, pdf_path, *, group_geometry=None):
"""Handles Gemini labeling errors, moves/copies files, and returns new tasks.""" """Handles Gemini labeling errors, moves/copies files, and returns new tasks."""
new_tasks = [] new_tasks = []
error_type = res.get("error") error_type = res.get("error")
@@ -410,7 +410,33 @@ def handle_label_errors(pid, label, res, pdf_path):
res["error"] = "wrg-lbl:cldtfix" res["error"] = "wrg-lbl:cldtfix"
return [] return []
if new_label == label: if new_label == label:
res["error"] = "" tprint(f"\t\tLabel confirmed for Copie{pid}: regrading {label}.")
contents, config = prompting.request_for_confirmed_label(INPUT_DIR, pdf_path, label, pid)
response = parse_correction_response(
call_gemini_with_retries(MODEL_ID_flash, contents, config)
)
if not isinstance(response, list) or len(response) != 1 or response[0].get("id") != pid:
raise ValueError(f"Confirmed-label regrading must return exactly Copie{pid}")
replacement = prompting.EvaluationEntry.model_validate(response[0]).result.model_dump()
if replacement["error"] == "wrong-label":
raise ValueError(f"Regrading rejected the confirmed label {label} for Copie{pid}")
if group_geometry is not None:
yming, ymaxg, width_r, total_height = group_geometry
for feedback in replacement["feedback"]:
box = feedback.get("box_2d")
if box and valid_feedback_box(box):
ymin, xmin, ymax, xmax = box
feedback["box_2d"] = [
(yming + ymin * (ymaxg - yming) / 1000) * 1000 / total_height,
xmin * width_r,
(yming + ymax * (ymaxg - yming) / 1000) * 1000 / total_height,
xmax * width_r,
]
# Keep the original response intact until regrading has succeeded.
res.clear()
res.update(replacement)
if res["error"] == "additional-answer":
return handle_label_errors(pid, label, res, pdf_path, group_geometry=group_geometry)
return [] return []
base_new_pdf_path = COPIES_DIR / f"Copie{pid}" / f"{new_label}.pdf" base_new_pdf_path = COPIES_DIR / f"Copie{pid}" / f"{new_label}.pdf"
@@ -574,14 +600,6 @@ def process_single_task(task_tuple, precomputed_response=None):
pdf_path = pdf_path.with_name(f"{label}_old.pdf") pdf_path = pdf_path.with_name(f"{label}_old.pdf")
current_suffix = "_old" current_suffix = "_old"
# 1. Gestion de empty-answer
if res.get("error") == "empty-answer":
old_path = pdf_path.with_name(f"{label}_old.pdf")
if pdf_path.exists() and pdf_path != old_path:
move_answer_pdf(pdf_path, old_path)
pdf_path = old_path
current_suffix = "_old"
if (not can_spawn_tasks) and res["error"] == "additional-answer": if (not can_spawn_tasks) and res["error"] == "additional-answer":
tprint("\tSwallowing an additional-answer from a subsequent task.") tprint("\tSwallowing an additional-answer from a subsequent task.")
res["error"]= "" res["error"]= ""
@@ -601,7 +619,8 @@ def process_single_task(task_tuple, precomputed_response=None):
raise CorrectionStopRequested raise CorrectionStopRequested
try: try:
new_tasks.extend( new_tasks.extend(
handle_label_errors(pid, label, res, pdf_path) handle_label_errors(pid, label, res, pdf_path,
group_geometry=(yming, ymaxg, width_r, total_height))
) )
except CorrectionStopRequested: except CorrectionStopRequested:
with io_lock: with io_lock:
@@ -614,6 +633,14 @@ def process_single_task(task_tuple, precomputed_response=None):
if res.get("error", "").startswith("wrg-lbl-moved-to:"): if res.get("error", "").startswith("wrg-lbl-moved-to:"):
current_suffix = "_old" current_suffix = "_old"
# Regrading a confirmed label can also identify an empty answer.
if res.get("error") == "empty-answer":
old_path = pdf_path.with_name(f"{label}_old.pdf")
if pdf_path.exists() and pdf_path != old_path:
move_answer_pdf(pdf_path, old_path)
pdf_path = old_path
current_suffix = "_old"
# 5. Enregistrer l'information dans correction.json # 5. Enregistrer l'information dans correction.json
if current_suffix: if current_suffix:
res["suffix"] = current_suffix res["suffix"] = current_suffix
+2 -1
View File
@@ -19,7 +19,7 @@ from copienator import (
workspace_from_args, workspace_from_args,
) )
from copienator.platform import WindowsLabelError, validate_windows_labels from copienator.platform import WindowsLabelError, validate_windows_labels
from copienator.utils import compile_to_pdf from copienator.utils import compile_to_pdf, warn_missing_exam_material
def fetch_and_save_sub_text(ex_id, indices, label, text_path): def fetch_and_save_sub_text(ex_id, indices, label, text_path):
@@ -327,6 +327,7 @@ def process_directory(workspace: EvaluationWorkspace) -> ExitCode:
labels_staging.replace(labels_file) labels_staging.replace(labels_file)
atomic_write_text(workspace.label_groups_file, atomic_write_text(workspace.label_groups_file,
"".join(", ".join(group) + "\n" for group in exercise_groups)) "".join(", ".join(group) + "\n" for group in exercise_groups))
warn_missing_exam_material(workspace)
return ExitCode.PARTIAL if had_errors else ExitCode.SUCCESS return ExitCode.PARTIAL if had_errors else ExitCode.SUCCESS
+5 -1
View File
@@ -1024,6 +1024,7 @@ def write_exam_outputs(workspace: EvaluationWorkspace, grouped_extraction: Group
print(f"Wrote {len(grouped_extraction.groups)} groups; checking PDF compilation...") print(f"Wrote {len(grouped_extraction.groups)} groups; checking PDF compilation...")
compilation_status = compile_existing_pdfs(workspace) compilation_status = compile_existing_pdfs(workspace)
utils.warn_missing_exam_material(workspace)
return ExitCode.PARTIAL if processing_errors or compilation_status != ExitCode.SUCCESS else ExitCode.SUCCESS return ExitCode.PARTIAL if processing_errors or compilation_status != ExitCode.SUCCESS else ExitCode.SUCCESS
@@ -1093,7 +1094,10 @@ def main(argv: Sequence[str] | None = None) -> int:
if args.restart and (args.compile_only or args.groups_only or args.persp_only): if args.restart and (args.compile_only or args.groups_only or args.persp_only):
raise CliError('--restart cannot be combined with a partial action', ExitCode.INVALID_ARGUMENTS) raise CliError('--restart cannot be combined with a partial action', ExitCode.INVALID_ARGUMENTS)
if args.compile_only: if args.compile_only:
return compile_existing_pdfs(workspace_from_args(args)) workspace = workspace_from_args(args)
status = compile_existing_pdfs(workspace)
utils.warn_missing_exam_material(workspace)
return status
if args.groups_only or args.persp_only: if args.groups_only or args.persp_only:
kwargs = {'delete_persp': True} if args.delete_persp else {} kwargs = {'delete_persp': True} if args.delete_persp else {}
return refine_existing(workspace_from_args(args), return refine_existing(workspace_from_args(args),
@@ -270,6 +270,12 @@ def apply_actions_and_regenerate(
concat_image.save(staging / "Concat.jpg") concat_image.save(staging / "Concat.jpg")
if filtered_image is not None: if filtered_image is not None:
filtered_image.save(staging / "Concat_F.jpg") filtered_image.save(staging / "Concat_F.jpg")
# Imported here because the grouped reader also uses this module.
from copienator.commands.reading_grouped_annotations import save_checked_pdf
save_checked_pdf(
workspace.root, list(zip(filtered_labels, filtered, strict=True)),
all_labels, staging / "Concat_F.pdf",
)
stage_baseline(workspace.root, "Bnot", output_dir, staging, dirty_images, stage_baseline(workspace.root, "Bnot", output_dir, staging, dirty_images,
header_heights, build_answer_info(scores, labels_data, answer_labels), header_heights, build_answer_info(scores, labels_data, answer_labels),
filtered_labels, scores) filtered_labels, scores)
@@ -147,6 +147,18 @@ def save_paginated_pdf(
) )
def save_checked_pdf(root, labelled_images, all_labels, output_path):
"""Build the checked-mode return PDF with the same supplements as BGnot."""
groups = []
seen_contexts = set()
for label, image in labelled_images:
extras = get_extra_pdfs_as_images(
root, label, annotating, all_labels, seen_contexts=seen_contexts
)
groups.append([*extras, image])
save_paginated_pdf(groups, output_path)
def _scan_annotation_directory( def _scan_annotation_directory(
directory: Path, directory: Path,
only_ids: set[str] | None = None, only_ids: set[str] | None = None,
@@ -420,6 +432,12 @@ def apply_actions_and_regenerate_grouped(
[image for group in filtered_groups for image in group] [image for group in filtered_groups for image in group]
) )
filtered_image.save(staging / "Concat_F.jpg") filtered_image.save(staging / "Concat_F.jpg")
if annotation_dir == "Bnot":
save_checked_pdf(
workspace.root,
[(label, dirty_images[label]) for label in filtered_labels],
all_labels, staging / "Concat_F.pdf",
)
stage_baseline(workspace.root, annotation_dir, output_dir, staging, dirty_images, stage_baseline(workspace.root, annotation_dir, output_dir, staging, dirty_images,
header_heights, build_answer_info(scores, labels_data, answer_labels, touched), header_heights, build_answer_info(scores, labels_data, answer_labels, touched),
filtered_labels, scores) filtered_labels, scores)
+27 -2
View File
@@ -227,7 +227,11 @@ def generate_request(input_dir, file, full_label, *, thinking_level=settings.THI
) )
] ]
generate_content_config = types.GenerateContentConfig( return (contents, grading_generation_config(thinking_level))
def grading_generation_config(thinking_level):
return types.GenerateContentConfig(
automatic_function_calling=types.AutomaticFunctionCallingConfig(disable=True), automatic_function_calling=types.AutomaticFunctionCallingConfig(disable=True),
thinking_config=thinking_config(thinking_level), thinking_config=thinking_config(thinking_level),
seed=fresh_seed(), seed=fresh_seed(),
@@ -235,7 +239,28 @@ def generate_request(input_dir, file, full_label, *, thinking_level=settings.THI
response_mime_type= "application/json", response_mime_type= "application/json",
response_json_schema= TypeAdapter(List[EvaluationEntry]).json_schema() response_json_schema= TypeAdapter(List[EvaluationEntry]).json_schema()
) )
return (contents, generate_content_config)
def request_for_confirmed_label(input_dir, pdf_path, label, student_id):
prompt = make_prompt(input_dir, label)
prompt += f"""
Cette image contient uniquement la réponse de l'élève d'identifiant `{student_id}`.
Il n'y a pas d'en-tête imprimé à rechercher : utilise cet identifiant exact.
Une vérification indépendante a confirmé que cette réponse appartient bien
à la question `{label}`. Reprends sa correction mathématique depuis le début,
avec une nouvelle note et des commentaires précis sur les erreurs ou les étapes
manquantes. Une preuve incorrecte ou incomplète n'est pas une erreur de label.
Ne renvoie pas `wrong-label` et ne rejette pas la réponse comme hors sujet.
Les autres cas `additional-answer` et `empty-answer` restent applicables.
Renvoie une liste JSON contenant exactement un résultat, pour `{student_id}`.
Les coordonnées éventuelles portent sur l'image fournie entière.
"""
contents = [types.Content(role="user", parts=[
types.Part.from_bytes(data=get_single_image_bytes(pdf_path), mime_type="image/jpeg"),
types.Part.from_text(text=prompt),
])]
return contents, grading_generation_config(settings.THINKING_LEVEL_FLASH)
from pdf2image import convert_from_path from pdf2image import convert_from_path
+52 -14
View File
@@ -93,10 +93,7 @@ def pdf_images_of_contexts(root_dir, label, all_labels):
def get_exam_file_content(folder_path, mode, label): def get_exam_file_content(folder_path, mode, label):
""" """Use the most specific exact, grouped, or enclosing section available."""
Retrieves content from the Text or Sol directory for a specific label.
Checks for exact filename matches or grouped 'prefix[a,b,c]' filenames.
"""
target_dir = Path(folder_path) / mode target_dir = Path(folder_path) / mode
if not target_dir.is_dir(): if not target_dir.is_dir():
return None return None
@@ -109,23 +106,37 @@ def get_exam_file_content(folder_path, mode, label):
if direct_file.is_file(): if direct_file.is_file():
return direct_file.read_text(encoding="utf-8") return direct_file.read_text(encoding="utf-8")
# 2. Search for grouped files: prefix[suffix1,suffix2,...] candidates = []
for file_path in target_dir.iterdir(): for file_path in sorted(target_dir.iterdir()):
if not file_path.is_file(): if not file_path.is_file():
continue continue
name = file_path.name name = file_path.name
if file_path.suffix in {".pdf", ".tex"}:
continue
covered_labels = [name]
if "[" in name and name.endswith("]"): if "[" in name and name.endswith("]"):
# Split 'prefix[suffixes]' -> 'prefix', 'suffixes'
prefix, rest = name.split("[", 1) prefix, rest = name.split("[", 1)
suffixes = rest[:-1].split(",") # Remove trailing ']' and split covered_labels = [prefix + suffix.strip() for suffix in rest[:-1].split(",")]
for covered in covered_labels:
# Check if any reconstructed label matches if covered == safe_label or _is_parent_label(covered, safe_label):
for s in suffixes: candidates.append((len(covered), name == covered, file_path))
if (prefix + s) == safe_label: if candidates:
return file_path.read_text(encoding="utf-8") # Prefer a narrower section, then its direct file over a grouped file.
candidates.sort(key=lambda item: (-item[0], -item[1], item[2].name))
return candidates[0][2].read_text(encoding="utf-8")
return None return None
def _is_parent_label(parent, label):
if not parent or not label.startswith(parent):
return False
suffix = label[len(parent):]
# Respect component boundaries: Ex 1 must not match Ex 10, nor 1) match 10).
return bool(
suffix.startswith((" ", ":", "."))
or (parent.endswith(")") and re.match(r"[A-Za-z0-9]+\)", suffix))
)
def get_label_text_content(folder, label): def get_label_text_content(folder, label):
return get_exam_file_content(folder, "Text", label) return get_exam_file_content(folder, "Text", label)
@@ -135,6 +146,33 @@ def get_label_sol_content(folder, label):
def get_label_persp_content(folder, label): def get_label_persp_content(folder, label):
return get_exam_file_content(folder, "Persp", label) return get_exam_file_content(folder, "Persp", label)
def warn_missing_exam_material(workspace):
"""Report prompt coverage after statement preparation without blocking work."""
if not workspace.labels_file.is_file():
print("[Vérification des prompts] Avertissement : fichier labels absent ; vérification impossible.")
return
labels = workspace.read_labels()
missing_count = 0
for label in labels:
missing = []
for mode, title in (("Text", "énoncé"), ("Sol", "corrigé")):
try:
content = get_exam_file_content(workspace.root, mode, label)
if not content or not content.strip() or content.strip().rstrip(" :") == label:
missing.append(f"{title} absent ou vide ({mode})")
except (OSError, UnicodeError) as exc:
missing.append(f"{title} illisible ({mode}) : {exc}")
if missing:
missing_count += 1
print(f"[Vérification des prompts] Avertissement — {label} : {' ; '.join(missing)}")
print(
f"[Vérification des prompts] {len(labels) - missing_count}/{len(labels)} labels "
"avec énoncé et corrigé non vides."
)
if missing_count:
print("[Vérification des prompts] Vous pouvez continuer ; les corrections des labels signalés utiliseront des documents incomplets.")
import tempfile import tempfile
import shutil import shutil
import subprocess import subprocess
+17 -4
View File
@@ -893,11 +893,13 @@ class CopienatorApp(tk.Tk):
value = False # Deletion is a one-run choice, never a saved default. value = False # Deletion is a one-run choice, never a saved default.
if step.id == 'statement' and spec.name == 'restart': if step.id == 'statement' and spec.name == 'restart':
value = False value = False
if step.id == "final_score" and spec.name == "copy_merged_histogramme" and spec.name not in values:
value = bool(self.state_store.step("personal_read").get("wsnm_run", False))
if step.id == "modify_annotations" and spec.name == "annotation_dir" and spec.name not in values: if step.id == "modify_annotations" and spec.name == "annotation_dir" and spec.name not in values:
reader = self.state_store.step("read_annotations") reader = self.state_store.step("read_annotations")
value = annotation_mode_for_reader(reader) value = annotation_mode_for_reader(reader)
choices = spec.choices choices = spec.choices
if spec.name == "annotation_dir" and step.id in {"export", "import"}: if spec.name == "annotation_dir" and step.id in {"export", "import", "giving_names"}:
choices, value = self._annotation_directory_choices(step.id) choices, value = self._annotation_directory_choices(step.id)
argument_label = ttk.Label(self.form, text=f"{spec.label} ⓘ") argument_label = ttk.Label(self.form, text=f"{spec.label} ⓘ")
argument_label.grid(row=row, column=0, sticky="nw", pady=4, padx=(0, 8)) argument_label.grid(row=row, column=0, sticky="nw", pady=4, padx=(0, 8))
@@ -1033,7 +1035,7 @@ class CopienatorApp(tk.Tk):
controls = ttk.Frame(self.form) controls = ttk.Frame(self.form)
controls.grid(row=row, column=0, columnspan=2, sticky="w", pady=8) controls.grid(row=row, column=0, columnspan=2, sticky="w", pady=8)
self.review_buttons = [] self.review_buttons = []
for command in ("re", "wsn", "we"): for command in ("re", "wsn", "we", "mei", "wsnm"):
button = ttk.Button( button = ttk.Button(
controls, text=f"gestion_classe {command}", controls, text=f"gestion_classe {command}",
command=lambda command=command: self._run_current_step(review_command=command), command=lambda command=command: self._run_current_step(review_command=command),
@@ -1301,8 +1303,10 @@ class CopienatorApp(tk.Tk):
evaluation = self.evaluation evaluation = self.evaluation
detected = detected_annotation_directories(evaluation) if evaluation else () detected = detected_annotation_directories(evaluation) if evaluation else ()
choices = detected or ANNOTATION_DIRECTORIES choices = detected or ANNOTATION_DIRECTORIES
if step_id == "giving_names":
choices = ANNOTATION_DIRECTORIES
if step_id == "export": if step_id in {"export", "giving_names"}:
annotation = self.state_store.step("annotation") annotation = self.state_store.step("annotation")
variant = annotation.get("last_run_variant", annotation.get("variant", "grouped")) variant = annotation.get("last_run_variant", annotation.get("variant", "grouped"))
preferred = ANNOTATION_VARIANT_DIRECTORIES.get(str(variant), "BGnot") preferred = ANNOTATION_VARIANT_DIRECTORIES.get(str(variant), "BGnot")
@@ -1608,7 +1612,7 @@ class CopienatorApp(tk.Tk):
if orientation_repair and (not step or step.id != 'labels'): if orientation_repair and (not step or step.id != 'labels'):
return return
if review_command is not None and ( if review_command is not None and (
not step or step.id != "personal_read" or review_command not in {"re", "wsn", "we"} not step or step.id != "personal_read" or review_command not in {"re", "wsn", "we", "mei", "wsnm"}
): ):
return return
if not step or not evaluation or not self.state_store.evaluation: if not step or not evaluation or not self.state_store.evaluation:
@@ -1671,6 +1675,10 @@ class CopienatorApp(tk.Tk):
command = [sys.executable, '-m', 'copienator', 'repair-orientation', str(evaluation)] command = [sys.executable, '-m', 'copienator', 'repair-orientation', str(evaluation)]
if step.id == "personal_read": if step.id == "personal_read":
command = ["gestion_classe", review_command] command = ["gestion_classe", review_command]
if review_command not in {"mei", "wsnm"}:
evaluation_name = str(self._values().get("evaluation_name", evaluation.name)).strip()
if evaluation_name:
command.append(evaluation_name)
if variant.dangerous and not messagebox.askyesno( if variant.dangerous and not messagebox.askyesno(
"Nettoyage irréversible" if step.id == "clean" else "Confirmation requise", "Nettoyage irréversible" if step.id == "clean" else "Confirmation requise",
variant.danger_warning variant.danger_warning
@@ -1718,6 +1726,11 @@ class CopienatorApp(tk.Tk):
self._append_console(f"\n$ {command_display(command)}\n") self._append_console(f"\n$ {command_display(command)}\n")
try: try:
self.runner.start(command, self.repository, environment, log_path) self.runner.start(command, self.repository, environment, log_path)
if step.id == "personal_read" and review_command == "wsnm" and not self.state_store.step(step.id).get("wsnm_run"):
self.state_store.update_step(step.id, wsnm_run=True)
values = dict(self.state_store.step("final_score").get("values", {}))
values["copy_merged_histogramme"] = True
self.state_store.update_step("final_score", values=values)
if step.id == 'statement_persp': if step.id == 'statement_persp':
self.arg_vars['delete_persp'].set(False) self.arg_vars['delete_persp'].set(False)
self._save_current_form() self._save_current_form()
+24 -4
View File
@@ -11,6 +11,7 @@ from copienator import configuration
from copienator.configuration import ALWAYS_CROP from copienator.configuration import ALWAYS_CROP
EVALUATION = "${evaluation}" EVALUATION = "${evaluation}"
EVALUATION_NAME = "${evaluation_name}"
@dataclass(frozen=True) @dataclass(frozen=True)
@@ -83,6 +84,14 @@ def arg_correction_workers(variants: tuple[str, ...] = ()) -> ArgumentSpec:
) )
def arg_evaluation_name() -> ArgumentSpec:
return ArgumentSpec(
"evaluation_name", "Nom de l’évaluation", default=EVALUATION_NAME,
positional=True,
help="Nom transmis à gestion_classe, prérempli avec le nom du dossier de l’évaluation et modifiable.",
)
def build_workflow(show_personal_steps: bool) -> list[StepDefinition]: def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
def python(ident, label, script, **kwargs): def python(ident, label, script, **kwargs):
supports_verbose = kwargs.pop("supports_verbose", True) supports_verbose = kwargs.pop("supports_verbose", True)
@@ -794,8 +803,9 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
"personal_read", "personal_read",
"Étapes personnelles", "Étapes personnelles",
"Relire avec gestion_classe", "Relire avec gestion_classe",
"Utilisez les boutons gestion_classe re, wsn et we, puis cliquez sur « Marquer réussie ».", "Utilisez les boutons gestion_classe re, wsn, we, mei et wsnm, puis cliquez sur « Marquer réussie ».",
(CommandVariant("default", "gestion_classe re", "gestion_classe", "external", ("re",)),), (CommandVariant("default", "gestion_classe re", "gestion_classe", "external", ("re",), fixed_args_before_positionals=True),),
arguments=(arg_evaluation_name(),),
personal=True, personal=True,
), ),
StepDefinition( StepDefinition(
@@ -803,7 +813,8 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
"Étapes personnelles", "Étapes personnelles",
"Écrire les notes finales", "Écrire les notes finales",
"Exécute gestion_classe wsent.", "Exécute gestion_classe wsent.",
(CommandVariant("default", "gestion_classe wsent", "gestion_classe", "external", ("wsent",)),), (CommandVariant("default", "gestion_classe wsent", "gestion_classe", "external", ("wsent",), fixed_args_before_positionals=True),),
arguments=(arg_evaluation_name(),),
personal=True, personal=True,
), ),
StepDefinition( StepDefinition(
@@ -814,7 +825,14 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
"Génère ensuite les dossiers de diffusion avec le score final, puis copie " "Génère ensuite les dossiers de diffusion avec le score final, puis copie "
"gestion_classe/Staging/histogramme.pdf dans le dossier Server/copies de l’évaluation.", "gestion_classe/Staging/histogramme.pdf dans le dossier Server/copies de l’évaluation.",
(python("default", "Score final", "add-final-score", supports_verbose=False),), (python("default", "Score final", "add-final-score", supports_verbose=False),),
arguments=(arg_target("le dossier de l’évaluation"),), arguments=(
arg_target("le dossier de l’évaluation"),
ArgumentSpec(
"copy_merged_histogramme", "copy merged histogramme", "bool",
"--copy-merged-histogramme",
help="Copie aussi Staging/merged_histogramme.pdf. Cochée par défaut après le premier lancement de gestion_classe wsnm pour cette évaluation.",
),
),
personal=True, personal=True,
), ),
StepDefinition( StepDefinition(
@@ -920,6 +938,8 @@ def build_refaire_workflow() -> list[StepDefinition]:
def value_for_default(value: object, evaluation_arg: str) -> object: def value_for_default(value: object, evaluation_arg: str) -> object:
if value == EVALUATION_NAME:
return Path(evaluation_arg).name
return evaluation_arg if value == EVALUATION else value return evaluation_arg if value == EVALUATION else value
+6 -5
View File
@@ -165,8 +165,9 @@ Every label in `score.json` has an object containing exactly four fields:
images; return publication uses current `score.json` values for this field. images; return publication uses current `score.json` values for this field.
`info.json` is always exported, even when individual JPEGs or the named PDF are `info.json` is always exported, even when individual JPEGs or the named PDF are
disabled. `touched` describes the source filtered PDF. In normal `Anot` and disabled. `touched` tracks BGnot filtered-PDF membership. In `Anot` and
`Bnot` flows, no filtered PDF is produced and all `touched` values are false. `Bnot` flows, all `touched` values remain false. Bnot also produces a filtered,
paginated PDF with context, questions and solutions, alongside its filtered JPEG.
This file replaces `touched.json` and the internal `answer_labels.json` manifest. This file replaces `touched.json` and the internal `answer_labels.json` manifest.
Recompile old annotations, then run `giving-names`; successful regeneration and Recompile old annotations, then run `giving-names`; successful regeneration and
@@ -273,8 +274,8 @@ ones. Weighting and the final overall mark belong to the separate grading flow.
source if `score.json` and either `Concat.jpg` or `info.json` exist, source if `score.json` and either `Concat.jpg` or `info.json` exist,
otherwise it tries `Anot/CopieXX`. This also permits returns for an entirely otherwise it tries `Anot/CopieXX`. This also permits returns for an entirely
empty copy. It links the PDF only if `Concat_F.pdf` exists. The normal empty copy. It links the PDF only if `Concat_F.pdf` exists. The normal
`Bnot` reader produces a filtered `Concat_F.jpg`, and simple annotation produces `Bnot` reader produces both `Concat_F.jpg` and `Concat_F.pdf` for nonempty
`Concat.jpg`; these paths do not guarantee a filtered PDF. filtered answers. Simple annotation produces `Concat.jpg` without a filtered PDF.
- Preparing returns removes disabled named outputs, and removes older named - Preparing returns removes disabled named outputs, and removes older named
outputs whose source is now absent, including broken symlinks. outputs whose source is now absent, including broken symlinks.
- `add-final-score` writes to `FINAL_SCORE_OUTPUT_DIR/<evaluation>/`. It stamps - `add-final-score` writes to `FINAL_SCORE_OUTPUT_DIR/<evaluation>/`. It stamps
@@ -324,7 +325,7 @@ applied snapshot, not the draft. Selective redo preserves unaffected baseline
layers so previous transfers are never included twice. layers so previous transfers are never included twice.
The operation rebuilds per-label JPEGs, `info.json`, `Concat.jpg`, and the filtered The operation rebuilds per-label JPEGs, `info.json`, `Concat.jpg`, and the filtered
BGnot PDF or Bnot JPEG. Every changed non-empty body is included in the filtered BGnot PDF or Bnot JPEG and PDF. Every changed non-empty body is included in the filtered
output, with its destination label's supplementary material. `score.json` is not output, with its destination label's supplementary material. `score.json` is not
modified. The GUI marks downstream return preparation stale. Rerun `giving-names` modified. The GUI marks downstream return preparation stale. Rerun `giving-names`
without `--update` to refresh both full outputs and individual answer exports. without `--update` to refresh both full outputs and individual answer exports.
+1 -1
View File
@@ -478,7 +478,7 @@ texte**, enregistrez puis **Recharger**. **Aperçu du résultat** montre les cor
résultants pour la copie saisie. Cliquez sur **Make modifications** pour appliquer résultants pour la copie saisie. Cliquez sur **Make modifications** pour appliquer
le fichier enregistré. Chaque label conserve son en-tête ; les scores restent le fichier enregistré. Chaque label conserve son en-tête ; les scores restent
inchangés. Les images individuelles, les images combinées, le PDF filtré BGnot inchangés. Les images individuelles, les images combinées, le PDF filtré BGnot
(ou l’image filtrée Bnot) et les informations de présence sont reconstruits. (ou l’image et le PDF filtrés Bnot) et les informations de présence sont reconstruits.
Les réponses modifiées non vides sont incluses dans le résultat filtré, avec les Les réponses modifiées non vides sont incluses dans le résultat filtré, avec les
compléments propres à leur label. compléments propres à leur label.
+26
View File
@@ -12,6 +12,32 @@ from copienator.commands import add_final_score
class AddFinalScoreTests(unittest.TestCase): class AddFinalScoreTests(unittest.TestCase):
def test_merged_histogram_is_copied_only_when_selected(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
histogram = root / 'Staging/histogramme.pdf'
histogram.parent.mkdir()
histogram.write_bytes(b'histogram')
merged = histogram.with_name('merged_histogramme.pdf')
merged.write_bytes(b'merged')
output = root / 'output'
output.mkdir()
with patch.object(add_final_score, 'HISTOGRAM_PATH', histogram):
add_final_score.copy_histogram(output)
self.assertFalse((output / merged.name).exists())
add_final_score.copy_histogram(output, copy_merged_histogramme=True)
self.assertEqual((output / 'histogramme.pdf').read_bytes(), b'histogram')
self.assertEqual((output / merged.name).read_bytes(), b'merged')
def test_cli_forwards_merged_histogram_option(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
with patch.object(add_final_score, 'OUTPUT_DIR', root / 'output'), patch.object(
add_final_score, 'process_images'
) as process:
self.assertEqual(add_final_score.main([str(root), '--copy-merged-histogramme']), 0)
process.assert_called_once_with(root, root / 'output' / root.name, copy_merged_histogramme=True)
def test_creates_complete_student_directory(self): def test_creates_complete_student_directory(self):
with tempfile.TemporaryDirectory() as temporary: with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary) root = Path(temporary)
+100
View File
@@ -0,0 +1,100 @@
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
import pymupdf
from PIL import Image
from copienator import EvaluationWorkspace, ExitCode, atomic_write_json, configuration
from copienator.commands import annotating, giving_names, reading_annotations as standard
from copienator.commands import reading_grouped_annotations as grouped
from copienator.annotation_modifications import BASELINE, rebuild
from copienator.filesystem import staged_files
from copienator.json_io import read_json
class BnotPdfReturnTests(unittest.TestCase):
def setUp(self):
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
self.root = Path(temporary.name)
self.workspace = EvaluationWorkspace(self.root)
self.answer = self.root / 'Copies/Copie01/Ex 1.pdf'
self.answer.parent.mkdir(parents=True)
self.answer.touch()
(self.root / 'labels').write_text('Ex 1\nEx 2\n')
atomic_write_json(self.root / 'Copies/Copie01.json', {'name': 'Student'})
(self.root / 'names').write_text('Student\n')
self.output = self.root / 'Bnot/Copie01'
self.output.mkdir(parents=True)
atomic_write_json(self.output / 'bnote.json', {'images': [
{'id': '01', 'label': 'Ex 1', 'header_height': 0, 'hmin': 0, 'hmax': 80},
]})
self.data = {'01': {'Ex 1': {'pdf_path': self.answer, 'coordinates': (0, 0),
'result': {'score': 2, 'error': '', 'feedback': []}}}}
for mode in ('Text2', 'Sol2'):
path = self.root / mode / 'Ex 1.pdf'
path.parent.mkdir()
path.touch()
self.rendered = []
def render(self, path):
self.rendered.append(Path(path))
color = {'Text2': 'red', 'Sol2': 'blue'}.get(Path(path).parent.name, 'yellow')
return Image.new('RGB', (100, 60), color), 100, 60
def read(self, grouped_mode=False, score=2):
self.data['01']['Ex 1']['result']['score'] = score
with (
patch.object(annotating, 'make_base_image', side_effect=self.render),
patch.object(annotating, 'compose_label_image', return_value=(Image.new('RGB', (100, 80), 'yellow'), 0)),
):
if grouped_mode:
result, _ = grouped.apply_actions_and_regenerate_grouped(
self.workspace, self.data, '01', [], {}, ['Ex 1', 'Ex 2'], annotation_dir='Bnot')
else:
result = standard.apply_actions_and_regenerate(
self.workspace, self.data, '01', [], None, ['Ex 1', 'Ex 2'])
self.assertEqual(result, ExitCode.SUCCESS)
def check_pdf(self):
pdf = self.output / 'Concat_F.pdf'
with pymupdf.open(pdf) as document:
self.assertGreater(len(document), 0)
pixmap = document[0].get_pixmap(matrix=pymupdf.Matrix(100 / 72, 100 / 72))
image = Image.frombytes('RGB', (pixmap.width, pixmap.height), pixmap.samples)
# Supplements precede the answer, with the existing page margins.
for y, channel in ((35, 0), (95, 2)):
pixel = image.getpixel((60, y))
self.assertGreater(pixel[channel], 180)
self.assertLess(pixel[1], 70)
self.assertTrue((self.output / 'Concat_F.jpg').is_file())
self.assertIn(self.root / 'Text2/Ex 1.pdf', self.rendered)
self.assertIn(self.root / 'Sol2/Ex 1.pdf', self.rendered)
return pdf
def test_reader_creates_pdf_and_naming_publishes_it(self):
self.read()
pdf = self.check_pdf()
with patch.multiple(configuration, RETURN_ANSWERS_ENABLED=False, RETURN_PDF_ENABLED=True):
self.assertEqual(giving_names.run(self.workspace, annotation_dir='Bnot'), ExitCode.SUCCESS)
named = self.root / 'A Rendre/Student (01)/Student.pdf'
self.assertEqual(named.read_bytes(), pdf.read_bytes())
# Recompilation with no filtered answers must remove the old PDF too.
self.read(score=4)
self.assertFalse(pdf.exists())
self.assertFalse((self.output / 'Concat_F.jpg').exists())
def test_grouped_redo_into_bnot_also_creates_pdf(self):
self.read(grouped_mode=True)
self.check_pdf()
def test_modification_rebuild_keeps_pdf_and_supplements(self):
self.read()
meta = read_json(self.output / BASELINE)
scores = read_json(self.output / 'score.json')
with patch.object(annotating, 'make_base_image', side_effect=self.render):
with staged_files(self.output) as staging:
rebuild(self.root, 'Bnot', self.output, staging, meta, [], scores)
self.check_pdf()
+120
View File
@@ -0,0 +1,120 @@
import json
import tempfile
import unittest
from pathlib import Path
from unittest.mock import Mock, patch
from copienator import EvaluationWorkspace, atomic_write_json, read_json
from copienator.commands import correction
class ConfirmedLabelRegradingTests(unittest.TestCase):
def setUp(self):
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
self.addCleanup(correction.stop_requested.clear)
self.root = Path(temporary.name)
self.workspace = EvaluationWorkspace(self.root)
(self.root / 'labels').write_text('Ex 1\nEx 2\n')
for mode in ('Text', 'Sol'):
(self.root / mode).mkdir()
(self.root / mode / 'Ex 1').write_text(f'{mode} material')
self.group = self.root / 'Par label/Ex 1/Group_1.jpg'
self.group.parent.mkdir(parents=True)
self.group.touch()
atomic_write_json(self.group.with_suffix('.json'), [
['02', 0, 400, 1], ['01', 450, 950, .5],
])
self.pdf = self.root / 'Copies/Copie01/Ex 1.pdf'
self.pdf.parent.mkdir(parents=True)
self.pdf.write_bytes(b'original answer')
args = correction.build_parser().parse_args([str(self.root)])
correction.configure_runtime(self.workspace, [(str(self.group), 'Ex 1')], args, api_client=Mock())
self.original = [
{'id': '02', 'result': {'score': 4, 'feedback': [], 'error': ''}},
{'id': '01', 'result': {'score': 0, 'error': 'wrong-label',
'feedback': [{'text': 'Wrong question', 'box_2d': None}]}},
]
self.regraded = [{'id': '01', 'result': {
'score': 2.5, 'error': '', 'feedback': [
{'text': 'Monotonicity is missing', 'box_2d': [100, 200, 300, 400]},
],
}}]
def process(self, responses):
with (
patch.object(correction.prompting, 'generate_request', return_value=([], {})),
patch.object(correction.prompting, 'get_single_image_bytes', return_value=b'image'),
patch.object(correction, 'call_gemini_with_retries', side_effect=responses) as call,
):
tasks = correction.process_single_task((str(self.group), 'Ex 1'), json.dumps(self.original))
return tasks, call
def test_confirmed_label_replaces_rejection_and_maps_boxes_to_original_group(self):
tasks, calls = self.process(['Ex 1', json.dumps(self.regraded)])
self.assertEqual(tasks, [])
self.assertEqual(calls.call_count, 2)
self.assertEqual(correction.errors_summary, [])
saved = read_json(self.workspace.correction_file)['Ex 1'][0]
self.assertEqual(saved[0], self.original[0])
result = saved[1]['result']
self.assertEqual(result['score'], 2.5)
self.assertEqual(result['error'], '')
self.assertEqual(result['feedback'][0]['text'], 'Monotonicity is missing')
box = result['feedback'][0]['box_2d']
self.assertAlmostEqual(box[0], 500 * 1000 / 950)
self.assertAlmostEqual(box[2], 600 * 1000 / 950)
self.assertEqual([box[1], box[3]], [100, 200])
self.assertEqual(self.pdf.read_bytes(), b'original answer')
self.assertEqual(correction.completed_tasks, [(str(self.group), 'Ex 1')])
def test_invalid_regrading_never_publishes_stale_zero_or_marks_group_complete(self):
for response in (
[],
[{'id': '02', 'result': self.regraded[0]['result']}],
[self.regraded[0], self.regraded[0]],
[self.original[1]],
[{'id': '01', 'result': {'error': ''}}],
):
with self.subTest(response=response):
self.process(['Ex 1', json.dumps(response)])
self.assertFalse(self.workspace.correction_file.exists())
self.assertEqual(correction.completed_tasks, [])
self.assertEqual(correction.results['Ex 1'], [])
self.assertEqual(self.pdf.read_bytes(), b'original answer')
def test_interrupted_regrading_keeps_wrong_label_in_pending_response(self):
with self.assertRaises(correction.CorrectionStopRequested):
self.process(['Ex 1', correction.CorrectionStopRequested()])
pending = read_json(self.workspace.correction_pending_responses_file)
self.assertEqual(json.loads(pending[str(self.group)]), self.original)
self.assertFalse(self.workspace.correction_file.exists())
self.process(['Ex 1', json.dumps(self.regraded)])
self.assertEqual(read_json(self.workspace.correction_pending_responses_file), {})
self.assertEqual(read_json(self.workspace.correction_file)['Ex 1'][0][1]['result']['score'], 2.5)
def test_empty_answer_from_regrading_uses_normal_empty_answer_handling(self):
response = [{'id': '01', 'result': {'score': 0, 'feedback': [], 'error': 'empty-answer'}}]
self.process(['Ex 1', json.dumps(response)])
result = read_json(self.workspace.correction_file)['Ex 1'][0][1]['result']
self.assertEqual(result['error'], 'empty-answer')
self.assertEqual(result['suffix'], '_old')
self.assertFalse(self.pdf.exists())
self.assertEqual(self.pdf.with_name('Ex 1_old.pdf').read_bytes(), b'original answer')
def test_different_label_keeps_existing_move_behavior(self):
(self.pdf.parent / 'Ex 2.pdf').touch()
_, calls = self.process(['Ex 2'])
self.assertEqual(calls.call_count, 1)
result = read_json(self.workspace.correction_file)['Ex 1'][0][1]['result']
self.assertEqual(result['error'], 'wrg-lbl:Ex 2?')
self.assertEqual(result['delayed'], [['wrong-label', 'Ex 2']])
def test_additional_answer_from_regrading_still_gets_resolved(self):
(self.pdf.parent / 'Ex 2.pdf').touch()
self.regraded[0]['result']['error'] = 'additional-answer'
_, calls = self.process(['Ex 1', json.dumps(self.regraded), '["Ex 1", "Ex 2"]'])
self.assertEqual(calls.call_count, 3)
result = read_json(self.workspace.correction_file)['Ex 1'][0][1]['result']
self.assertEqual(result['score'], 2.5)
self.assertEqual(result['delayed'], [['add-label', 'Ex 2']])
+150
View File
@@ -0,0 +1,150 @@
import io
import tempfile
import unittest
from pathlib import Path
from unittest.mock import Mock, patch
from copienator import EvaluationWorkspace, ExitCode
from copienator import prompting
from copienator.commands import correction, enonce_info, gemini_for_enonce
from copienator.utils import get_exam_file_content, warn_missing_exam_material
class ExamMaterialTests(unittest.TestCase):
def setUp(self):
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
self.root = Path(temporary.name)
def write(self, mode, label, text):
path = self.root / mode / label
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(text)
return path
def test_prompt_uses_section_statement_and_question_solution_for_nested_label(self):
self.write('Text', 'Ex 3', 'Entire exercise')
self.write('Text', 'Ex 3 : I)', 'Definitions and all questions in part I')
self.write('Sol', 'Ex 3', 'Entire solution')
self.write('Sol', 'Ex 3 : I)3)', 'Solution to 3)a) and 3)b)')
self.write('Persp', 'Ex 3', 'Exercise marking guidance')
for label in ('Ex 3 : I)3)a)', 'Ex 3 : I)3)b)'):
prompt = prompting.make_prompt(self.root, label)
self.assertIn('Definitions and all questions in part I', prompt)
self.assertIn('Solution to 3)a) and 3)b)', prompt)
self.assertIn('Exercise marking guidance', prompt)
self.assertNotIn('Entire exercise', prompt)
self.assertNotIn('Entire solution', prompt)
def test_exact_then_grouped_then_most_specific_parent_precedence(self):
label = 'Ex 3 : I)3)b)'
self.write('Sol', 'Ex 3', 'exercise')
self.write('Sol', 'Ex 3 : I)', 'section')
self.write('Sol', 'Ex 3 : I)3)', 'question')
group = self.write('Sol', 'Ex 3 : I)3)[a),b)]', 'grouped answers')
self.assertEqual(get_exam_file_content(self.root, 'Sol', label), 'grouped answers')
exact = self.write('Sol', label, 'exact answer')
self.assertEqual(get_exam_file_content(self.root, 'Sol', label), 'exact answer')
exact.unlink()
group.unlink()
self.assertEqual(get_exam_file_content(self.root, 'Sol', label), 'question')
def test_grouped_parent_can_cover_deeper_questions(self):
self.write('Text', 'Ex 3 : [I),II)]', 'two sections')
self.assertEqual(get_exam_file_content(self.root, 'Text', 'Ex 3 : I)3)b)'), 'two sections')
self.assertIsNone(get_exam_file_content(self.root, 'Text', 'Ex 3 : III)1)'))
def test_parent_lookup_respects_boundaries_and_ignores_rendered_files(self):
self.write('Text', 'Ex 1', 'wrong exercise')
self.write('Text', 'Ex 10 : I)1)', 'wrong question')
self.write('Text', 'Ex 10 : I)10)a).tex', 'rendered latex')
self.write('Text', 'Ex 10 : I)10)a).pdf', 'rendered pdf')
self.assertIsNone(get_exam_file_content(self.root, 'Text', 'Ex 10 : I)10)a)'))
self.write('Text', 'Ex 10 : I)', 'right section')
self.assertEqual(get_exam_file_content(self.root, 'Text', 'Ex 10 : I)10)a)'), 'right section')
def test_slash_sanitization_and_dot_hierarchies(self):
self.write('Text', 'Sheet_A', 'sheet context')
self.assertEqual(get_exam_file_content(self.root, 'Text', 'Sheet/A : 1)a)'), 'sheet context')
self.write('Sol', '1.a', 'solution')
self.assertEqual(get_exam_file_content(self.root, 'Sol', '1.a.i'), 'solution')
self.assertIsNone(get_exam_file_content(self.root, 'Sol', '1.ab.i'))
def test_missing_material_does_not_block_prompt_or_regrading_request(self):
for mode in ('Text', 'Sol'):
self.write(mode, 'Ex 1', ' \n ')
self.assertIsInstance(prompting.make_prompt(self.root, 'Ex 1 : 1)'), str)
with patch.object(prompting, 'get_single_image_bytes', return_value=b'answer') as render:
prompting.request_for_confirmed_label(self.root, 'answer.pdf', 'Ex 1 : 1)', '01')
render.assert_called_once()
def test_coverage_warns_for_every_missing_label_and_accepts_parent_material(self):
self.write('', 'labels', 'Ex 1 : 1)\nEx 2 : 1)\nEx 3 : 1)\n')
self.write('Text', 'Ex 1', 'all questions')
self.write('Sol', 'Ex 1', 'all solutions')
self.write('Text', 'Ex 2', 'questions')
self.write('Sol', 'Ex 2 : 1)', ' \n ')
output = io.StringIO()
with patch('sys.stdout', output):
warn_missing_exam_material(EvaluationWorkspace(self.root))
report = output.getvalue()
self.assertIn('1/3 labels', report)
self.assertIn('Ex 2 : 1) : corrigé absent ou vide', report)
self.assertIn('Ex 3 : 1) : énoncé absent ou vide', report)
self.assertIn('Vous pouvez continuer', report)
def test_correction_can_continue_when_material_is_missing(self):
workspace = EvaluationWorkspace(self.root)
(self.root / 'Copies').mkdir()
(self.root / 'Par label').mkdir()
(self.root / 'labels').write_text('Ex 1 : 1)\n')
args = correction.build_parser().parse_args([str(self.root)])
with (
patch.object(correction, 'discover_tasks', return_value=([('group.jpg', 'Ex 1 : 1)')], [])),
patch.object(correction, 'configure_runtime') as configure,
patch.object(correction, 'run_configured', return_value=ExitCode.SUCCESS) as run,
):
self.assertEqual(correction.run(workspace, [self.root], args, api_client=Mock()), ExitCode.SUCCESS)
configure.assert_called_once()
run.assert_called_once()
def test_personal_statement_checks_material_without_changing_success(self):
self.write('', 'enonce.tex', '%%SHEETINFO : {"id": 42}\n')
output = io.StringIO()
def fetch(url):
return io.BytesIO(b'Question### ###Rubric' if '/emacs/' in url else b'fragment')
with (
patch.object(enonce_info.urllib.request, 'urlopen', side_effect=fetch),
patch.object(enonce_info, 'compile_to_pdf'),
patch('sys.stdout', output),
):
status = enonce_info.process_directory(EvaluationWorkspace(self.root))
self.assertEqual(status, ExitCode.SUCCESS)
self.assertIn('Ex 1 : corrigé absent ou vide', output.getvalue())
self.assertIn('0/1 labels', output.getvalue())
def test_gemini_statement_checks_material_without_changing_success(self):
extraction = gemini_for_enonce.GroupedExamExtraction(groups=[[
gemini_for_enonce.QuestionItem(label='Ex 1', question_content='Question', solution_content=''),
]])
output = io.StringIO()
with (
patch.object(gemini_for_enonce, 'generate_rubrics', return_value={'Ex 1': 'Criteria'}),
patch.object(gemini_for_enonce, 'compile_existing_pdfs', return_value=ExitCode.SUCCESS),
patch('sys.stdout', output),
):
status = gemini_for_enonce.write_exam_outputs(EvaluationWorkspace(self.root), extraction, Mock())
self.assertEqual(status, ExitCode.SUCCESS)
self.assertIn('Ex 1 : corrigé absent ou vide', output.getvalue())
self.assertIn('0/1 labels', output.getvalue())
def test_compile_only_rechecks_existing_material(self):
self.write('', 'labels', 'Ex 1\n')
output = io.StringIO()
with (
patch.object(gemini_for_enonce, 'compile_existing_pdfs', return_value=ExitCode.SUCCESS),
patch('sys.stdout', output),
):
status = gemini_for_enonce.main([str(self.root), '--compile-only'])
self.assertEqual(status, ExitCode.SUCCESS)
self.assertIn('Ex 1 : énoncé absent ou vide', output.getvalue())
+43 -3
View File
@@ -71,14 +71,19 @@ class GuiConvenienceTests(unittest.TestCase):
self.app.current_step = step self.app.current_step = step
self.app._render_step() self.app._render_step()
self.assertEqual([button.cget("text") for button in self.app.review_buttons], self.assertEqual([button.cget("text") for button in self.app.review_buttons],
["gestion_classe re", "gestion_classe wsn", "gestion_classe we"]) ["gestion_classe re", "gestion_classe wsn", "gestion_classe we", "gestion_classe mei", "gestion_classe wsnm"])
self.assertEqual(self.app.command_box.winfo_manager(), "") self.assertEqual(self.app.command_box.winfo_manager(), "")
self.assertEqual(self.app.run_button.cget("text"), "Marquer réussie") self.assertEqual(self.app.run_button.cget("text"), "Marquer réussie")
self.assertEqual(self.app.arg_vars["evaluation_name"].get(), self.evaluation.name)
self.app.arg_vars["evaluation_name"].set("DS02VB custom")
with patch.object(self.app.runner, "start") as start, \ with patch.object(self.app.runner, "start") as start, \
patch.object(self.app, "_move_selection_from") as advance: patch.object(self.app, "_move_selection_from") as advance:
for button, command in zip(self.app.review_buttons, ("re", "wsn", "we")): for button, command in zip(self.app.review_buttons, ("re", "wsn", "we", "mei", "wsnm")):
button.invoke() button.invoke()
self.assertEqual(start.call_args.args[0], ["gestion_classe", command]) expected = ["gestion_classe", command]
if command not in {"mei", "wsnm"}:
expected.append("DS02VB custom")
self.assertEqual(start.call_args.args[0], expected)
self.app._finish_process(0, False) self.app._finish_process(0, False)
self.assertNotEqual(self.app.state_store.step(step.id)["status"], "success") self.assertNotEqual(self.app.state_store.step(step.id)["status"], "success")
advance.assert_not_called() advance.assert_not_called()
@@ -87,6 +92,41 @@ class GuiConvenienceTests(unittest.TestCase):
start.assert_not_called() start.assert_not_called()
self.assertEqual(self.app.state_store.step(step.id)["status"], "success") self.assertEqual(self.app.state_store.step(step.id)["status"], "success")
def test_merged_histogram_default_tracks_first_wsnm_launch_and_remains_editable(self):
from copienator_gui.workflow import build_workflow
from copienator_gui.state import StateStore
steps = {step.id: step for step in build_workflow(True)}
self.app.step_by_id.update(steps)
self.app.current_step = steps['final_score']
self.app._render_step()
self.assertFalse(self.app.arg_vars['copy_merged_histogramme'].get())
self.app._save_current_form()
def launch():
self.app.current_step = steps['personal_read']
self.app._render_step()
with patch.object(self.app.runner, 'start') as start:
self.app.review_buttons[4].invoke()
self.assertEqual(start.call_args.args[0], ['gestion_classe', 'wsnm'])
self.app._finish_process(0, False)
launch()
reloaded = StateStore()
reloaded.load(self.evaluation)
self.assertTrue(reloaded.step('personal_read')['wsnm_run'])
self.assertTrue(reloaded.step('final_score')['values']['copy_merged_histogramme'])
self.app.current_step = steps['final_score']
self.app._render_step()
self.assertTrue(self.app.arg_vars['copy_merged_histogramme'].get())
self.assertIn('--copy-merged-histogramme', self.app._make_command())
self.app.arg_vars['copy_merged_histogramme'].set(False)
self.app._save_current_form()
launch()
self.app.current_step = steps['final_score']
self.app._render_step()
self.assertFalse(self.app.arg_vars['copy_merged_histogramme'].get())
self.assertNotIn('--copy-merged-histogramme', self.app._make_command())
def test_annotation_import_cleanup_can_be_unchecked(self): def test_annotation_import_cleanup_can_be_unchecked(self):
self.app.tree.selection_set("annotation") self.app.tree.selection_set("annotation")
self.app.update() self.app.update()
+1 -1
View File
@@ -1761,7 +1761,7 @@ class StandardCliTests(unittest.TestCase):
module.stop_requested.set() module.stop_requested.set()
yield SimpleNamespace(text=response) yield SimpleNamespace(text=response)
def resolve_error(_pid, _label, result, _pdf_path): def resolve_error(_pid, _label, result, _pdf_path, **_kwargs):
self.assertEqual(result["error"], error_type) self.assertEqual(result["error"], error_type)
result["error"] = "" result["error"] = ""
return [] return []
+19
View File
@@ -7,8 +7,27 @@ from copienator import prompting
class PromptingTests(unittest.TestCase): class PromptingTests(unittest.TestCase):
def test_confirmed_label_request_keeps_grading_material_and_identifies_single_answer(self):
with (
patch.object(prompting, "get_single_image_bytes", return_value=b"answer"),
patch.object(prompting, "get_label_text_content", return_value="Question"),
patch.object(prompting, "get_label_sol_content", return_value="Solution"),
patch.object(prompting, "get_label_persp_content", return_value="Marking criteria"),
):
contents, config = prompting.request_for_confirmed_label("evaluation", "answer.pdf", "Ex 2 : 2)b)", "01")
prompt = contents[0].parts[1].text
self.assertEqual(contents[0].parts[0].inline_data.data, b"answer")
for text in ("Question", "Solution", "Marking criteria", "`01`", "`Ex 2 : 2)b)`",
"Reprends sa correction mathématique", "Ne renvoie pas `wrong-label`"):
self.assertIn(text, prompt)
self.assertEqual(config.response_mime_type, "application/json")
def test_live_prompt_includes_only_metadata_ids_in_order(self) -> None: def test_live_prompt_includes_only_metadata_ids_in_order(self) -> None:
with tempfile.TemporaryDirectory() as directory: with tempfile.TemporaryDirectory() as directory:
for mode in ("Text", "Sol"):
folder = Path(directory) / mode
folder.mkdir()
(folder / "I)").write_text(f"{mode} material")
image = Path(directory) / "Group_16.jpg" image = Path(directory) / "Group_16.jpg"
image.write_bytes(b"image") image.write_bytes(b"image")
image.with_suffix(".json").write_text( image.with_suffix(".json").write_text(