Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
55d5cf16c9 | ||
|
|
f39868207d |
No files matched your search
@@ -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
|
||||||
|
|||||||
@@ -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,
|
||||||
|
|||||||
@@ -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
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -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
@@ -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
@@ -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
@@ -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()
|
||||||
|
|||||||
@@ -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
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -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.
|
||||||
@@ -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.
|
||||||
|
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|||||||
@@ -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()
|
||||||
@@ -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']])
|
||||||
@@ -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())
|
||||||
@@ -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()
|
||||||
|
|||||||
@@ -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 []
|
||||||
|
|||||||
@@ -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(
|
||||||
|
|||||||
Reference in new issue
Block a user