Compare commits

..
3 Commits
Author SHA1 Message Date
sebastien 06d7bad04e Tradution en français des prompts 2026-09-06 19:28:54 +02:00
sebastien 2313d51a62 Refaire session. 2026-09-06 19:20:51 +02:00
sebastien 2ff1a9b7b9 Refaire fixes and GUI support 2026-09-06 19:06:42 +02:00
18 changed files with 2318 additions and 269 deletions
+126 -29
View File
@@ -346,39 +346,136 @@ le plan sans rien supprimer, et =--yes= pour omettre la confirmation
interactive.
* Autres
** Recorrection d'une seule copie (peu testé)
** Recorrection d'une copie ou de quelques questions
!! Attention, refaire ne marchera pas si tu fais une annotation non
groupée into refaire !!
Dans le GUI, ouvrir la section =Refaire des copies (facultatif)=,
repliée par défaut. Ajouter les copies et leurs questions depuis les
listes déroulantes, ou choisir =Toute la copie=, puis enregistrer la
sélection. Le dossier du passage principal est présélectionné d'après
le dernier mode utilisé et les dossiers présents ; vérifier ce choix.
Pour reprendre une même question dans toute la classe, choisir
=Toutes les copies=, sélectionner la question, puis =+ Ajouter=.
Seules les copies ayant un PDF de réponse pour cette question (normal
ou =_new=) sont ajoutées ; le nombre de copies sans réponse est affiché.
Une copie déjà sélectionnée entièrement reste sélectionnée entièrement.
1. Redécoupage
+ =python -m copienator review-labels InterroTest/Copie01.pdf=
+ =python -m copienator split-answers InterroTest/Copie20.pdf=
2. Créer =refaire.json=, avec un contenu comme
Les boutons =Corrigés (Sol)= et =Consignes de notation (Persp)= ouvrent
les dossiers des textes utilisés par la correction. Modifier et enregistrer
les fichiers des questions concernées avant de relancer =Refaire la correction=.
Ces boutons sont aussi disponibles dans le parcours principal.
Le choix =PDF à vérifier= propose =Automatique=, =Par question (groupé)=
ou =Par copie=. En automatique, une question présente dans plusieurs
copies déclenche le regroupement ; sinon les PDF sont produits par copie.
Les groupes sont limités en hauteur : une question pour toute la classe
peut donc produire quelques PDF plutôt qu'un fichier par élève.
Le GUI écrit =refaire.json= et guide ensuite le parcours ci-dessous.
La vérification du découpage, le redécoupage et la recorrection peuvent
être ignorés. Pour plusieurs copies, les vérifications et découpages
s'exécutent successivement ; un échec ou une interruption arrête la suite.
Ce parcours a sa propre progression : les boutons de navigation du
passage principal ne l'ouvrent pas automatiquement. Après la fusion,
les étapes de restitution déjà terminées sont marquées à revalider.
Pour enchaîner les reprises, deux boutons sont disponibles dans ce parcours :
- =Nouvelle reprise= vide la sélection et remet les étapes de reprise à zéro.
- =Refaire la même sélection= conserve les copies et les questions du dernier
enregistrement, mais remet également les étapes de reprise à zéro.
Les choix du passage principal et de présentation des PDF sont conservés.
Enregistrer ensuite la sélection avant de poursuivre.
Attention : appeler =Nouvelle reprise= seulement après avoir importé les
résultats et exécuté =Mettre à jour les copies finales=. La même précaution
s'applique à =Refaire la même sélection=. Un avertissement est affiché avant
les deux actions, avec une mention supplémentaire si la fusion n'est pas
marquée réussie. Annuler conserve la reprise actuelle. Après une fusion
réussie, utiliser ces boutons pour recommencer, plutôt que modifier la
sélection du passage terminé.
Chaque nouveau passage utilise =Reprises/reprise-DATE-HEURE-ID/BRnot=.
Le passage précédent garde ses PDF, ses retours manuscrits, sa sélection
et une copie de la progression du GUI. Au premier changement de passage,
l'ancien =BRnot= à la racine est également copié dans =Reprises=.
Ces archives concernent les fichiers de vérification, pas un mécanisme
permettant d'annuler les modifications des copies finales.
Le fichier =refaire-session.json= désigne le passage actif ; les commandes
habituelles =--refaire= le suivent automatiquement. Sans ce fichier, le
fonctionnement historique dans =BRnot= à la racine reste disponible.
Dans la suite, =BRnot= désigne le dossier de la reprise active.
L'export d'un passage identifié utilise son propre sous-dossier dans
=EXPORT_DIR/Évaluation= et préfixe les noms des PDF par son identifiant.
Conserver les noms complets au retour et placer les PDF directement dans
=IMPORT_DIR=. L'import ignore les retours des autres passages ; aucun retour
du passage actif donne un résultat partiel. Ainsi, deux reprises de la même
question ne partagent pas les mêmes noms de fichiers exportés.
Ce flux fonctionne après =annotate-grouped= (=BGnot=),
=annotate-checks= (=Bnot=) ou =annotate-simple= (=Anot=).
Terminer d'abord la lecture des annotations du passage principal
(=read-grouped= ou =read-annotations= pour les modes à cases).
Conserver les dossiers d'annotation et leurs fichiers de référence.
1. Si nécessaire, reprendre le découpage :
+ =python -m copienator review-labels Interro/Copies/Copie01.pdf=
+ =python -m copienator split-answers Interro/Copies/Copie01.pdf=
Vérifier les réponses découpées avant de relancer la correction,
notamment les fichiers =_new= et =_old= issus de résolutions manuelles.
2. Créer =Interro/refaire.json= :
: [["Copie02", []],
: ["Copie01", ["Ex 1 : 1)"]]]
3. Appeler =correction= avec --refaire. Il doit créer des groupes
individuels, faire des requêtes, et remplacer les corrections
précédentes (à sauver ailleurs).
Une liste vide sélectionne toute la copie ; sinon donner les labels
exacts des questions (pas seulement le nom de l'exercice).
3. =python -m copienator correct Interro --refaire=
Crée des groupes individuels et remplace les corrections sélectionnées.
Les anciennes corrections sont conservées dans =overwritten_correction.json=.
Cette étape peut être omise si les corrections sont modifiées à la main.
4. Générer les PDF de vérification, selon la présentation souhaitée :
+ Par question : =python -m copienator annotate-grouped Interro --refaire --overwrite=
+ Par copie : =python -m copienator annotate-checks Interro --refaire --overwrite=
Les deux commandes produisent uniquement les réponses sélectionnées
dans =BRnot=, avec des cases à cocher, quel que soit le mode du passage
principal. Le mode groupé garde les identifiants des élèves et regroupe
les réponses par label sans demander de modifier =label_groups=.
Cela ne nécessite pas d'avoir généré =Bnot= ou =BGnot= auparavant.
Attention : =--overwrite= remplace le contenu du =BRnot= actif,
y compris ses annotations manuscrites, mais pas les autres reprises. Une génération incomplète
conserve l'ancien =BRnot=. Sans =--overwrite=, un =BRnot= existant est refusé.
5. Vider les dossiers personnels d'export/import des anciens fichiers,
puis =python -m copienator export Interro --refaire=.
Annoter les PDF sur la tablette, puis placer les PDF retournés dans
=IMPORT_DIR= en conservant leur nom exporté (nom de groupe ou =Copie01.pdf=).
Même sans modification manuscrite, retourner le PDF pour valider ce passage.
6. =python -m copienator import Interro --refaire=
7. Fusionner dans le dossier du passage principal :
+ Groupé : =python -m copienator read-grouped Interro --refaire=
+ Cases : =python -m copienator read-grouped Interro --refaire --annotation-dir Bnot=
+ Simple : =python -m copienator read-grouped Interro --refaire --annotation-dir Anot=
Ou non, si tu veux le faire à la main.
4. ?? Si je fais refaire, avant d'avoir créer les annotating with
checks, que se passe-t-il ???
5. Appeler =python -m copienator annotate-checks --refaire --overwrite=
6. =python -m copienator export --refaire Interro24=
6. =python -m copienator import --refaire Interro24=
7. =python -m copienator read-grouped --refaire Interro24=
Le lecteur reconnaît les retours par question comme les retours par
copie grâce aux métadonnées et réattribue les cases et notes à chaque
élève. Un groupe manquant laisse intactes les copies qui en dépendent.
Il reconstruit la copie complète, conserve les réponses non
sélectionnées et leurs scores, et remplace les réponses sélectionnées
par celles de =BRnot=. Dans la compilation filtrée, les images déjà
enregistrées des questions non sélectionnées sont conservées par
prudence, même si leur score est parfait, pour ne pas perdre de notes.
Les anciennes cases et notes manuscrites des
questions refaites sont remplacées. Les autres copies restent intactes.
En mode simple, une image =Concat_annotated.jpg= ou =.jpeg= importée
doit conserver les dimensions de l'image exportée ; les parties non
sélectionnées sont conservées. Le fichier =refaire_simple_layout.json=
mémorise le découpage de cette image pour les passages suivants.
Avec =--refaire=, =refaire.json= et le dossier =BRnot= sont des
prérequis obligatoires ; leur absence produit le code de sortie 3.
Les sorties finales (=Concat.jpg=, images par question, =score.json=
et compilation filtrée) sont mises à jour dans =BGnot=, =Bnot= ou
=Anot= ; les PDF et références du passage principal restent ceux de
ce passage. Pour une nouvelle retouche, reprendre ce flux =--refaire=,
sans relire ensuite les anciennes annotations avec le lecteur normal.
Relancer ensuite les étapes habituelles de calcul des notes et de diffusion.
** Exemple de replotting, refaire d'une copie
1. replot it.
2. `python -m copienator split-answers DS09VA/Copies/Copie25.pdf`
this will get rid of old/new.
!! Attention, et si ça dégage un new : bad bad bad.
3. Make `refaire.json`, avec la copie, et les labels à refaire.
4. `python -m copienator correct DS09VA --refaire`
5. `python -m copienator annotate-checks DS09VA --refaire`
6. `python -m copienator import Interro24 --refaire`
=refaire.json=, =BRnot= et le dossier du passage principal sont
obligatoires (code de sortie 3 s'ils manquent). Une copie dont les
fichiers de retour sont incomplets est laissée intacte (code 4).
Ne pas ajouter =--update-score= sauf pour imposer volontairement les
anciens scores, y compris ceux des questions refaites.
+12 -3
View File
@@ -36,7 +36,12 @@ def _coordinate_index(
if not workspace.groups_dir.is_dir():
return index, [f"Group directory not found: {workspace.groups_dir}"]
for metadata_path in sorted(workspace.groups_dir.glob("*/Group_*.json")):
# A redo appends a new numbered group; its coordinates supersede the old group.
for metadata_path in sorted(
workspace.groups_dir.glob("*/Group_*.json"),
key=lambda path: int(path.stem.removeprefix("Group_")),
reverse=True,
):
image_path = metadata_path.with_suffix(".jpg")
try:
entries = read_json(metadata_path)
@@ -160,12 +165,16 @@ def load_annotation_data(
warnings.append(f"Ignoring malformed correction batch for {label!r}")
continue
for item in raw_batch:
if not isinstance(item, dict) or not isinstance(item.get("result"), dict):
if not isinstance(item, dict) or not isinstance(
item.get("result"), dict
):
warnings.append(f"Ignoring malformed correction item for {label!r}")
continue
student_id = str(item.get("id", ""))
if not student_id:
warnings.append(f"Ignoring correction item without an id for {label!r}")
warnings.append(
f"Ignoring correction item without an id for {label!r}"
)
continue
result = item["result"]
suffix = str(result.get("suffix", ""))
+42 -12
View File
@@ -10,9 +10,6 @@ from typing import Any
from PIL import Image, ImageDraw
from reportlab.pdfgen import canvas
from copienator.commands import annotating
from copienator.commands import annotating_with_checks
from copienator import utils
from copienator import (
CliError,
EvaluationWorkspace,
@@ -22,9 +19,11 @@ from copienator import (
evaluation_parser,
execute,
read_json,
utils,
workspace_from_args,
)
from copienator.annotation_data import load_annotation_data
from copienator.commands import annotating, annotating_with_checks
from copienator.filesystem import staged_directory
from copienator.utils import natural_key
@@ -184,7 +183,9 @@ def _serialize_label_groups(groups: list[list[str]]) -> str:
return "".join(",".join(group) + "\n" for group in groups)
def _load_label_groups(workspace: EvaluationWorkspace, labels: list[str]) -> list[list[str]]:
def _load_label_groups(
workspace: EvaluationWorkspace, labels: list[str]
) -> list[list[str]]:
label_groups = workspace.label_groups_file
if not label_groups.exists():
gemini_groups = _gemini_label_groups(workspace, labels)
@@ -328,20 +329,38 @@ def _generate_groups(
return generated, problems
def run(workspace: EvaluationWorkspace, *, overwrite: bool = False) -> ExitCode:
def run(
workspace: EvaluationWorkspace, *, overwrite: bool = False, refaire: bool = False
) -> ExitCode:
workspace.require_files("labels", "correction.json")
workspace.require_directories("Copies", "Par label")
labels = utils.read_all_labels(workspace.root)
groups = _load_label_groups(workspace, labels)
loaded = load_annotation_data(workspace)
refaire_list = annotating_with_checks._load_refaire(workspace) if refaire else None
loaded = load_annotation_data(workspace, refaire_list=refaire_list)
groups = (
[
[label]
for label in sorted(
{label for answers in loaded.data.values() for label in answers},
key=natural_key,
)
]
if refaire
else _load_label_groups(workspace, labels)
)
for warning in loaded.warnings:
print(f"Warning: {warning}")
if not loaded.data:
print("Warning: no annotation data was found.")
return ExitCode.PARTIAL
output_root = workspace.annotation_dir("grouped")
if overwrite:
output_root = workspace.annotation_dir("refaire" if refaire else "grouped")
if refaire and output_root.exists() and not overwrite:
raise CliError(
"BRnot already exists; use --overwrite to replace the previous redo."
)
if overwrite or refaire:
class IncompleteGroupedOutput(Exception):
pass
@@ -356,7 +375,9 @@ def run(workspace: EvaluationWorkspace, *, overwrite: bool = False) -> ExitCode:
if generated == 0 or problems or loaded.warnings:
raise IncompleteGroupedOutput
except IncompleteGroupedOutput:
print("Warning: grouped overwrite was incomplete; previous BGnot was preserved.")
print(
f"Warning: grouped overwrite was incomplete; previous {output_root.name} was preserved."
)
return ExitCode.PARTIAL
else:
output_root.mkdir(parents=True, exist_ok=True)
@@ -375,7 +396,14 @@ def run(workspace: EvaluationWorkspace, *, overwrite: bool = False) -> ExitCode:
def build_parser() -> argparse.ArgumentParser:
parser = evaluation_parser("Generate annotated PDFs grouped by labels.")
parser.add_argument("--overwrite", action="store_true", help="Replace BGnot safely")
parser.add_argument(
"--overwrite", action="store_true", help="Replace annotation outputs safely"
)
parser.add_argument(
"--refaire",
action="store_true",
help="Group only the answers in refaire.json, writing to BRnot",
)
return parser
@@ -384,7 +412,9 @@ def main(argv: Sequence[str] | None = None) -> int:
return execute(
parser,
argv,
lambda args: run(workspace_from_args(args), overwrite=args.overwrite),
lambda args: run(
workspace_from_args(args), overwrite=args.overwrite, refaire=args.refaire
),
)
+48 -6
View File
@@ -14,8 +14,6 @@ matplotlib.use("Agg")
from PIL import Image, ImageFont
from reportlab.pdfgen import canvas
from copienator.commands import annotating
from copienator import utils
from copienator import (
CliError,
EvaluationWorkspace,
@@ -24,9 +22,11 @@ from copienator import (
execute,
read_json,
target_parser,
utils,
workspace_from_target,
)
from copienator.annotation_data import load_annotation_data
from copienator.commands import annotating
from copienator.filesystem import staged_directory
from copienator.utils import natural_key
@@ -46,7 +46,9 @@ except OSError:
def draw_checkbox(draw, x, y, size=BOX_SIZE, label=None, fill="white"):
if label:
draw.text((x - BOX_SIZE - 5, y + 2), str(label), fill="black", font=CHECKBOX_FONT)
draw.text(
(x - BOX_SIZE - 5, y + 2), str(label), fill="black", font=CHECKBOX_FONT
)
draw.rectangle([x, y, x + size, y + size], fill=fill, outline="black", width=2)
return [x, y, x + size, y + size]
@@ -150,8 +152,13 @@ def _render_student(
*,
overwrite: bool,
output_mode: str,
output_root: Path | None = None,
) -> str:
output_dir = workspace.annotation_dir(output_mode) / f"Copie{student_id}"
output_dir = (
output_root
if output_root is not None
else workspace.annotation_dir(output_mode)
) / f"Copie{student_id}"
if _output_complete(output_dir) and not overwrite:
print(f"Skipping {student_id}: output is complete.")
return "skipped"
@@ -178,6 +185,7 @@ def _render_student(
draw_callback=checkbox_renderer.callback,
)
if final_image is None:
problems = True
continue
label_images.append(final_image)
checkbox_groups.append(checkbox_renderer.checkboxes)
@@ -280,6 +288,39 @@ def run(
output_mode = "refaire" if refaire else "checks"
tasks = sorted(loaded.data.items(), key=lambda item: natural_key(item[0]))
if refaire:
output_root = workspace.annotation_dir("refaire")
if output_root.exists() and not overwrite:
raise CliError(
"BRnot already exists; use --overwrite to replace the previous redo."
)
class IncompleteRedo(Exception):
pass
try:
with staged_directory(output_root) as staging:
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
futures = [
executor.submit(
_render_student,
workspace,
student_id,
labels,
overwrite=True,
output_mode="refaire",
output_root=staging,
)
for student_id, labels in tasks
]
statuses = [future.result() for future in futures]
if loaded.warnings or any(status != "success" for status in statuses):
raise IncompleteRedo
except IncompleteRedo:
print("Warning: incomplete redo generation; previous BRnot was preserved.")
return ExitCode.PARTIAL
return ExitCode.SUCCESS
statuses: list[str] = []
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
futures = [
@@ -302,7 +343,9 @@ def run(
def build_parser() -> argparse.ArgumentParser:
parser = target_parser("Generate annotated PDFs with checkboxes.")
parser.add_argument("--overwrite", action="store_true", help="Replace existing outputs")
parser.add_argument(
"--overwrite", action="store_true", help="Replace existing outputs"
)
parser.add_argument(
"--refaire",
action="store_true",
@@ -328,4 +371,3 @@ def main(argv: Sequence[str] | None = None) -> int:
if __name__ == "__main__":
raise SystemExit(main())
+7 -3
View File
@@ -3,7 +3,6 @@ import sys
from collections.abc import Sequence
from pathlib import Path
from copienator.configuration import EXPORT_DIR
from copienator import (
EvaluationWorkspace,
ExitCode,
@@ -11,6 +10,7 @@ from copienator import (
execute,
workspace_from_args,
)
from copienator.configuration import EXPORT_DIR
from copienator.platform import replace_with_link_or_copy
ANNOTATION_DIRECTORIES = ("BGnot", "Bnot", "Anot")
@@ -21,8 +21,12 @@ def export_directory(
source_dir_name: str,
) -> ExitCode:
workspace.require_directories(source_dir_name)
source_dir = workspace.root / source_dir_name
source_dir = workspace.annotation_dir("refaire") if source_dir_name == "BRnot" else workspace.root / source_dir_name
session_id = workspace.refaire_session_id if source_dir_name == "BRnot" else None
prefix = f"{session_id}__" if session_id else ""
sync_dir = Path(EXPORT_DIR).expanduser() / workspace.name
if session_id:
sync_dir /= session_id
sync_dir.mkdir(parents=True, exist_ok=True)
subdirs = [directory for directory in source_dir.iterdir() if directory.is_dir()]
@@ -48,7 +52,7 @@ def export_directory(
)
missing_outputs += 1
continue
destination = sync_dir / f"{subdir.name}{concat_file.suffix.lower()}"
destination = sync_dir / f"{prefix}{subdir.name}{concat_file.suffix.lower()}"
method = replace_with_link_or_copy(concat_file, destination, prefer="hardlink")
print(f"Exported: {destination} ({method})")
return ExitCode.PARTIAL if missing_outputs else ExitCode.SUCCESS
+67 -45
View File
@@ -42,47 +42,54 @@ api_key = config.API_KEY
# --- Modèles pour la Requête 1 ---
class QuestionOnlyItem(BaseModel):
label: str = Field(description="The unique label of the question (e.g., '1.a', 'Exercice 1')")
question_content: str = Field(description="The source text of the question, strictly extracted from the enonce file, EXCLUDING the label itself.")
label: str = Field(description="Label unique de la question (par exemple '1.a' ou 'Exercice 1').")
question_content: str = Field(description="Texte source de la question, extrait exactement du fichier d’énoncé, SANS le label lui-même.")
class ExamQuestions(BaseModel):
questions: list[QuestionOnlyItem]
# --- Modèles pour la Requête 2 ---
class SolutionOnlyItem(BaseModel):
label: str = Field(description="The exact unique label of the question provided in the input.")
solution_content: str = Field(description="The source text of the solution, strictly extracted from the correction file.")
label: str = Field(description="Label exact de la question fourni en entrée, à conserver sans traduction.")
solution_content: str = Field(description="Texte source de la solution, extrait exactement du fichier de correction.")
class ExamSolutions(BaseModel):
solutions: list[SolutionOnlyItem]
# --- Modèles pour la Requête 3 ---
class ExtractedContext(BaseModel):
target_question_label: str = Field(description="The exact label of the FIRST question that comes immediately AFTER this information in the exam.")
last_question_label: str = Field(description="The exact label of the LAST question that uses or relies on this information.")
context_content: str = Field(description="The source text of the definitions, notations, or hypotheses, extracted from the enonce.")
target_question_label: str = Field(description="Label exact de la PREMIÈRE question située immédiatement APRÈS cette information dans l’énoncé.")
last_question_label: str = Field(description="Label exact de la DERNIÈRE question qui utilise cette information.")
context_content: str = Field(description="Texte source des définitions, notations ou hypothèses, extrait de l’énoncé.")
class ExamContext(BaseModel):
contexts: list[ExtractedContext]
# --- Modèles pour la Requête 4 (Barèmes) ---
class RubricItem(BaseModel):
label: str = Field(description="The exact label of the question.")
rubric_content: str = Field(description="Le barème détaillé en français.")
label: str = Field(description="Label exact de la question, à conserver sans traduction.")
rubric_content: str = Field(description="Barème détaillé sur 4 points : toutes les consignes, explications et justifications doivent être rédigées en français.")
class GroupRubrics(BaseModel):
rubrics: list[RubricItem]
PROMPT_4 = """Je te fournis les questions, le contexte éventuel, et les corrections pour un groupe de questions d'un examen.
Ta tâche :
Établir un barème de correction détaillé en français pour CHAQUE question.
Établir un barème de correction détaillé pour CHAQUE question.
Rédige intégralement en français le contenu de chaque champ `rubric_content`,
y compris les consignes de notation, les explications et les justifications,
même si certains textes fournis sont dans une autre langue.
Conserve les formules mathématiques, les labels exacts des questions et les
clés JSON `rubrics`, `label` et `rubric_content` sans les traduire.
Chaque question DOIT être notée sur exactement 4 points. Propose une répartition logique de ces points.
Par exemple :
- Au moins 2 points si le résultat est correct.
- Mettre la moitié des points si le raisonnement est correct mais pas le résultat.
- Retirer 1.5 points si les hypothèses d'un théorème ou d'une question précédente ne sont pas vérifiées.
- Retirer 1,5 point si les hypothèses d'un théorème ou d'une question précédente ne sont pas vérifiées.
Renvoie le résultat sous forme de liste JSON correspondant aux labels des questions fournies.
Renvoie uniquement un objet JSON contenant une liste `rubrics`. Pour chaque
question fournie, cette liste contient un objet avec son `label` exact et
son barème en français dans `rubric_content`.
"""
# --- Modèle fusionné (pour le reste du script) ---
@@ -102,50 +109,64 @@ class ExamExtraction(BaseModel):
class GroupedExamExtraction(BaseModel):
groups: list[list[QuestionItem | ContextItem]]
PROMPT_1 = """I am providing:
1. A PDF of an exam (`enonce.pdf`)
2. The source code of the exam questions (`enonce` file)
PROMPT_1 = """Je te fournis :
1. Le PDF d'un examen (`enonce.pdf`).
2. Le code source de ses questions (fichier `enonce`).
Your task:
1. Identify all distinct question labels using the PDF document.
These labels should be unique : use `Ex 1 : 1)a)` or `I)1)b)`.
2. For each label, extract its exact corresponding question text
from the `enonce` source file. Do not include the label itself
in this extracted text (nor LaTeX like `item` nor org-mode list
labelling like `2.`).
Return the result as a JSON list in the exact reading order of the document.
Ta tâche :
1. Identifie tous les labels distincts des questions à l'aide du PDF.
Ils doivent être uniques : utilise par exemple `Ex 1 : 1)a)` ou `I)1)b)`.
2. Pour chaque label, extrais exactement le texte de la question
correspondante dans le fichier source `enonce`. N'inclus ni le label
lui-même, ni les commandes de liste LaTeX comme `item`, ni les marques
de liste org-mode comme `2.`.
Ne reformule pas et ne traduis pas le texte extrait ; conserve le LaTeX.
Renvoie les questions dans l'ordre exact de lecture du document, dans la
liste `questions` de l'objet JSON attendu. Conserve les clés `label` et
`question_content`.
"""
PROMPT_2 = """I am providing:
1. A JSON list of question labels and their texts extracted from an exam.
2. The source code of the exam solutions (`correction` file).
PROMPT_2 = """Je te fournis :
1. Une liste JSON des labels des questions d'un examen et de leurs textes.
2. Le code source du corrigé de l'examen (fichier `correction`).
Your task:
For each question label provided in the JSON, extract its exact corresponding solution textual
content from the `correction` source file. Return the result as a JSON list in the exact same order.
Pour chaque label fourni, extrais exactement le texte de la solution
correspondante dans le fichier source `correction`. Ne reformule pas et
ne traduis pas le texte extrait ; conserve le LaTeX.
Renvoie les solutions dans le même ordre que les questions, dans la liste
`solutions` de l'objet JSON attendu. Conserve les clés `label` et
`solution_content` ainsi que les labels exacts des questions.
"""
PROMPT_3 = """I am providing:
1. A JSON list of question labels and their texts extracted from an exam.
2. The source code of the exam questions (`enonce` file).
PROMPT_3 = """Je te fournis :
1. Une liste JSON des labels des questions d'un examen et de leurs textes.
2. Le code source des questions de l'examen (fichier `enonce`).
Your task:
Extract important information necessary to understand the questions (e.g., definitions of objects, global notations, hypotheses, context) that are NOT part of the question texts themselves. Often, this information can be in a previous \\item that is not itself a question, but contains the question items.
Extrais les informations importantes nécessaires à la compréhension des
questions, mais qui ne font PAS partie des textes des questions :
définitions des objets, notations générales, hypothèses ou contexte.
Ces informations figurent souvent dans un \\item précédent qui ne constitue
pas lui-même une question, mais contient une liste de questions.
For example, given LaTeX code like
Par exemple, dans ce code LaTeX :
\\item Let N, M be two commutating matrices
\\item Soient N et M deux matrices qui commutent.
\\begin{itemize}
\\item Prove that N, M have a common eigenvector
\\item Prove that N, M are co-trigonalizable.
\\item Montrer que N et M ont un vecteur propre commun.
\\item Montrer que N et M sont simultanément trigonalisables.
\\end{itemize}
the `Let N, M be two commutating matrices` part is not a question itself, and is important information to understand the next two questions.
La phrase « Soient N et M deux matrices qui commutent » n'est pas une
question ; elle est nécessaire pour comprendre les deux questions suivantes.
For each extracted piece of information, identify:
1. The label of the FIRST question that comes immediately AFTER this information in the exam.
2. The label of the LAST question that uses or relies on this information.
Return the result as a JSON list.
Pour chaque information extraite, identifie :
1. Le label de la PREMIÈRE question située immédiatement APRÈS cette
information dans l'énoncé (`target_question_label`).
2. Le label de la DERNIÈRE question qui utilise cette information
(`last_question_label`).
Conserve le texte source dans `context_content`, sans le reformuler ni le
traduire, et conserve le LaTeX ainsi que les labels exacts.
Renvoie le résultat dans la liste `contexts` de l'objet JSON attendu.
"""
def find_file(folder: Path, base_name: str) -> Path | None:
@@ -245,7 +266,7 @@ def process_exam(
role="user",
parts=[
types.Part.from_text(text=PROMPT_2),
types.Part.from_text(text=f"--- EXTRACTED QUESTIONS ---\n{extracted_questions_json}"),
types.Part.from_text(text=f"--- QUESTIONS EXTRAITES ---\n{extracted_questions_json}"),
types.Part.from_text(text=f"--- CORRECTION SOURCE ({correction_path.name}) ---\n{correction_text}"),
],
)
@@ -281,7 +302,7 @@ def process_exam(
role="user",
parts=[
types.Part.from_text(text=PROMPT_3),
types.Part.from_text(text=f"--- EXTRACTED QUESTIONS ---\n{extracted_questions_json}"),
types.Part.from_text(text=f"--- QUESTIONS EXTRAITES ---\n{extracted_questions_json}"),
types.Part.from_text(text=f"--- ENONCE SOURCE ({enonce_path.name}) ---\n{enonce_text}"),
],
)
@@ -691,6 +712,7 @@ def process_exam(
]
config_4 = types.GenerateContentConfig(
system_instruction="Rédige tous les barèmes et consignes de notation en français. Conserve les clés JSON, les labels et les formules mathématiques.",
temperature=0.2,
response_mime_type="application/json",
response_json_schema=GroupRubrics.model_json_schema(),
+13 -3
View File
@@ -4,7 +4,6 @@ import sys
from collections.abc import Sequence
from pathlib import Path
from copienator.configuration import IMPORT_DIR
from copienator import (
EvaluationWorkspace,
ExitCode,
@@ -12,6 +11,7 @@ from copienator import (
execute,
workspace_from_args,
)
from copienator.configuration import IMPORT_DIR
ANNOTATION_DIRECTORIES = ("BGnot", "Bnot", "Anot")
@@ -23,7 +23,10 @@ def sync_annotated(
import_dir: Path,
) -> ExitCode:
workspace.require_directories(annotation_dir_name)
annotation_dir = workspace.root / annotation_dir_name
annotation_dir = workspace.annotation_dir("refaire") if annotation_dir_name == "BRnot" else workspace.root / annotation_dir_name
session_id = workspace.refaire_session_id if annotation_dir_name == "BRnot" else None
prefix = f"{session_id}__" if session_id else ""
accepted = 0
annotated_dir = Path(import_dir).expanduser()
if not annotated_dir.is_dir():
print(f"Error: directory does not exist: {annotated_dir}", file=sys.stderr)
@@ -39,7 +42,10 @@ def sync_annotated(
key=lambda path: path.name.casefold(),
)
for annotated_file in annotated_files:
target_subdir = annotation_dir / annotated_file.stem
if prefix and not annotated_file.stem.startswith(prefix):
print(f"Ignoring return from another pass: {annotated_file.name}")
continue
target_subdir = annotation_dir / annotated_file.stem.removeprefix(prefix)
if not target_subdir.is_dir():
print(f"Warning: directory not found: {target_subdir}", file=sys.stderr)
@@ -49,6 +55,10 @@ def sync_annotated(
dest_file = target_subdir / f"Concat_annotated{suffix}"
print(f"Copying {annotated_file} to {dest_file}")
shutil.copy2(annotated_file, dest_file)
accepted += 1
if prefix and not accepted:
print(f"No returns for the active pass {session_id} were imported.")
return ExitCode.PARTIAL
return ExitCode.PARTIAL if missing_targets else ExitCode.SUCCESS
@@ -9,8 +9,6 @@ from typing import Any
from PIL import Image, ImageDraw
from copienator.commands import annotating
from copienator import utils
from copienator import (
EvaluationWorkspace,
ExitCode,
@@ -18,16 +16,18 @@ from copienator import (
evaluation_parser,
execute,
read_json,
utils,
workspace_from_args,
)
from copienator.annotation_actions import apply_checkbox_actions, apply_score_overrides
from copienator.annotation_data import AnnotationData, RefaireList, load_annotation_data
from copienator.filesystem import staged_files
from copienator.commands import annotating
from copienator.commands.reading_annotations import (
concatenate,
detect_checks_and_notes,
has_significant_notes,
)
from copienator.filesystem import staged_files
LabelNotes = dict[str, dict[str, Any]]
ScanResult = tuple[dict[str, list[dict[str, Any]]], dict[str, LabelNotes]]
@@ -54,7 +54,9 @@ def get_extra_pdfs_as_images(
return images
def save_paginated_pdf(image_groups: list[list[Image.Image]], output_path: Path) -> None:
def save_paginated_pdf(
image_groups: list[list[Image.Image]], output_path: Path
) -> None:
"""Paginate vertically concatenated image groups and save them as a PDF."""
non_empty = [group for group in image_groups if group]
if not non_empty:
@@ -118,6 +120,8 @@ def _scan_annotation_directory(
directory: Path,
only_ids: set[str] | None = None,
default_student_id: str | None = None,
*,
required: bool = False,
) -> ScanResult:
bnote_path = directory / "bnote.json"
if not bnote_path.is_file():
@@ -133,6 +137,8 @@ def _scan_annotation_directory(
actions, notes_image = detect_checks_and_notes(directory)
if notes_image is None:
if required:
raise ValueError(f"Could not read annotations in {directory}")
return {}, {}
actions_by_student: dict[str, list[dict[str, Any]]] = defaultdict(list)
notes_by_student: dict[str, LabelNotes] = defaultdict(dict)
@@ -176,10 +182,12 @@ def apply_actions_and_regenerate_grouped(
all_labels: list[str],
*,
update_score: bool = False,
annotation_dir: str = "BGnot",
selected_labels: set[str] | None = None,
) -> tuple[ExitCode, str]:
"""Apply grouped annotations and atomically merge regenerated student files."""
"""Regenerate a copy, preserving reviewed images outside the redo selection."""
logs = [f"\nProcessing compilation for: Copie{student_id}"]
output_dir = workspace.annotation_dir("grouped") / f"Copie{student_id}"
output_dir = workspace.root / annotation_dir / f"Copie{student_id}"
labels_data = data.get(student_id, {})
dirty_labels = apply_checkbox_actions(labels_data, actions, logs.append)
if update_score:
@@ -187,15 +195,91 @@ def apply_actions_and_regenerate_grouped(
labels_data, output_dir / "score.json", logs.append
)
selected_labels = selected_labels if selected_labels is not None else set()
dirty_labels |= selected_labels
simple_layout = None
simple_annotated = None
if selected_labels and annotation_dir == "Anot":
imported = next(
(
output_dir / name
for name in ("Concat_annotated.jpg", "Concat_annotated.jpeg")
if (output_dir / name).is_file()
),
None,
)
if imported is not None:
layout_path = output_dir / "refaire_simple_layout.json"
if layout_path.is_file():
simple_layout = read_json(layout_path)
else:
simple_layout = {"images": {}, "replaced": []}
y = 0
for label in sorted(labels_data, key=utils.natural_key):
path = output_dir / f"{label}.jpg"
if (
path.is_file()
and labels_data[label]["result"].get("error") != "empty-answer"
):
with Image.open(path) as saved:
simple_layout["images"][label] = [y, y + saved.height]
y += saved.height
with Image.open(imported) as saved:
simple_annotated = saved.convert("RGB").copy()
expected_height = max(
(bounds[1] for bounds in simple_layout["images"].values()), default=0
)
if simple_annotated.height != expected_height:
raise ValueError(
"Imported simple image height does not match the original copy layout"
)
old_scores = (
read_json(output_dir / "score.json")
if selected_labels and (output_dir / "score.json").is_file()
else {}
)
scores = dict.fromkeys(all_labels, "")
dirty_images: dict[str, Image.Image] = {}
concat_images: list[Image.Image] = []
filtered_groups: list[list[Image.Image]] = []
incomplete = False
for label, content in sorted(labels_data.items(), key=lambda item: utils.natural_key(item[0])):
for label, content in sorted(
labels_data.items(), key=lambda item: utils.natural_key(item[0])
):
result = content["result"]
if (
selected_labels
and label not in selected_labels
and old_scores.get(label, "") != ""
):
result["score"] = old_scores[label]
scores[label] = str(result.get("score", 0))
saved_image = output_dir / f"{label}.jpg"
if selected_labels and label not in selected_labels and saved_image.is_file():
with Image.open(saved_image) as saved:
final_image = saved.convert("RGB").copy()
if (
simple_annotated is not None
and label in simple_layout["images"]
and label not in simple_layout["replaced"]
):
hmin, hmax = simple_layout["images"][label]
final_image = simple_annotated.crop(
(0, hmin, simple_annotated.width, hmax)
)
dirty_images[label] = final_image
scores[label] = str(old_scores.get(label, scores[label]))
concat_images.append(final_image)
# Keep previously reviewed content, including handwriting.
if annotation_dir == "BGnot":
extras = get_extra_pdfs_as_images(
workspace.root, label, annotating, all_labels
)
filtered_groups.append([*extras, final_image])
else:
filtered_groups.append([final_image])
continue
pdf_path = Path(content["pdf_path"])
if not pdf_path.is_file():
logs.append(f" Missing answer PDF: {pdf_path}")
@@ -221,13 +305,15 @@ def apply_actions_and_regenerate_grouped(
if has_notes:
width, height = sub_note.size
if old_header_height > 0:
header = sub_note.crop((0, 0, width, min(height, old_header_height)))
header = sub_note.crop(
(0, 0, width, min(height, old_header_height))
)
final_image.paste(header, (0, 0), mask=header)
if height > old_header_height:
body = sub_note.crop((0, old_header_height, width, height))
final_image.paste(body, (0, new_header_height), mask=body)
if label in dirty_labels or has_notes:
if label in dirty_labels or has_notes or selected_labels:
dirty_images[label] = final_image
logs.append(f" Saved dirty image: {label}.jpg")
concat_images.append(final_image)
@@ -237,33 +323,54 @@ def apply_actions_and_regenerate_grouped(
feedback.get("to_delete", False) for feedback in feedbacks
)
if not perfect or has_notes:
extras = get_extra_pdfs_as_images(
workspace.root, label, annotating, all_labels
extras = (
get_extra_pdfs_as_images(workspace.root, label, annotating, all_labels)
if annotation_dir == "BGnot"
else []
)
filtered_groups.append([*extras, final_image])
concat_image = concatenate(concat_images)
with staged_files(output_dir) as staging:
if incomplete:
return ExitCode.PARTIAL, "\n".join(logs)
with staged_files(output_dir, remove=("Concat_F.pdf", "Concat_F.jpg")) as staging:
if simple_layout is not None:
simple_layout["replaced"] = sorted(
set(simple_layout["replaced"]) | selected_labels
)
atomic_write_json(staging / "refaire_simple_layout.json", simple_layout)
for label, image in dirty_images.items():
image.save(staging / f"{label}.jpg")
atomic_write_json(staging / "score.json", scores)
if concat_image is not None:
concat_image.save(staging / "Concat.jpg")
if filtered_groups:
save_paginated_pdf(filtered_groups, staging / "Concat_F.pdf")
if annotation_dir == "BGnot":
save_paginated_pdf(filtered_groups, staging / "Concat_F.pdf")
else:
filtered_image = concatenate(
[image for group in filtered_groups for image in group]
)
filtered_image.save(staging / "Concat_F.jpg")
logs.append(f" Saved regenerated files in {output_dir}")
status = ExitCode.PARTIAL if incomplete else ExitCode.SUCCESS
return status, "\n".join(logs)
def _read_refaire(workspace: EvaluationWorkspace) -> tuple[RefaireList, dict[str, list[str]]]:
def _read_refaire(
workspace: EvaluationWorkspace,
) -> tuple[RefaireList, dict[str, list[str]]]:
loaded = read_json(workspace.refaire_file)
if not isinstance(loaded, list):
raise TypeError("refaire.json must contain a JSON array")
entries: RefaireList = []
by_student: dict[str, list[str]] = {}
for entry in loaded:
if not isinstance(entry, list) or len(entry) != 2 or not isinstance(entry[1], list):
if (
not isinstance(entry, list)
or len(entry) != 2
or not isinstance(entry[1], list)
):
raise TypeError(f"Malformed refaire entry: {entry!r}")
copy_name, labels = entry
student_id = str(copy_name).removeprefix("Copie")
@@ -273,14 +380,78 @@ def _read_refaire(workspace: EvaluationWorkspace) -> tuple[RefaireList, dict[str
return entries, by_student
def _scan_redo_annotations(
directory: Path,
expected: dict[str, set[str]],
) -> tuple[dict[str, list[dict[str, Any]]], dict[str, LabelNotes], set[str]]:
"""Read either grouped or per-copy redo PDFs using their student/label metadata."""
actions: dict[str, list[dict[str, Any]]] = defaultdict(list)
notes: dict[str, LabelNotes] = defaultdict(dict)
seen: dict[str, set[str]] = defaultdict(set)
incomplete: set[str] = set()
plans = []
required = ("checkboxes.json", "Reference.jpg", "Concat_annotated.pdf")
for path in sorted(directory.iterdir()):
if not path.is_dir():
continue
default_id = (
path.name.removeprefix("Copie") if path.name.startswith("Copie") else None
)
try:
metadata = read_json(path / "bnote.json")
pairs = [
(str(item.get("id", default_id)), str(item["label"]))
for item in metadata["images"]
]
except (OSError, ValueError, TypeError, KeyError) as exc:
print(f"Warning: unreadable redo metadata in {path}: {exc}")
incomplete.update(expected)
continue
students = {
student_id for student_id, _label in pairs if student_id in expected
}
if not students:
continue
for student_id, label in pairs:
if student_id not in expected:
continue
if label in seen[student_id]:
incomplete.add(student_id)
seen[student_id].add(label)
if any(not (path / name).is_file() for name in required):
print(f"Warning: missing returned redo inputs in {path}")
incomplete.update(students)
else:
plans.append((path, default_id, students))
for student_id, labels in expected.items():
if seen[student_id] != labels:
print(
f"Warning: redo labels do not match refaire.json for Copie{student_id}; regenerate BRnot"
)
incomplete.add(student_id)
for path, default_id, students in plans:
if students <= incomplete:
continue
try:
result = _scan_annotation_directory(
path, default_student_id=default_id, required=True
)
_merge_scan_result(actions, notes, result)
except (OSError, ValueError, TypeError) as exc:
print(f"Warning: could not read redo annotations in {path}: {exc}")
incomplete.update(students)
return dict(actions), dict(notes), incomplete
def run(
workspace: EvaluationWorkspace,
*,
refaire: bool = False,
update_score: bool = False,
annotation_dir: str = "BGnot",
) -> ExitCode:
workspace.require_files("labels", "correction.json")
workspace.require_directories("Copies", "Par label", "BGnot")
workspace.require_directories("Copies", "Par label", annotation_dir)
refaire_list: RefaireList | None = None
refaire_by_student: dict[str, list[str]] = {}
if refaire:
@@ -289,7 +460,12 @@ def run(
refaire_list, refaire_by_student = _read_refaire(workspace)
all_labels = utils.read_all_labels(workspace.root)
loaded = load_annotation_data(workspace, refaire_list=refaire_list)
loaded = load_annotation_data(workspace)
if refaire_list:
# Add explicitly requested answers without filtering out the rest of a copy.
selected_data = load_annotation_data(workspace, refaire_list=refaire_list)
for student_id, labels in selected_data.data.items():
loaded.data.setdefault(student_id, {}).update(labels)
for warning in loaded.warnings:
print(f"Warning: {warning}")
if not loaded.data:
@@ -301,8 +477,10 @@ def run(
only_ids = set(refaire_by_student) or None
group_dirs = [
path
for path in workspace.annotation_dir("grouped").iterdir()
if path.is_dir() and not path.name.startswith("Copie")
for path in (workspace.root / annotation_dir).iterdir()
if annotation_dir == "BGnot"
and path.is_dir()
and not path.name.startswith("Copie")
]
with concurrent.futures.ThreadPoolExecutor(max_workers=6) as executor:
futures = [
@@ -312,39 +490,58 @@ def run(
for future in concurrent.futures.as_completed(futures):
_merge_scan_result(actions_by_student, notes_by_student, future.result())
refaire_incomplete = False
if annotation_dir == "Bnot":
for student_id in refaire_by_student:
directory = workspace.root / annotation_dir / f"Copie{student_id}"
if directory.is_dir():
_merge_scan_result(
actions_by_student,
notes_by_student,
_scan_annotation_directory(
directory, default_student_id=student_id
),
)
skipped_students: set[str] = set()
if refaire:
for student_id, requested_labels in refaire_by_student.items():
selected = requested_labels or list(loaded.data.get(student_id, {}))
selected_set = set(selected)
directory = workspace.annotation_dir("refaire") / f"Copie{student_id}"
if not directory.is_dir():
print(f"Warning: missing refaire annotation directory {directory}")
refaire_incomplete = True
expected = {
student_id: set(labels or loaded.data.get(student_id, {}))
for student_id, labels in refaire_by_student.items()
}
redo_actions, redo_notes, skipped_students = _scan_redo_annotations(
workspace.annotation_dir("refaire"), expected
)
for student_id, selected in expected.items():
if student_id in skipped_students:
continue
actions_by_student[student_id] = [
action
for action in actions_by_student[student_id]
if str(action.get("label")) not in selected_set
if str(action.get("label")) not in selected
]
for label in selected:
notes_by_student[student_id].pop(label, None)
refaire_actions, refaire_notes = _scan_annotation_directory(
directory, default_student_id=student_id
actions_by_student[student_id].extend(
action
for action in redo_actions.get(student_id, [])
if str(action.get("label")) in selected
)
notes_by_student[student_id].update(
{
label: note
for label, note in redo_notes.get(student_id, {}).items()
if label in selected
}
)
for action in refaire_actions.get(student_id, []):
if str(action.get("label")) in selected_set:
actions_by_student[student_id].append(action)
for label, note in refaire_notes.get(student_id, {}).items():
if label in selected_set:
notes_by_student[student_id][label] = note
status = (
ExitCode.PARTIAL
if loaded.warnings or refaire_incomplete
else ExitCode.SUCCESS
ExitCode.PARTIAL if loaded.warnings or skipped_students else ExitCode.SUCCESS
)
student_ids = (
list(refaire_by_student)
if refaire
else sorted(loaded.data, key=utils.natural_key)
)
student_ids = list(refaire_by_student) if refaire else sorted(loaded.data, key=utils.natural_key)
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
futures = {
executor.submit(
@@ -356,9 +553,15 @@ def run(
notes_by_student[student_id],
all_labels,
update_score=update_score,
annotation_dir=annotation_dir,
selected_labels=(
set(refaire_by_student[student_id] or loaded.data[student_id])
if refaire
else None
),
): student_id
for student_id in student_ids
if student_id in loaded.data
if student_id in loaded.data and student_id not in skipped_students
}
for future in concurrent.futures.as_completed(futures):
result, output = future.result()
@@ -370,6 +573,12 @@ def run(
def build_parser() -> argparse.ArgumentParser:
parser = evaluation_parser("Read grouped annotations and regenerate copies")
parser.add_argument(
"--annotation-dir",
choices=("BGnot", "Bnot", "Anot"),
default="BGnot",
help="Original annotation directory for --refaire (default: BGnot)",
)
parser.add_argument(
"--refaire",
action="store_true",
@@ -387,10 +596,13 @@ def main(argv: Sequence[str] | None = None) -> int:
parser = build_parser()
def handle(args: argparse.Namespace) -> ExitCode:
if args.annotation_dir != "BGnot" and not args.refaire:
parser.error("--annotation-dir requires --refaire")
return run(
workspace_from_args(args),
refaire=args.refaire,
update_score=args.update_score,
annotation_dir=args.annotation_dir,
)
return execute(parser, argv, handle)
@@ -398,4 +610,3 @@ def main(argv: Sequence[str] | None = None) -> int:
if __name__ == "__main__":
raise SystemExit(main())
+120 -100
View File
@@ -2,89 +2,97 @@ from pathlib import Path
import io
from . import utils
main_prompt = """I'm giving you an image of several written answers to an exam.
main_prompt = """Je te fournis une image contenant plusieurs réponses manuscrites à un examen.
Each answer is separated by a black horizontal line, and underneath,
to the left, is indicated the ID of the answer, from `01` to `50`.
Chaque réponse est séparée de la précédente par une ligne horizontale noire.
En dessous de cette ligne, à gauche, figure l'identifiant de la réponse,
compris entre `01` et `50`.
I want you to score each answer, from 0 to 4, you may score half
points, such as 2.5. Even if a result is wrong, if the reasoning is
correct and could lead to a right answer, you should give at least
half the points.
Attribue à chaque réponse une note de 0 à 4. Les demi-points sont autorisés,
par exemple 2.5. Même si le résultat est faux, accorde au moins la moitié
des points si le raisonnement est correct et pourrait conduire au bon résultat.
You also need to give feedback to the student, in french :
- which part of his answer is wrong,
- why is it wrong
- possibly, what he should have done instead.
Your feedback may contain LaTeX fragments written like `$a^2 + b^2 = c^2$`.
Rédige tous les commentaires destinés à l'élève en français. Indique :
- quelle partie de sa réponse est fausse ;
- pourquoi elle est fausse ;
- éventuellement, ce qu'il aurait fallu faire à la place.
Les commentaires peuvent contenir des fragments LaTeX, par exemple
`$a^2 + b^2 = c^2$`.
If your score is not 4, you should always provide some feedback
explaining what's missing.
Si la note n'est pas 4, fournis toujours un commentaire expliquant ce qui
manque, sauf dans le cas `empty-answer` décrit ci-dessous.
For each piece of feedback, if it is related to a specific part of the
answer that is wrong, you may provide a `box_2d`, to locate this
specific part of the answer. This `box_2d` should be in the form
[ymin, xmin, ymax, xmax] normalized to 0-1000. If you do not provide
one, set `box_2d` to `null`.
Lorsqu'un commentaire concerne une erreur située dans une partie précise
de la réponse, tu peux fournir un champ `box_2d` pour la localiser.
Ses coordonnées doivent être de la forme [ymin, xmin, ymax, xmax],
normalisées entre 0 et 1000. Sinon, attribue la valeur `null` à `box_2d`.
If the answer is correct, there is no need to provide feedback. You do
not have to give positive feedback, but if you do, do not provide a
`box_2d` for it.
Si la réponse est correcte, aucun commentaire n'est nécessaire. Tu n'es pas
obligé de faire des commentaires positifs ; si tu en fais, ne leur associe
pas de `box_2d`.
For example, if the student says a function is continuous when it
isn't, provide the coordinates where the word «continuous» is. If a
calculation went wrong, gives the coordinates of the step where it
goes wrong, and as feedback, what went wrong.
Par exemple, si l'élève affirme à tort qu'une fonction est continue,
localise le mot « continue ». Si un calcul est faux, localise l'étape où
l'erreur apparaît et explique cette erreur dans le commentaire.
Avoid giving feedback about confusing letters `n` with `m`, `x` with
`n` or `h` with `k`. If it looks wrong, assume you read it wrong,
unless the distinction is very important.
Évite les commentaires portant sur une confusion entre les lettres `n`
et `m`, `x` et `n`, ou `h` et `k`. En cas de doute, suppose que tu as mal
lu, sauf si la distinction est très importante.
In some case, you may find that either
- The student didn't answer the right question. Set the score to 0.
Since it could be a labeling error, indicate it by setting `error`
to \"wrong-label\".
- You can find an answer to another question of the exercice (taking
more than a couple of lines). Score the question you are supposed
to score, but set `error` to \"additional-answer\".
- The answer to the question is empty, or the student has only
rewritten the statement of the question. In this case, set `error`
to \"empty-answer\" and do not provide any kind of feedback.
If there's no error, set `error` to `\"\"`.
Certains cas nécessitent une valeur particulière du champ `error` :
- L'élève n'a pas répondu à la bonne question : attribue la note 0 et
indique `wrong-label`, car il peut s'agir d'une erreur de label.
- La réponse contient aussi une réponse à une autre question de
l'exercice, sur plus de quelques lignes : note la question demandée,
mais indique `additional-answer`.
- La réponse est vide, ou l'élève a seulement recopié l'énoncé : indique
`empty-answer` et ne fournis aucun commentaire.
S'il n'y a aucune de ces erreurs, attribue la chaîne vide `""` à `error`.
You will answer using json describing a list of dictionary with a key
\"id\", and a key \"result\" that contains the \"score\", a list
\"feedback\", and possibly an \"error\". Like this example :
Réponds uniquement en JSON, sous la forme d'une liste d'objets contenant
les clés `id` et `result`. L'objet `result` contient `score`, la liste
`feedback` et `error`. Chaque commentaire contient `text` et `box_2d`.
Conserve exactement ces clés, les identifiants et les valeurs techniques
de `error` : ne les traduis pas. Le contenu de chaque champ `text` doit
être en français, même si certains documents fournis sont dans une autre langue.
[{ \"id\": \"01\",
\"result\": {\"score\" : 2.5,
\"feedback\": [{text: \"Un retour générique. Il faut apprendre le cours.\", box_2d: null},
{text: \"Non, la fonction n'est pas forcément continue\", pos: [145, 280, 340, 500]}],
\"error\": \"\"}
},
{ \"id\": \"04\",
\"result\": {\"score\" : 4.,
\"feedback\" : []
\"error\": \"\" }
}
Exemple :
```json
[
{
"id": "01",
"result": {
"score": 2.5,
"feedback": [
{"text": "Il manque la vérification des hypothèses du théorème.", "box_2d": null},
{"text": "Non, la fonction n'est pas forcément continue.", "box_2d": [145, 280, 340, 500]}
],
"error": ""
}
},
{
"id": "04",
"result": {"score": 4.0, "feedback": [], "error": ""}
}
]
```
Here is the text of the exercice (or the relevant part of the problem)
of the exam :
Voici l'énoncé de l'exercice ou la partie pertinente du problème :
```
<<text>>
```
Here is a possible correct answer :
Voici un corrigé possible :
```
<<corr>>
```
<<persp>>
You are asked to score the question or exercice labeled `<<label>>`,
do not score or give feedback to any other question."""
Tu dois noter uniquement la question ou l'exercice portant le label
`<<label>>`. Ne note aucune autre question et ne formule aucun commentaire
sur les autres questions."""
from .utils import get_label_text_content, get_label_sol_content, get_label_persp_content
@@ -95,7 +103,7 @@ def make_prompt(input_dir,full_label):
# print("Debug : l/t/c/p", full_label, text, corr, persp)
if persp:
persp = "\n\nHere are additional scoring instructions : \n\n```\n" + persp +"\n```\n"
persp = "\n\nVoici des consignes de notation complémentaires : \n\n```\n" + persp +"\n```\n"
return main_prompt.replace("<<text>>", text).replace("<<corr>>", corr).replace("<<persp>>", persp).replace("<<label>>", full_label)
@@ -103,17 +111,17 @@ from pydantic import BaseModel, Field, TypeAdapter
from typing import List, Optional, Tuple
class FeedbackItem(BaseModel):
text: str = Field(description="Feedback content")
box_2d: Optional[List[int]] = Field(None, description="box coordinates or null")
text: str = Field(description="Commentaire destiné à l’élève, rédigé en français.")
box_2d: Optional[List[int]] = Field(None, description="Coordonnées [ymin, xmin, ymax, xmax] normalisées entre 0 et 1000, ou null.")
class ResultData(BaseModel):
score: float = Field(description="The numeric score")
feedback: List[FeedbackItem] = Field(description="List of feedback items")
error: str = Field(description="Indicates if an error occurred")
score: float = Field(description="Note numérique de la réponse, sur 4 points.")
feedback: List[FeedbackItem] = Field(description="Liste des commentaires destinés à l’élève, rédigés en français.")
error: str = Field(description="Type derreur : wrong-label, additional-answer, empty-answer, ou chaîne vide.")
class EvaluationEntry(BaseModel):
id: str = Field(description="Entry identifier")
result: ResultData = Field(description="Result details")
id: str = Field(description="Identifiant exact de la réponse.")
result: ResultData = Field(description="Note, commentaires en français et éventuelle erreur pour cette réponse.")
# These nested definitions do not work with the batch api, unroll them
UNROLLED_SCHEMA = {
@@ -121,24 +129,24 @@ UNROLLED_SCHEMA = {
"items": {
"type": "OBJECT",
"properties": {
"id": {"type": "STRING", "description": "Entry identifier"},
"id": {"type": "STRING", "description": "Identifiant exact de la réponse."},
"result": {
"type": "OBJECT",
"properties": {
"score": {"type": "NUMBER", "description": "The numeric score"},
"error": {"type": "STRING", "description": "Indicates if an error occurred"},
"score": {"type": "NUMBER", "description": "Note numérique de la réponse, sur 4 points."},
"error": {"type": "STRING", "description": "Type derreur : wrong-label, additional-answer, empty-answer, ou chaîne vide."},
"feedback": {
"type": "ARRAY",
"description": "List of feedback items",
"description": "Liste des commentaires destinés à l’élève, rédigés en français.",
"items": {
"type": "OBJECT",
"properties": {
"text": {"type": "STRING", "description": "Feedback content"},
"text": {"type": "STRING", "description": "Commentaire destiné à l’élève, rédigé en français."},
"box_2d": {
"type": "ARRAY",
"items": {"type": "INTEGER"},
"nullable": True,
"description": "box coordinates or null"
"description": "Coordonnées [ymin, xmin, ymax, xmax] normalisées entre 0 et 1000, ou null."
}
},
"required": ["text"]
@@ -216,20 +224,27 @@ def request_for_box_correction(pdf_path, original_feedbacks):
localized_feedbacks = [f for f in original_feedbacks if f["box_2d"]]
prompt = f"""
Here is a single student's submission to a question in a written exam. The following JSON contains feedback items with bounding boxes (box_2d) that are incorrect. Each piece of feedback is supposed to be related to a piece of the answer that is wrong.
prompt = f"""Voici la réponse d'un élève à une question d'examen. Le JSON
ci-dessous contient des commentaires dont les rectangles de localisation
(`box_2d`) sont incorrects. Chaque commentaire doit correspondre à la
partie de la réponse se trouve l'erreur signalée.
For example, if the student says a function is continuous when it
isn't, the coordinates should be where the word «continuous» is. If a
calculation went wrong, the coordinates should be where the step where
it goes wrong, and the feedback is what went wrong.
Par exemple, si l'élève affirme à tort qu'une fonction est continue,
les coordonnées doivent localiser le mot « continue ». Si un calcul est
faux, elles doivent localiser l'étape où apparaît l'erreur expliquée dans
le commentaire.
Please analyze the image and return the same feedback json content, but with ONLY the box_2d coordinates corrected for this specific image.
Coordinates must be [ymin, xmin, ymax, xmax] scaled to 1000. If a box is invalid/not found, return null for it.
Original feedback:
Analyse l'image et renvoie le même contenu JSON en corrigeant UNIQUEMENT
les coordonnées `box_2d` pour cette image. Conserve les commentaires en
français à l'identique : ne les reformule pas et ne les traduis pas.
Conserve les noms des clés JSON.
Les coordonnées doivent être [ymin, xmin, ymax, xmax], normalisées entre
0 et 1000. Si la zone est introuvable ou le rectangle invalide, renvoie
`null` pour ce rectangle.
{json.dumps(localized_feedbacks, indent=2)}
"""
Commentaires d'origine :
{json.dumps(localized_feedbacks, indent=2, ensure_ascii=False)}
"""
@@ -252,20 +267,21 @@ it goes wrong, and the feedback is what went wrong.
def request_for_wrong_label(pdf_path, label, enonce, labels_txt):
prompt = f"""This image is a part of the answer of a student to a written exam.
prompt = f"""Cette image représente une partie de la réponse d'un élève à un examen.
It was initially labeled '{label}' but I suspect this label is wrong. Perhaps the student himself wrote the wrong label.
Elle porte initialement le label '{label}', mais je soupçonne une erreur
de label. L'élève a peut-être lui-même écrit le mauvais label.
You need to analyse this image, and find the label of the question it answers. Do not trust the label written by the student but instead check the content of its answer and the notation he uses to identify the correct label of the question the student answered.
Analyse l'image et identifie le label de la question à laquelle cette
réponse correspond. Ne te fie pas au label écrit par l'élève : examine
le contenu de la réponse et les notations utilisées.
Return ONLY the exact label string.
Here is the full content of the exam :
Renvoie UNIQUEMENT le label exact, sans le modifier ni le traduire.
Voici l'énoncé complet de l'examen :
{enonce}
Here is a list of all possible labels. You need to answer with one of these :
Voici les labels possibles. Ta réponse doit être l'un d'entre eux :
{labels_txt}
"""
@@ -276,22 +292,26 @@ Here is a list of all possible labels. You need to answer with one of these :
return contents, config
def request_for_additional_answer(pdf_path, label, enonce, labels_txt):
prompt = f"""This image is a part of the answer of a student to a written exam.
prompt = f"""Cette image représente une partie de la réponse d'un élève à un examen.
It was initially labeled '{label}' but I suspect this image also contains answers to another, or several other questions.
Elle porte initialement le label '{label}', mais je soupçonne qu'elle
contient aussi des réponses à une ou plusieurs autres questions.
You need to analyse this image, and find the list of the labels of the questions it answers. Return ONLY the list of the exact label strings.
Analyse l'image et identifie les labels des questions auxquelles elle
répond. Renvoie UNIQUEMENT une liste JSON contenant les labels exacts,
sans les modifier ni les traduire.
If the end of the image only contains the first line of an answer to another question, ignore it.
Here is the full content of the exam :
Si le bas de l'image ne contient que la première ligne d'une réponse à
une autre question, ignore cette ligne.
Voici l'énoncé complet de l'examen :
{enonce}
Here is a list of all possible labels. You need to answer with a list one of these :
Voici les labels possibles. Chaque élément de ta liste doit être l'un
d'entre eux :
{labels_txt}
"""
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)
+20 -1
View File
@@ -165,7 +165,26 @@ class EvaluationWorkspace:
def return_dir(self) -> Path:
return self.root / "A Rendre"
@property
def refaire_session_id(self) -> str | None:
from .json_io import read_json
session = read_json(self.root / "refaire-session.json", default=None)
if session is None:
return None
ident = session.get("id") if isinstance(session, dict) else None
if not isinstance(ident, str) or not re.fullmatch(r"reprise-[0-9]{8}-[0-9]{6}-[a-f0-9]{8}", ident):
raise ValueError("Invalid refaire-session.json")
return ident
@property
def refaire_session_dir(self) -> Path | None:
ident = self.refaire_session_id
return self.root / "Reprises" / ident if ident else None
def annotation_dir(self, mode: str) -> Path:
if mode == "refaire" and self.refaire_session_dir is not None:
return self.refaire_session_dir / "BRnot"
directories = {
"simple": "Anot",
"checks": "Bnot",
@@ -215,7 +234,7 @@ class EvaluationWorkspace:
missing = [
relative_path
for relative_path in relative_paths
if not (self.root / relative_path).is_dir()
if not (self.annotation_dir("refaire") if relative_path == "BRnot" else self.root / relative_path).is_dir()
]
if missing:
raise WorkspaceValidationError(self.root, missing)
+244 -25
View File
@@ -3,11 +3,12 @@ from __future__ import annotations
import os
import queue
import tkinter as tk
from dataclasses import replace
from pathlib import Path
from tkinter import filedialog, messagebox, ttk
from typing import Any
from copienator import ExitCode
from copienator import EvaluationWorkspace, ExitCode, atomic_write_json
from copienator.platform import (
WindowsLabelError,
add_platform_executable_paths,
@@ -16,6 +17,15 @@ from copienator.platform import (
)
from .diagnostics import collect_diagnostics
from .refaire import SECTION as REFAIRE_SECTION
from .refaire import (
RefaireSelection,
available_copies,
load_selection,
resolve_layout,
validate_selection,
)
from .refaire_sessions import RESTART_WARNING, begin_pass
from .runner import ProcessRunner
from .state import StateStore
from .workflow import (
@@ -154,6 +164,9 @@ class CopienatorApp(tk.Tk):
self.arg_vars: dict[str, tk.Variable] = {}
self.copy_paths: dict[str, Path] = {}
self._rendering = False
self.refaire_panel: RefaireSelection | None = None
self.pending_refaire_commands: list[list[str]] = []
self.refaire_command_index = 0
self.title("Copienator — assistant de correction")
self.geometry("1180x820")
@@ -245,8 +258,19 @@ class CopienatorApp(tk.Tk):
self.title_label.grid(row=0, column=0, sticky="w")
self.description_label = ttk.Label(self.detail, text="", wraplength=680, justify="left")
self.description_label.grid(row=1, column=0, sticky="ew", pady=(5, 8))
self.form = ttk.Frame(self.detail)
self.form.grid(row=2, column=0, sticky="nsew")
form_container = ttk.Frame(self.detail)
form_container.grid(row=2, column=0, sticky="nsew")
form_container.columnconfigure(0, weight=1)
form_container.rowconfigure(0, weight=1)
self.form_canvas = tk.Canvas(form_container, highlightthickness=0, width=1, height=1)
self.form_canvas.grid(row=0, column=0, sticky="nsew")
form_scroll = ttk.Scrollbar(form_container, orient="vertical", command=self.form_canvas.yview)
form_scroll.grid(row=0, column=1, sticky="ns")
self.form_canvas.configure(yscrollcommand=form_scroll.set)
self.form = ttk.Frame(self.form_canvas)
form_window = self.form_canvas.create_window((0, 0), window=self.form, anchor="nw")
self.form.bind("<Configure>", lambda _event: self.form_canvas.configure(scrollregion=self.form_canvas.bbox("all")))
self.form_canvas.bind("<Configure>", lambda event: self.form_canvas.itemconfigure(form_window, width=event.width))
self.form.columnconfigure(1, weight=1)
command_box = ttk.LabelFrame(self.detail, text="Commande", padding=6)
@@ -362,6 +386,9 @@ class CopienatorApp(tk.Tk):
self.proxy_entry.configure(state="normal" if self.use_proxy_var.get() else "disabled")
def _load_evaluation(self) -> None:
if self.active_step_id or self.runner.running:
messagebox.showinfo("Traitement en cours", "Attendez la fin du traitement avant de changer d’évaluation.")
return
evaluation = self.evaluation
if not evaluation or not evaluation.is_dir():
messagebox.showerror("Dossier invalide", "Choisissez un dossier d’évaluation existant.")
@@ -392,13 +419,14 @@ class CopienatorApp(tk.Tk):
def _populate_tree(self) -> None:
selected = self.current_step.id if self.current_step else None
opened = {self.tree.item(item, "text"): self.tree.item(item, "open") for item in self.tree.get_children()}
self.tree.delete(*self.tree.get_children())
section_items: dict[str, str] = {}
for step in self.steps:
if step.section not in section_items:
section_id = f"section:{len(section_items)}"
section_items[step.section] = section_id
self.tree.insert("", "end", iid=section_id, text=step.section, values=("",), open=True)
self.tree.insert("", "end", iid=section_id, text=step.section, values=("",), open=opened.get(step.section, step.section != REFAIRE_SECTION))
status = self._step_status(step)
suffix = " (facultative)" if step.optional else ""
self.tree.insert(
@@ -443,7 +471,8 @@ class CopienatorApp(tk.Tk):
if any(char in pattern for char in "*?["):
exists = next(evaluation.glob(pattern), None) is not None
else:
exists = (evaluation / pattern).exists()
path = EvaluationWorkspace(evaluation).annotation_dir("refaire") if pattern == "BRnot" else evaluation / pattern
exists = path.exists()
if not exists:
missing.append(pattern)
return missing
@@ -513,9 +542,16 @@ class CopienatorApp(tk.Tk):
if not step:
return
self._rendering = True
redo = step.section == REFAIRE_SECTION
self.console.configure(height=8 if redo else 14)
self.rowconfigure(1, weight=4 if redo else 3)
self.rowconfigure(2, weight=1 if redo else 2)
self.form_canvas.configure(height=(340 if step.id == "refaire_selection" else 260) if redo else 200)
self.form_canvas.yview_moveto(0)
for child in self.form.winfo_children():
child.destroy()
self.arg_vars.clear()
self.refaire_panel = None
entry = self.state_store.step(step.id) if self.state_store.evaluation else {}
saved_variant = entry.get("variant", step.variants[0].id)
if saved_variant not in {variant.id for variant in step.variants}:
@@ -544,6 +580,8 @@ class CopienatorApp(tk.Tk):
variant = self._current_variant()
evaluation_arg = self._evaluation_arg()
for spec in step.arguments:
if step.section == REFAIRE_SECTION:
continue
if spec.variants and variant.id not in spec.variants:
continue
value = values.get(spec.name, value_for_default(spec.default, evaluation_arg))
@@ -582,27 +620,45 @@ class CopienatorApp(tk.Tk):
row += 1
row += 1
if not step.is_manual:
if not step.is_manual and step.section != REFAIRE_SECTION:
ttk.Label(self.form, text="Arguments supplémentaires").grid(
row=row, column=0, sticky="w", pady=(10, 4), padx=(0, 8)
)
ttk.Entry(self.form, textvariable=self.extra_var).grid(row=row, column=1, sticky="ew", pady=(10, 4))
self.run_button.configure(text="Marquer terminée" if step.is_manual else "Exécuter")
self.run_button.configure(text="Enregistrer la sélection" if step.id == "refaire_selection" else ("Marquer terminée" if step.is_manual else "Exécuter"))
self.skip_button.configure(state="normal" if step.optional else "disabled")
self._rendering = False
self._update_command_preview()
self._update_controls()
def _render_context_controls(self, step: StepDefinition, row: int) -> int:
if step.id == "review_persp":
ttk.Button(
self.form,
text="Ouvrir le dossier Persp",
command=self._open_persp,
).grid(row=row, column=0, columnspan=2, sticky="w", pady=(0, 8))
row += 1
if step.section == REFAIRE_SECTION:
evaluation = self.evaluation
if step.id in {"refaire_selection", "refaire_correct"}:
row = self._correction_folder_buttons(row)
row = self._refaire_restart_controls(row)
if step.id == "refaire_selection" and evaluation:
choices, preferred = self._annotation_directory_choices("export")
draft = self.state_store.step(step.id).get("values", {})
self.refaire_panel = RefaireSelection(self.form, evaluation, draft, choices, preferred)
self.refaire_panel.grid(row=row, column=0, columnspan=2, sticky="nsew")
else:
try:
entries = self._refaire_selection()
summary = f"{len(entries)} copie(s) sélectionnée(s) :\n" + " ; ".join(f"{name} : {', '.join(labels) or 'toute la copie'}" for name, labels in entries[:5])
if len(entries) > 5:
summary += f" ; et {len(entries) - 5} autres (voir la sélection)."
source = self.state_store.step("refaire_selection").get("values", {}).get("annotation_dir", "")
choice = self.state_store.step("refaire_selection").get("values", {}).get("layout", "auto")
layout = resolve_layout(entries, EvaluationWorkspace(evaluation).read_labels(), choice)
summary += f"\nPassage principal : {source} — PDF à vérifier : {'par question' if layout == 'grouped' else 'par copie'}"
except (OSError, ValueError, TypeError) as exc:
summary = str(exc)
ttk.Label(self.form, text=summary, wraplength=600, justify="left").grid(row=row, column=0, columnspan=2, sticky="w")
return row + 1
if step.id in {"review_persp", "correction"}:
row = self._correction_folder_buttons(row)
if step.section == "Prétraitement des copies":
evaluation = self.evaluation
paths = copy_pdf_paths(evaluation) if evaluation else []
@@ -646,6 +702,58 @@ class CopienatorApp(tk.Tk):
row += 1
return row
def _refaire_restart_controls(self, row: int) -> int:
controls = ttk.Frame(self.form)
controls.grid(row=row, column=0, columnspan=2, sticky="w", pady=(0, 5))
running = bool(self.active_step_id) or self.runner.running
ttk.Button(controls, text="Nouvelle reprise", command=lambda: self._start_refaire_pass(False),
state="disabled" if running else "normal").pack(side="left", padx=(0, 6))
ttk.Button(controls, text="Refaire la même sélection", command=lambda: self._start_refaire_pass(True),
state="disabled" if running else "normal").pack(side="left")
ttk.Label(self.form, text="« Nouvelle reprise » : à utiliser après avoir importé les résultats et mis à jour les copies finales.",
wraplength=580, justify="left").grid(row=row + 1, column=0, columnspan=2, sticky="w", pady=(0, 6))
return row + 2
def _start_refaire_pass(self, keep_selection: bool) -> None:
workspace = self.state_store.workspace
if workspace is None:
messagebox.showerror("Évaluation absente", "Chargez dabord une évaluation.")
return
if self.active_step_id or self.runner.running:
messagebox.showwarning("Traitement en cours", "Attendez la fin du traitement avant de commencer une reprise.")
return
try:
if keep_selection:
load_selection(workspace.root)
title = "Refaire la même sélection" if keep_selection else "Nouvelle reprise"
message = RESTART_WARNING
if self.state_store.step("refaire_merge").get("status") != "success":
message += "\n\nLa reprise actuelle nest pas marquée comme fusionnée. Les résultats non fusionnés ne seront pas intégrés aux copies finales."
message += "\n\nLes fichiers de la reprise précédente seront conservés. Continuer ?"
if not messagebox.askyesno(title, message, icon="warning", default="no"):
return
self._save_current_form()
updated, ident = begin_pass(workspace, self.state_store.data, keep_selection=keep_selection)
self.state_store.data = updated
# Do not save the old form over the fresh pass during navigation.
self.current_step = None
self.refaire_panel = None
self.arg_vars.clear()
self._populate_tree()
self.tree.selection_set("refaire_selection")
self.tree.see("refaire_selection")
self.info_var.set(f"Nouvelle reprise : {ident}. Enregistrez la sélection pour continuer.")
except (OSError, ValueError, TypeError) as exc:
messagebox.showerror("Nouvelle reprise impossible", str(exc))
def _correction_folder_buttons(self, row: int) -> int:
buttons = ttk.Frame(self.form)
buttons.grid(row=row, column=0, columnspan=2, sticky="w", pady=(0, 6))
for directory, label in (("Sol", "Corrigés (Sol)"), ("Persp", "Consignes de notation (Persp)")):
ttk.Button(buttons, text=label,
command=lambda name=directory: self._open_desktop_path(self.evaluation / name if self.evaluation else None, f"dossier {name}")).pack(side="left", padx=(0, 6))
return row + 1
def _annotation_directory_choices(self, step_id: str) -> tuple[tuple[str, ...], str]:
evaluation = self.evaluation
detected = detected_annotation_directories(evaluation) if evaluation else ()
@@ -696,6 +804,10 @@ class CopienatorApp(tk.Tk):
return next((variant for variant in self.current_step.variants if variant.id == selected), self.current_step.variants[0])
def _values(self) -> dict[str, object]:
if self.current_step and self.current_step.section == REFAIRE_SECTION:
if self.current_step.id == "refaire_selection" and self.refaire_panel:
return self.refaire_panel.values()
return {"target": self._evaluation_arg(), "annotation_dir": self.state_store.step("refaire_selection").get("values", {}).get("annotation_dir", "BGnot")}
return {name: variable.get() for name, variable in self.arg_vars.items()}
def _save_current_form(self) -> None:
@@ -715,9 +827,74 @@ class CopienatorApp(tk.Tk):
evaluation = self.evaluation
return evaluation_argument(self.repository, evaluation) if evaluation else "<évaluation>"
def _progression_ids(self, step_id: str) -> list[str]:
redo = step_id.startswith("refaire_")
return [step.id for step in self.steps if step.id.startswith("refaire_") == redo]
def _refaire_selection(self) -> list[list]:
evaluation = self.evaluation
if not evaluation or not self.state_store.evaluation:
raise ValueError("Chargez dabord une évaluation.")
entry = self.state_store.step("refaire_selection")
if entry.get("status") != "success":
raise ValueError("Enregistrez dabord les copies et les questions à refaire.")
values = entry.get("values", {})
selection = load_selection(evaluation)
if selection != validate_selection(values.get("selection"), available_copies(evaluation), EvaluationWorkspace(evaluation).read_labels()):
raise ValueError("La sélection a changé. Enregistrez-la avant de continuer.")
source = values.get("annotation_dir")
if source not in ANNOTATION_DIRECTORIES or not (evaluation / source).is_dir():
raise ValueError("Choisissez un dossier existant pour le passage principal.")
return selection
def _save_refaire_selection(self) -> None:
if not self.refaire_panel or not self.evaluation:
return
try:
if self.state_store.step("refaire_merge").get("status") == "success":
messagebox.showinfo("Reprise terminée", "Utilisez « Nouvelle reprise » ou « Refaire la même sélection » pour commencer un autre passage.")
return
values = self.refaire_panel.values()
entries = validate_selection(values["selection"], available_copies(self.evaluation), EvaluationWorkspace(self.evaluation).read_labels())
source = values["annotation_dir"]
if source not in detected_annotation_directories(self.evaluation):
raise ValueError("Choisissez un dossier existant pour le passage principal.")
workspace = EvaluationWorkspace(self.evaluation)
output = workspace.annotation_dir("refaire")
if output.is_dir() and any(output.iterdir()) and workspace.refaire_file.is_file():
from copienator import read_json
if read_json(workspace.refaire_file) != entries:
raise ValueError("Des PDF de vérification existent déjà. Importez et fusionnez leurs résultats, puis utilisez « Nouvelle reprise » pour changer de sélection.")
if workspace.refaire_session_dir is not None:
atomic_write_json(workspace.refaire_session_dir / "refaire.json", entries)
atomic_write_json(workspace.refaire_session_dir / "session.json", {"id": workspace.refaire_session_id, "values": values})
atomic_write_json(self.evaluation / "refaire.json", entries)
self._mark_step("success")
self._move_selection_from("refaire_selection", 1)
except (OSError, ValueError, TypeError) as exc:
messagebox.showerror("Sélection à vérifier", str(exc))
def _refaire_commands(self) -> list[list[str]]:
assert self.current_step is not None
selection = self._refaire_selection()
values = self._values()
targets = [self._evaluation_arg()]
if self.current_step.id in {"refaire_review", "refaire_split"}:
copies = available_copies(self.evaluation)
targets = [str(copies[name]) for name, _labels in selection]
variant = self._current_variant()
if self.current_step.id == "refaire_annotate":
choice = self.state_store.step("refaire_selection").get("values", {}).get("layout", "auto")
layout = resolve_layout(selection, EvaluationWorkspace(self.evaluation).read_labels(), choice)
variant = replace(variant, program="annotate-grouped" if layout == "grouped" else "annotate-checks")
return [build_command(self.repository, self.current_step, variant,
{**values, "target": target}, self._evaluation_arg()) for target in targets]
def _make_command(self) -> list[str]:
if not self.current_step:
return []
if self.current_step.section == REFAIRE_SECTION:
return self._refaire_commands()[0]
return build_command(
self.repository,
self.current_step,
@@ -731,11 +908,15 @@ class CopienatorApp(tk.Tk):
if self._rendering or not self.current_step:
return
if self.current_step.is_manual:
self.command_var.set("Étape manuelle — aucune commande ne sera exécutée.")
self.command_var.set("La sélection sera utilisée pour toutes les étapes de ce parcours." if self.current_step.id == "refaire_selection" else "Étape manuelle — aucune commande ne sera exécutée.")
return
try:
self.command_var.set(command_display(self._make_command()))
except ValueError as exc:
commands = self._refaire_commands() if self.current_step.section == REFAIRE_SECTION else [self._make_command()]
preview = "\n".join(command_display(command) for command in commands[:2])
if len(commands) > 2:
preview += f"\nPuis {len(commands) - 2} autres copies, successivement."
self.command_var.set(preview)
except (OSError, ValueError, TypeError) as exc:
self.command_var.set(f"Arguments invalides : {exc}")
def _browse_target_file(self, variable: tk.Variable) -> None:
@@ -764,7 +945,7 @@ class CopienatorApp(tk.Tk):
return False
try:
self._make_command()
except ValueError as exc:
except (OSError, ValueError, TypeError) as exc:
messagebox.showerror("Arguments invalides", str(exc))
return False
return True
@@ -775,11 +956,22 @@ class CopienatorApp(tk.Tk):
if not step or not evaluation or not self.state_store.evaluation:
messagebox.showerror("Évaluation absente", "Chargez dabord un dossier d’évaluation.")
return
if self.runner.running:
if self.active_step_id or self.runner.running:
messagebox.showwarning("Traitement en cours", "Interrompez le traitement actuel avant den lancer un autre.")
return
if step.id == "refaire_selection":
self._save_refaire_selection()
return
if step.section == REFAIRE_SECTION:
try:
self._refaire_selection()
except (OSError, ValueError, TypeError) as exc:
messagebox.showerror("Sélection à vérifier", str(exc))
return
if step.is_manual:
self._mark_step("success")
if step.section == REFAIRE_SECTION:
self._move_selection_from(step.id, 1)
return
if os.name == "nt" and step.id != "statement":
labels_path = evaluation / "labels"
@@ -816,7 +1008,7 @@ class CopienatorApp(tk.Tk):
self._save_current_form()
run_values = self._values()
ordered_ids = [item.id for item in self.steps]
ordered_ids = self._progression_ids(step.id)
self.state_store.invalidate_after(ordered_ids, step.id)
self.state_store.update_step(
step.id,
@@ -837,6 +1029,8 @@ class CopienatorApp(tk.Tk):
self.use_proxy_var.get(),
)
self.pending_refaire_commands = self._refaire_commands()[1:] if step.section == REFAIRE_SECTION else []
self.refaire_command_index = 1
self._append_console(f"\n$ {command_display(command)}\n")
try:
self.runner.start(command, self.repository, environment, log_path)
@@ -846,6 +1040,7 @@ class CopienatorApp(tk.Tk):
{"step": step.id, "command": command_display(command), "status": "failed", "error": str(exc)}
)
self.active_step_id = None
self.pending_refaire_commands = []
self._append_console(f"Impossible de lancer la commande : {exc}\n")
messagebox.showerror("Échec du lancement", str(exc))
self._populate_tree()
@@ -854,10 +1049,10 @@ class CopienatorApp(tk.Tk):
def _mark_step(
self, status: str, *, automatic: bool = False, reason: str | None = None
) -> None:
if not self.current_step or not self.state_store.evaluation:
if not self.current_step or not self.state_store.evaluation or self.active_step_id:
return
self._save_current_form()
self.state_store.invalidate_after([item.id for item in self.steps], self.current_step.id)
self.state_store.invalidate_after(self._progression_ids(self.current_step.id), self.current_step.id)
self.state_store.update_step(self.current_step.id, status=status)
history: dict[str, object] = {
"step": self.current_step.id,
@@ -875,7 +1070,10 @@ class CopienatorApp(tk.Tk):
def _skip_step(self) -> None:
if self.current_step and self.current_step.optional:
step_id = self.current_step.id
self._mark_step("skipped")
if step_id.startswith("refaire_"):
self._move_selection_from(step_id, 1)
def _poll_runner(self) -> None:
while True:
@@ -899,6 +1097,26 @@ class CopienatorApp(tk.Tk):
if not step_id:
return
status = process_status(return_code, interrupted)
if status == "success" and self.pending_refaire_commands:
command = self.pending_refaire_commands.pop(0)
self.refaire_command_index += 1
self._append_console(f"\n$ {command_display(command)}\n")
self.state_store.update_step(step_id, command=command_display(command))
try:
self.runner.start(command, self.repository,
build_runner_environment(os.environ, self.api_key_var.get(), self.proxy_var.get(), self.use_proxy_var.get()),
self.state_store.workspace.log_path(f"{step_id}_{self.refaire_command_index}"))
self._update_controls()
return
except (OSError, RuntimeError) as exc:
self._append_console(f"Impossible de lancer la copie suivante : {exc}\n")
status, return_code = "failed", 1
self.pending_refaire_commands = []
if step_id == "refaire_merge" and status == "success":
for downstream in ("giving_names", "update_ods", "final_score", "personal_deploy", "personal_sent"):
entry = self.state_store.step(downstream)
if entry.get("status") in {"success", "detected", "skipped"}:
self.state_store.update_step(downstream, status="stale")
if step_id == "clean" and status == "success":
self._append_console(
f"\n[Terminé — code {return_code}{STATUS_LABELS[status]}]\n"
@@ -977,8 +1195,9 @@ class CopienatorApp(tk.Tk):
self.console.configure(state="disabled")
def _update_controls(self) -> None:
running = self.runner.running
running = bool(self.active_step_id) or self.runner.running
self.run_button.configure(state="disabled" if running or not self.current_step else "normal")
self.skip_button.configure(state="normal" if not running and self.current_step and self.current_step.optional else "disabled")
self.interrupt_button.configure(state="normal" if running else "disabled")
self.force_button.configure(state="normal" if running else "disabled")
self.send_button.configure(state="normal" if running else "disabled")
@@ -990,7 +1209,7 @@ class CopienatorApp(tk.Tk):
self._move_selection_from(self.current_step.id, delta)
def _move_selection_from(self, step_id: str, delta: int) -> None:
ids = [step.id for step in self.steps]
ids = self._progression_ids(step_id)
try:
index = ids.index(step_id)
except ValueError:
+244
View File
@@ -0,0 +1,244 @@
"""Selection and command planning for the optional redo workflow."""
from __future__ import annotations
import re
import tkinter as tk
from pathlib import Path
from tkinter import ttk
from copienator import EvaluationWorkspace, read_json
from copienator.utils import natural_key
SECTION = "Refaire des copies (facultatif)"
ALL_LABELS = "Toute la copie"
ALL_COPIES = "Toutes les copies"
LAYOUTS = {
"Automatique": "auto",
"Par question (groupé)": "grouped",
"Par copie": "copies",
}
def resolve_layout(
selection: list[list], labels: list[str], choice: str = "auto"
) -> str:
if choice in {"grouped", "copies"}:
return choice
seen = set()
for _name, selected in selection:
current = set(selected or labels)
if seen & current:
return "grouped"
seen.update(current)
return "copies"
def copies_with_answer(copies: dict[str, Path], label: str) -> list[str]:
return [
name
for name, path in copies.items()
if any(
(path.with_suffix("") / f"{label}{suffix}.pdf").is_file()
for suffix in ("", "_new")
)
]
def available_copies(evaluation: Path) -> dict[str, Path]:
return {
path.stem: path
for path in sorted((evaluation / "Copies").glob("Copie*.pdf"), key=natural_key)
if re.fullmatch(r"Copie\d+", path.stem)
}
def validate_selection(
entries: object, copies: dict[str, Path], labels: list[str]
) -> list[list]:
if not isinstance(entries, list) or not entries:
raise ValueError("Ajoutez au moins une copie à refaire.")
result = {}
for entry in entries:
if not isinstance(entry, list) or len(entry) != 2:
raise ValueError("Sélection de copies invalide.")
name, selected = entry
if not isinstance(name, str) or name not in copies:
raise ValueError(f"Copie introuvable : {name}")
if not isinstance(selected, list) or any(
not isinstance(label, str) or label not in labels for label in selected
):
raise ValueError(f"Question inconnue pour {name}. Reprenez la sélection.")
if name in result:
raise ValueError(f"Copie sélectionnée plusieurs fois : {name}")
result[name] = sorted(set(selected), key=natural_key)
return [[name, result[name]] for name in sorted(result, key=natural_key)]
def load_selection(evaluation: Path) -> list[list]:
return validate_selection(
read_json(evaluation / "refaire.json"),
available_copies(evaluation),
EvaluationWorkspace(evaluation).read_labels(),
)
class RefaireSelection(ttk.Frame):
def __init__(
self,
parent,
evaluation: Path,
draft: dict,
directories: tuple[str, ...],
preferred: str,
):
super().__init__(parent)
self.columnconfigure(1, weight=1)
self.copies = available_copies(evaluation)
self.labels = (
EvaluationWorkspace(evaluation).read_labels()
if (evaluation / "labels").is_file()
else []
)
self.entries = {}
self.error = ""
try:
entries = draft.get("selection")
if entries is None:
entries = read_json(evaluation / "refaire.json", default=[])
if entries:
self.entries = dict(
validate_selection(entries, self.copies, self.labels)
)
except (OSError, ValueError, TypeError) as exc:
self.error = str(exc)
self.copy_var = tk.StringVar(value=next(iter(self.copies), ""))
self.label_var = tk.StringVar(value=ALL_LABELS)
source = draft.get("annotation_dir", preferred)
self.source_var = tk.StringVar(
value=source if source in directories else preferred
)
ttk.Label(self, text="Copie").grid(row=0, column=0, sticky="w", padx=(0, 8))
ttk.Combobox(
self,
textvariable=self.copy_var,
values=(ALL_COPIES, *self.copies),
state="readonly",
width=12,
).grid(row=0, column=1, sticky="ew")
ttk.Label(self, text="Question").grid(row=1, column=0, sticky="w", pady=5)
ttk.Combobox(
self,
textvariable=self.label_var,
values=(ALL_LABELS, *self.labels),
state="readonly",
width=12,
).grid(row=1, column=1, sticky="ew", pady=5)
ttk.Button(self, text="+ Ajouter", command=self.add).grid(
row=1, column=2, padx=(8, 0)
)
self.table = ttk.Treeview(
self, columns=("labels",), height=4, selectmode="browse"
)
self.table.heading("#0", text="Copie")
self.table.heading("labels", text="Questions à refaire")
self.table.column("#0", width=100, stretch=False)
self.table.column("labels", width=330)
self.table.grid(row=4, column=0, columnspan=3, sticky="nsew")
scrollbar = ttk.Scrollbar(self, orient="vertical", command=self.table.yview)
scrollbar.grid(row=4, column=3, sticky="ns")
self.table.configure(yscrollcommand=scrollbar.set)
ttk.Button(
self, text="Retirer la copie sélectionnée", command=self.remove
).grid(row=5, column=0, columnspan=3, sticky="w", pady=5)
ttk.Label(self, text="Passage principal").grid(
row=3, column=0, sticky="w", padx=(0, 8)
)
ttk.Combobox(
self,
textvariable=self.source_var,
values=directories,
state="readonly",
width=12,
).grid(row=3, column=1, sticky="ew")
layout = draft.get("layout", "auto")
self.layout_var = tk.StringVar(
value=next(
(name for name, code in LAYOUTS.items() if code == layout),
"Automatique",
)
)
ttk.Label(self, text="PDF à vérifier").grid(row=2, column=0, sticky="w")
ttk.Combobox(
self,
textvariable=self.layout_var,
values=tuple(LAYOUTS),
state="readonly",
width=12,
).grid(row=2, column=1, sticky="ew")
self.message_var = tk.StringVar(
value=self.error
or "Choisissez « Toutes les copies » pour refaire une question dans toute la classe."
)
help_label = ttk.Label(
self,
textvariable=self.message_var,
wraplength=500,
)
help_label.grid(row=6, column=0, columnspan=3, sticky="w", pady=5)
self.bind(
"<Configure>",
lambda event: help_label.configure(wraplength=max(200, event.width - 10)),
)
self.refresh()
def refresh(self):
self.table.delete(*self.table.get_children())
for name in sorted(self.entries, key=natural_key):
self.table.insert(
"",
"end",
iid=name,
text=name,
values=(", ".join(self.entries[name]) or ALL_LABELS,),
)
def add(self):
name, label = self.copy_var.get(), self.label_var.get()
if name not in (ALL_COPIES, *self.copies) or label not in (
ALL_LABELS,
*self.labels,
):
return
names = (
[name]
if name != ALL_COPIES
else list(self.copies)
if label == ALL_LABELS
else copies_with_answer(self.copies, label)
)
for copy_name in names:
# Adding a question must not narrow a copy already selected in full.
if label == ALL_LABELS:
self.entries[copy_name] = []
elif copy_name not in self.entries or self.entries[copy_name]:
self.entries[copy_name] = sorted(
set(self.entries.get(copy_name, [])) | {label}, key=natural_key
)
message = f"{len(names)} copie(s) ajoutée(s)."
if name == ALL_COPIES and len(names) < len(self.copies):
message += f" {len(self.copies) - len(names)} sans réponse découpée pour cette question."
self.message_var.set(message)
self.refresh()
def remove(self):
for name in self.table.selection():
self.entries.pop(name, None)
self.refresh()
def values(self):
return {
"selection": [[name, labels] for name, labels in self.entries.items()],
"annotation_dir": self.source_var.get(),
"layout": LAYOUTS[self.layout_var.get()],
}
+86
View File
@@ -0,0 +1,86 @@
"""Preserve completed redo passes and activate a fresh working directory."""
from __future__ import annotations
import copy
import shutil
import uuid
from datetime import datetime
from typing import Any
from copienator import EvaluationWorkspace, atomic_write_json, read_json
from copienator.filesystem import staged_files
RESTART_WARNING = (
"Appelez « Nouvelle reprise » seulement après avoir importé les résultats "
"de la reprise en cours et exécuté « Mettre à jour les copies finales ». "
"Cette consigne sapplique aussi à « Refaire la même sélection »."
)
def _identifier() -> str:
return f"reprise-{datetime.now().astimezone():%Y%m%d-%H%M%S}-{uuid.uuid4().hex[:8]}"
def begin_pass(
workspace: EvaluationWorkspace,
state: dict[str, Any],
*,
keep_selection: bool,
) -> tuple[dict[str, Any], str]:
"""Activate a pass atomically with its selection and reset GUI state.
Existing review files remain in place. Legacy BRnot is copied once into
the archive; failures before activation leave the old pass active.
"""
selection = read_json(workspace.refaire_file, default=[])
previous = workspace.refaire_session_dir
if previous is None and (
workspace.refaire_file.exists() or workspace.annotation_dir("refaire").exists()
):
previous = workspace.root / "Reprises" / _identifier()
previous.mkdir(parents=True)
legacy = workspace.annotation_dir("refaire")
if legacy.is_dir():
shutil.copytree(legacy, previous / "BRnot")
if previous is not None:
atomic_write_json(previous / "refaire.json", selection)
atomic_write_json(
previous / "progression.json",
{
name: entry
for name, entry in state.get("steps", {}).items()
if name.startswith("refaire_")
},
)
ident = _identifier()
directory = workspace.root / "Reprises" / ident
directory.mkdir(parents=True)
new_selection = selection if keep_selection else []
updated = copy.deepcopy(state)
values = dict(
updated.get("steps", {}).get("refaire_selection", {}).get("values", {})
)
values["selection"] = new_selection
updated["steps"] = {
name: entry
for name, entry in updated.get("steps", {}).items()
if not name.startswith("refaire_")
}
updated["steps"]["refaire_selection"] = {"values": values}
updated.setdefault("history", []).append(
{
"step": "refaire_selection",
"action": "repeat" if keep_selection else "new",
"session": ident,
"timestamp": datetime.now().astimezone().isoformat(),
}
)
atomic_write_json(directory / "refaire.json", new_selection)
atomic_write_json(directory / "session.json", {"id": ident, "values": values})
with staged_files(workspace.root) as staging:
atomic_write_json(staging / workspace.refaire_file.name, new_selection)
atomic_write_json(staging / workspace.gui_state_file.name, updated)
atomic_write_json(staging / "refaire-session.json", {"id": ident})
return updated, ident
+37
View File
@@ -505,9 +505,46 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
requires=("Copies", "correction.json", "A Rendre"),
),
]
steps[-1:-1] = build_refaire_workflow()
return [step for step in steps if show_personal_steps or not step.personal]
def build_refaire_workflow() -> list[StepDefinition]:
from .refaire import SECTION
selection = StepDefinition(
"refaire_selection", SECTION, "Choisir les copies et les questions",
"Sélectionnez les copies et les questions à refaire, puis enregistrez la sélection. "
"Le passage principal doit être terminé ; conservez ses annotations.",
(CommandVariant("default", "Sélection", None, "manual"),),
requires=("Copies", "labels", "correction.json"),
)
definitions = [
("review", "Reprendre le découpage", "Vérifiez et ajustez les labels des copies sélectionnées. Chaque copie souvre à son tour. Fermez la fenêtre pour passer à la suivante.", "review-labels", (), True),
("split", "Redécouper les réponses", "À exécuter après une modification du découpage. Traite toutes les copies sélectionnées ; vérifiez les fichiers _new et _old en cas de résolution manuelle.", "split-answers", (), True),
("correct", "Refaire la correction", "Relance la correction des seules questions sélectionnées. Peut être ignorée pour corriger manuellement les résultats.", "correct", ("--refaire",), True),
("annotate", "Préparer les copies à vérifier", "Génère les questions sélectionnées avec des cases dans BRnot, par question ou par copie selon la sélection. Remplace le précédent passage dans BRnot.", "annotate-checks", ("--refaire", "--overwrite"), False),
("export", "Exporter vers la tablette", "Exporte BRnot vers EXPORT_DIR. Retirez les anciens fichiers dexport avant le transfert.", "export", ("--refaire",), True),
("tablet", "Vérifier sur la tablette", "Annotez les PDF exportés, puis placez les retours dans IMPORT_DIR sans changer leur nom (nom de groupe ou Copie01.pdf…). Retournez aussi les PDF sans modification. Retirez les anciens fichiers dimport.", None, (), False),
("import", "Importer les copies vérifiées", "Importe les PDF retournés dans BRnot. Vous pouvez ignorer cette étape si les fichiers Concat_annotated.pdf y sont déjà en place.", "import", ("--refaire",), True),
("merge", "Mettre à jour les copies finales", "Fusionne les questions refaites avec le reste de chaque copie dans le dossier du passage principal. Relancez ensuite la préparation de A Rendre, le calcul des notes et la diffusion.", "read-grouped", ("--refaire",), False),
]
steps = [selection]
for suffix, title, description, program, flags, optional in definitions:
arguments = (arg_target(),) if program else ()
if suffix == "merge":
arguments += (ArgumentSpec("annotation_dir", "Passage principal", "choice", "--annotation-dir", default="BGnot", choices=("BGnot", "Bnot", "Anot")),)
requirements = ("refaire.json", "Copies", "labels", "correction.json")
if suffix in {"export", "tablet", "import", "merge"}:
requirements += ("BRnot",)
steps.append(StepDefinition(
f"refaire_{suffix}", SECTION, title, description,
(CommandVariant("default", title, program, "python" if program else "manual", flags),),
arguments=arguments, optional=optional, requires=requirements,
))
return steps
def value_for_default(value: object, evaluation_arg: str) -> object:
return evaluation_arg if value == EVALUATION else value
+206
View File
@@ -0,0 +1,206 @@
from __future__ import annotations
import shutil
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from PIL import Image, ImageDraw
from copienator import EvaluationWorkspace, ExitCode, atomic_write_json, read_json
from copienator.annotation_data import AnnotationLoadResult
from copienator.commands import annotating_by_label as grouped
from copienator.commands import annotating_with_checks as checks
from copienator.commands import export, import_annotations
from copienator.commands import reading_grouped_annotations as reader
class GroupedRedoTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.root = Path(self.temp.name) / "Exam"
for directory in ("Copies", "Par label", "BGnot", "BRnot"):
(self.root / directory).mkdir(parents=True)
(self.root / "labels").write_text("Ex 1\nEx 2\n")
atomic_write_json(self.root / "correction.json", {})
atomic_write_json(
self.root / "refaire.json", [["Copie01", ["Ex 1"]], ["Copie02", ["Ex 1"]]]
)
(self.root / "BRnot/previous.txt").write_text("previous redo")
(self.root / "BGnot/main.txt").write_text("main run")
self.workspace = EvaluationWorkspace(self.root)
self.data = {"01": {"Ex 1": {}}, "02": {"Ex 1": {}}}
def tearDown(self):
self.temp.cleanup()
@staticmethod
def render(item):
student_id, label, _content = item
image = Image.new("RGB", (100, 100), "white")
ImageDraw.Draw(image).rectangle((10, 10, 50, 50), outline="black", width=2)
return (
student_id,
label,
image,
0,
[
{
"type": "score",
"label": label,
"value": 3,
"final_box": [10, 10, 50, 50],
}
],
)
def generate(self):
with (
patch.object(
grouped,
"load_annotation_data",
return_value=AnnotationLoadResult(self.data, []),
) as load,
patch.object(grouped, "render_item", side_effect=self.render),
patch.object(grouped, "_load_label_groups") as label_groups,
):
self.assertEqual(
grouped.run(self.workspace, refaire=True, overwrite=True),
ExitCode.SUCCESS,
)
self.assertEqual(
load.call_args.kwargs["refaire_list"],
read_json(self.root / "refaire.json"),
)
label_groups.assert_not_called()
directories = list((self.root / "BRnot").iterdir())
self.assertEqual(len(directories), 1)
self.assertTrue(directories[0].is_dir())
self.assertEqual((self.root / "BGnot/main.txt").read_text(), "main run")
return directories[0]
def test_grouped_redo_export_import_and_actual_annotation_detection(self):
directory = self.generate()
metadata = read_json(directory / "bnote.json")["images"]
self.assertEqual(
[(item["id"], item["label"]) for item in metadata],
[("01", "Ex 1"), ("02", "Ex 1")],
)
export_root = Path(self.temp.name) / "Export"
with patch.object(export, "EXPORT_DIR", export_root):
self.assertEqual(export.run(self.workspace, refaire=True), ExitCode.SUCCESS)
self.assertEqual(len(list((export_root / "Exam").glob("*.pdf"))), 1)
imported_root = Path(self.temp.name) / "Import"
imported_root.mkdir()
with Image.open(directory / "Reference.jpg") as reference:
annotated = reference.convert("RGB")
draw = ImageDraw.Draw(annotated)
draw.rectangle((17, 17, 43, 43), fill="black")
draw.rectangle((70, 170, 90, 190), fill="black")
annotated.save(imported_root / f"{directory.name}.pdf", "PDF", resolution=72)
with patch.object(import_annotations, "IMPORT_DIR", imported_root):
self.assertEqual(
import_annotations.run(self.workspace, refaire=True), ExitCode.SUCCESS
)
actions, notes, incomplete = reader._scan_redo_annotations(
self.root / "BRnot", {"01": {"Ex 1"}, "02": {"Ex 1"}}
)
self.assertFalse(incomplete)
self.assertEqual(actions["01"][0]["value"], 3)
self.assertFalse(actions.get("02"))
self.assertIn("Ex 1", notes["02"])
full_data = {student: {"Ex 1": {}, "Ex 2": {}} for student in ("01", "02")}
for mode in ("BGnot", "Bnot", "Anot"):
(self.root / mode).mkdir(exist_ok=True)
with (
self.subTest(mode=mode),
patch.object(
reader,
"load_annotation_data",
return_value=AnnotationLoadResult(full_data, []),
),
patch.object(
reader,
"apply_actions_and_regenerate_grouped",
return_value=(ExitCode.SUCCESS, ""),
) as regenerate,
):
self.assertEqual(
reader.run(self.workspace, refaire=True, annotation_dir=mode),
ExitCode.SUCCESS,
)
calls = {call.args[2]: call for call in regenerate.call_args_list}
self.assertEqual(set(calls), {"01", "02"})
self.assertEqual(set(calls["01"].args[1]["01"]), {"Ex 1", "Ex 2"})
self.assertEqual(calls["01"].args[3][0]["value"], 3)
self.assertIn("Ex 1", calls["02"].args[4])
def test_missing_group_leaves_affected_copy_incomplete(self):
directory = self.generate()
shutil.copy2(directory / "Concat.pdf", directory / "Concat_annotated.pdf")
extra = self.root / "BRnot/Ex 2 G1"
extra.mkdir()
atomic_write_json(
extra / "bnote.json", {"images": [{"id": "01", "label": "Ex 2"}]}
)
_actions, _notes, incomplete = reader._scan_redo_annotations(
self.root / "BRnot", {"01": {"Ex 1", "Ex 2"}, "02": {"Ex 1"}}
)
self.assertEqual(incomplete, {"01"})
def test_failed_generation_preserves_previous_redo_for_both_layouts(self):
for module, worker in ((grouped, "render_item"), (checks, "_render_student")):
with (
self.subTest(module=module),
patch.object(
module,
"load_annotation_data",
return_value=AnnotationLoadResult(self.data, []),
),
patch.object(
module,
worker,
return_value=None if module is grouped else "partial",
),
):
if module is grouped:
status = module.run(self.workspace, refaire=True, overwrite=True)
else:
status = module.run(
self.workspace, self.root, refaire=True, overwrite=True
)
self.assertEqual(status, ExitCode.PARTIAL)
self.assertEqual(
(self.root / "BRnot/previous.txt").read_text(), "previous redo"
)
def test_switching_to_per_copy_removes_previous_group_layout(self):
self.generate()
def render(_workspace, student_id, _labels, **kwargs):
output = kwargs["output_root"] / f"Copie{student_id}"
output.mkdir()
(output / "Concat.pdf").touch()
return "success"
with (
patch.object(
checks,
"load_annotation_data",
return_value=AnnotationLoadResult(self.data, []),
),
patch.object(checks, "_render_student", side_effect=render),
):
self.assertEqual(
checks.run(self.workspace, self.root, refaire=True, overwrite=True),
ExitCode.SUCCESS,
)
self.assertEqual(
{path.name for path in (self.root / "BRnot").iterdir()},
{"Copie01", "Copie02"},
)
if __name__ == "__main__":
unittest.main()
+308
View File
@@ -0,0 +1,308 @@
from __future__ import annotations
import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from copienator import atomic_write_json, read_json
from copienator_gui.app import CopienatorApp
from copienator_gui.refaire import (
ALL_COPIES,
SECTION,
available_copies,
resolve_layout,
validate_selection,
)
from copienator_gui.workflow import build_command, build_workflow
class SelectionTests(unittest.TestCase):
def test_selection_validates_and_deduplicates_labels(self):
copies = {"Copie01": Path("/tmp/Copie01.pdf")}
self.assertEqual(
validate_selection(
[["Copie01", ["Ex 2", "Ex 1", "Ex 2"]]], copies, ["Ex 1", "Ex 2"]
),
[["Copie01", ["Ex 1", "Ex 2"]]],
)
for invalid in (
[],
[["missing", []]],
[["Copie01", ["unknown"]]],
[["../Copie01", []]],
):
with self.subTest(invalid=invalid), self.assertRaises(ValueError):
validate_selection(invalid, copies, ["Ex 1"])
def test_automatic_layout_groups_shared_labels_and_honors_explicit_choice(self):
selection = [["Copie01", ["Ex 1"]], ["Copie02", ["Ex 1"]]]
self.assertEqual(resolve_layout(selection, ["Ex 1", "Ex 2"]), "grouped")
self.assertEqual(resolve_layout(selection, ["Ex 1"], "copies"), "copies")
self.assertEqual(resolve_layout([["Copie01", ["Ex 1"]]], ["Ex 1"]), "copies")
self.assertEqual(
resolve_layout([["Copie01", ["Ex 1"]], ["Copie02", []]], ["Ex 1"]),
"grouped",
)
def test_redo_steps_are_in_both_profiles_and_never_autostart(self):
for personal in (False, True):
steps = [
step for step in build_workflow(personal) if step.section == SECTION
]
self.assertEqual(len(steps), 9)
self.assertEqual(steps[0].id, "refaire_selection")
self.assertTrue(all(not step.auto_start_first_visit for step in steps))
merge = steps[-1]
for mode in ("BGnot", "Bnot", "Anot"):
command = build_command(
Path.cwd(),
merge,
merge.variants[0],
{"target": "/tmp/Exam", "annotation_dir": mode},
"/tmp/Exam",
)
self.assertEqual(
command[4:],
[
"read-grouped",
"/tmp/Exam",
"--refaire",
"--annotation-dir",
mode,
],
)
@unittest.skipUnless(
os.environ.get("DISPLAY"), "Tk tests require a display (use xvfb-run)"
)
class RefaireGuiTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.root = Path(self.temp.name)
(self.root / "Copies").mkdir()
(self.root / "Anot").mkdir()
(self.root / "Par label").mkdir()
(self.root / "BRnot").mkdir()
for name in ("Copie01", "Copie02"):
(self.root / "Copies" / f"{name}.pdf").touch()
(self.root / "labels").write_text("Ex 1\nEx 2\n", encoding="utf-8")
atomic_write_json(self.root / "correction.json", {})
self.app = CopienatorApp(Path.cwd(), False, self.root)
self.app.update()
def tearDown(self):
for callback in self.app.tk.splitlist(self.app.tk.call("after", "info")):
self.app.after_cancel(callback)
self.app.destroy()
self.temp.cleanup()
def select(self, ident):
self.app.tree.selection_set(ident)
self.app.tree.see(ident)
self.app.update()
def save_selection(self):
self.select("refaire_selection")
panel = self.app.refaire_panel
panel.label_var.set("Ex 1")
panel.add()
panel.label_var.set("Ex 2")
panel.add()
panel.copy_var.set("Copie02")
panel.label_var.set("Toute la copie")
panel.add()
self.app._run_current_step()
self.app.update()
def test_collapsed_branch_is_separate_and_stays_open_when_refreshed(self):
section = self.app.tree.parent("refaire_selection")
self.assertFalse(self.app.tree.item(section, "open"))
self.assertNotIn("refaire_selection", self.app._progression_ids("giving_names"))
self.assertNotIn("clean", self.app._progression_ids("refaire_merge"))
self.select("refaire_selection")
self.app._populate_tree()
self.assertTrue(self.app.tree.item(section, "open"))
def test_dropdown_selection_saves_json_and_drives_all_commands(self):
self.save_selection()
self.assertEqual(
read_json(self.root / "refaire.json"),
[["Copie01", ["Ex 1", "Ex 2"]], ["Copie02", []]],
)
self.assertEqual(self.app.current_step.id, "refaire_review")
commands = self.app._refaire_commands()
self.assertEqual(
[command[5] for command in commands],
[str(path) for path in available_copies(self.root).values()],
)
self.select("refaire_annotate")
self.assertEqual(
self.app._make_command()[4:],
["annotate-grouped", str(self.root), "--refaire", "--overwrite"],
)
self.select("refaire_merge")
self.assertEqual(
self.app._make_command()[4:],
["read-grouped", str(self.root), "--refaire", "--annotation-dir", "Anot"],
)
self.assertEqual(self.app.state_store.step("annotation").get("status"), None)
def test_one_label_for_all_copies_and_layout_override(self):
for name in ("Copie01", "Copie02"):
directory = self.root / "Copies" / name
directory.mkdir()
(directory / "Ex 1.pdf").touch()
self.select("refaire_selection")
panel = self.app.refaire_panel
panel.copy_var.set(ALL_COPIES)
panel.label_var.set("Ex 1")
panel.add()
self.assertEqual(panel.entries, {"Copie01": ["Ex 1"], "Copie02": ["Ex 1"]})
panel.layout_var.set("Par copie")
self.app._run_current_step()
self.app.update()
self.select("refaire_annotate")
self.assertEqual(self.app._make_command()[4], "annotate-checks")
def test_bulk_add_skips_absent_answers_and_preserves_whole_copy_selection(self):
directory = self.root / "Copies/Copie01"
directory.mkdir()
(directory / "Ex 1_new.pdf").touch()
self.select("refaire_selection")
panel = self.app.refaire_panel
panel.copy_var.set("Copie01")
panel.add()
panel.copy_var.set(ALL_COPIES)
panel.label_var.set("Ex 1")
panel.add()
self.assertEqual(panel.entries, {"Copie01": []})
self.assertIn("1 sans réponse", panel.message_var.get())
def test_correction_folder_buttons_open_expected_folders(self):
for name in ("Sol", "Persp"):
(self.root / name).mkdir()
self.select("refaire_selection")
buttons = self.app.form.winfo_children()[0].winfo_children()
with patch("copienator_gui.app.open_path") as opened:
for button in buttons:
button.invoke()
self.assertEqual(
[call.args[0] for call in opened.call_args_list],
[self.root / "Sol", self.root / "Persp"],
)
def test_restart_actions_warn_reset_and_preserve_or_clear_selection(self):
self.save_selection()
self.app.state_store.update_step("refaire_merge", status="success")
self.app.state_store.update_step("annotation", status="success")
selection = read_json(self.root / "refaire.json")
with patch(
"copienator_gui.app.messagebox.askyesno", return_value=False
) as confirm:
self.app._start_refaire_pass(False)
self.assertIn("importé", confirm.call_args.args[1])
self.assertIn("Mettre à jour", confirm.call_args.args[1])
self.assertEqual(read_json(self.root / "refaire.json"), selection)
with patch("copienator_gui.app.messagebox.askyesno", return_value=True):
self.app._start_refaire_pass(True)
self.app.update()
self.assertEqual(self.app.current_step.id, "refaire_selection")
self.assertEqual(self.app.refaire_panel.values()["selection"], selection)
self.assertIsNone(self.app.state_store.step("refaire_merge").get("status"))
self.assertEqual(self.app.state_store.step("annotation")["status"], "success")
first = self.app.state_store.workspace.refaire_session_id
with patch(
"copienator_gui.app.messagebox.askyesno", return_value=True
) as confirm:
self.app._start_refaire_pass(False)
self.app.update()
self.assertIn("nest pas marquée comme fusionnée", confirm.call_args.args[1])
self.assertEqual(self.app.refaire_panel.entries, {})
self.assertEqual(read_json(self.root / "refaire.json"), [])
self.assertNotEqual(self.app.state_store.workspace.refaire_session_id, first)
self.assertEqual(self.app.refaire_panel.source_var.get(), "Anot")
def test_restart_is_blocked_while_a_command_is_active(self):
self.save_selection()
self.app.active_step_id = "refaire_correct"
with (
patch("copienator_gui.app.messagebox.showwarning") as warning,
patch("copienator_gui.app.messagebox.askyesno") as confirm,
):
self.app._start_refaire_pass(False)
warning.assert_called_once()
confirm.assert_not_called()
self.assertIsNone(self.app.state_store.workspace.refaire_session_id)
def test_small_window_keeps_selection_accessible_by_scrolling(self):
self.select("refaire_selection")
self.app.geometry("900x640")
self.app.update()
self.app.form_canvas.yview_moveto(1)
self.app.update()
self.assertAlmostEqual(self.app.form_canvas.yview()[1], 1.0)
self.assertLess(
self.app.run_button.winfo_rooty() + self.app.run_button.winfo_height(),
self.app.winfo_rooty() + self.app.winfo_height(),
)
def test_queue_runs_copies_in_order_and_stops_on_failure(self):
self.save_selection()
self.select("refaire_split")
with patch.object(self.app.runner, "start") as start:
self.app._run_current_step()
self.assertEqual(start.call_count, 1)
self.assertEqual(len(self.app.pending_refaire_commands), 1)
self.app._finish_process(0, False)
self.assertEqual(start.call_count, 2)
self.assertEqual(self.app.active_step_id, "refaire_split")
self.app._finish_process(1, False)
self.assertEqual(
self.app.state_store.step("refaire_split")["status"], "failed"
)
self.assertIsNone(self.app.active_step_id)
self.assertFalse(self.app.pending_refaire_commands)
def test_interruption_does_not_launch_next_copy(self):
self.save_selection()
self.select("refaire_split")
with patch.object(self.app.runner, "start") as start:
self.app._run_current_step()
self.app._finish_process(130, True)
self.assertEqual(start.call_count, 1)
self.assertFalse(self.app.pending_refaire_commands)
self.assertEqual(
self.app.state_store.step("refaire_split")["status"], "interrupted"
)
def test_unsaved_or_external_changes_block_commands(self):
self.save_selection()
atomic_write_json(self.root / "refaire.json", [["Copie02", []]])
with self.assertRaisesRegex(ValueError, "sélection a changé"):
self.app._refaire_commands()
def test_selection_reload_and_finalization_invalidation(self):
self.save_selection()
self.app.state_store.update_step("giving_names", status="success")
self.app.state_store.update_step(
"annotation", status="success", last_run_variant="simple"
)
self.select("refaire_merge")
with patch.object(self.app.runner, "start"):
self.app._run_current_step()
self.app._finish_process(0, False)
self.app.update()
self.assertEqual(self.app.state_store.step("giving_names")["status"], "stale")
self.assertEqual(self.app.state_store.step("annotation")["status"], "success")
self.assertEqual(self.app.current_step.id, "refaire_merge")
self.select("refaire_selection")
self.assertEqual(
self.app.refaire_panel.entries, {"Copie01": ["Ex 1", "Ex 2"], "Copie02": []}
)
if __name__ == "__main__":
unittest.main()
+280
View File
@@ -0,0 +1,280 @@
from __future__ import annotations
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from PIL import Image
from copienator import EvaluationWorkspace, ExitCode, atomic_write_json, read_json
from copienator.annotation_data import AnnotationLoadResult, _coordinate_index
from copienator.commands import reading_grouped_annotations as reader
class RefaireTests(unittest.TestCase):
def workspace(self, root, mode):
for name in ("Copies", "Par label", mode, "BRnot/Copie01"):
(root / name).mkdir(parents=True)
(root / "labels").write_text("Ex 1\nEx 2\n")
atomic_write_json(root / "correction.json", {})
atomic_write_json(root / "refaire.json", [["Copie01", ["Ex 2"]]])
for name in (
"bnote.json",
"checkboxes.json",
"Reference.jpg",
"Concat_annotated.pdf",
):
(root / "BRnot/Copie01" / name).touch()
atomic_write_json(
root / "BRnot/Copie01/bnote.json", {"images": [{"label": "Ex 2"}]}
)
return EvaluationWorkspace(root)
def test_reader_merges_full_copy_for_every_original_mode(self):
for mode in ("BGnot", "Bnot", "Anot"):
with self.subTest(mode=mode), tempfile.TemporaryDirectory() as tmp:
workspace = self.workspace(Path(tmp), mode)
data = {"01": {"Ex 1": {}, "Ex 2": {}}, "02": {"Ex 1": {}}}
def load(_workspace, data=data, **kwargs):
return AnnotationLoadResult(
{"01": {"Ex 2": {}}} if kwargs else data, []
)
with (
patch.object(reader, "load_annotation_data", side_effect=load),
patch.object(
reader,
"_scan_annotation_directory",
return_value=(
{"01": [{"label": "Ex 2", "type": "score", "value": 3}]},
{},
),
),
patch.object(
reader,
"apply_actions_and_regenerate_grouped",
return_value=(ExitCode.SUCCESS, ""),
) as render,
):
self.assertEqual(
reader.run(workspace, refaire=True, annotation_dir=mode),
ExitCode.SUCCESS,
)
render.assert_called_once()
self.assertEqual(set(render.call_args.args[1]["01"]), {"Ex 1", "Ex 2"})
self.assertEqual(render.call_args.args[2], "01")
self.assertEqual(render.call_args.kwargs["selected_labels"], {"Ex 2"})
self.assertEqual(render.call_args.kwargs["annotation_dir"], mode)
def test_whole_copy_selection_replaces_all_old_actions(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
workspace = self.workspace(root, "BGnot")
(root / "BGnot/Ex 1").mkdir()
atomic_write_json(root / "refaire.json", [["Copie01", []]])
atomic_write_json(
root / "BRnot/Copie01/bnote.json",
{"images": [{"label": label} for label in ("Ex 1", "Ex 2")]},
)
loaded = AnnotationLoadResult({"01": {"Ex 1": {}, "Ex 2": {}}}, [])
def scan(directory, *args, **kwargs):
return (
({"01": [{"label": "Ex 1", "type": "score", "value": 1}]}, {})
if directory.parent.name == "BGnot"
else ({"01": [{"label": "Ex 2", "type": "score", "value": 4}]}, {})
)
with (
patch.object(reader, "load_annotation_data", return_value=loaded),
patch.object(reader, "_scan_annotation_directory", side_effect=scan),
patch.object(
reader,
"apply_actions_and_regenerate_grouped",
return_value=(ExitCode.SUCCESS, ""),
) as render,
):
self.assertEqual(reader.run(workspace, refaire=True), ExitCode.SUCCESS)
self.assertEqual(
render.call_args.kwargs["selected_labels"], {"Ex 1", "Ex 2"}
)
self.assertEqual(
render.call_args.args[3],
[{"label": "Ex 2", "type": "score", "value": 4}],
)
def test_stale_redo_selection_is_rejected(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
workspace = self.workspace(root, "Anot")
atomic_write_json(
root / "BRnot/Copie01/bnote.json", {"images": [{"label": "Ex 1"}]}
)
with (
patch.object(
reader,
"load_annotation_data",
return_value=AnnotationLoadResult({"01": {"Ex 2": {}}}, []),
),
patch.object(reader, "apply_actions_and_regenerate_grouped") as render,
):
self.assertEqual(
reader.run(workspace, refaire=True, annotation_dir="Anot"),
ExitCode.PARTIAL,
)
render.assert_not_called()
def test_missing_redo_return_leaves_copy_untouched(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
workspace = self.workspace(root, "BGnot")
(root / "BRnot/Copie01/Concat_annotated.pdf").unlink()
with (
patch.object(
reader,
"load_annotation_data",
return_value=AnnotationLoadResult({"01": {"Ex 2": {}}}, []),
),
patch.object(reader, "apply_actions_and_regenerate_grouped") as render,
):
self.assertEqual(reader.run(workspace, refaire=True), ExitCode.PARTIAL)
render.assert_not_called()
def test_regeneration_preserves_untouched_image_and_score_and_saves_redo(self):
for mode in ("BGnot", "Bnot", "Anot"):
with self.subTest(mode=mode), tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
workspace = self.workspace(root, mode)
output = root / mode / "Copie01"
output.mkdir()
Image.new("RGB", (40, 30), "red").save(output / "Ex 1.jpg")
before = (output / "Ex 1.jpg").read_bytes()
atomic_write_json(output / "score.json", {"Ex 1": "3.5", "Ex 2": "0"})
answer = root / "answer.pdf"
answer.touch()
data = {
"01": {
label: {
"result": {"score": 2, "feedback": []},
"pdf_path": answer,
"coordinates": (0, 0),
}
for label in ("Ex 1", "Ex 2")
}
}
with (
patch.object(
reader.annotating,
"make_base_image",
return_value=(Image.new("RGB", (40, 20)), 0, 0),
),
patch.object(
reader.annotating,
"compose_label_image",
return_value=(Image.new("RGB", (40, 20), "blue"), 0),
),
patch.object(reader, "get_extra_pdfs_as_images", return_value=[]),
):
status, _ = reader.apply_actions_and_regenerate_grouped(
workspace,
data,
"01",
[],
{},
["Ex 1", "Ex 2"],
annotation_dir=mode,
selected_labels={"Ex 2"},
)
self.assertEqual(status, ExitCode.SUCCESS)
self.assertEqual(
read_json(output / "score.json"), {"Ex 1": "3.5", "Ex 2": "2"}
)
self.assertEqual((output / "Ex 1.jpg").read_bytes(), before)
self.assertTrue((output / "Ex 2.jpg").is_file())
with Image.open(output / "Concat.jpg") as concat:
self.assertEqual(concat.size, (40, 50))
def test_simple_import_survives_successive_redos(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
workspace = self.workspace(root, "Anot")
output = root / "Anot/Copie01"
output.mkdir()
for label in ("Ex 1", "Ex 2"):
Image.new("RGB", (40, 30), "white").save(output / f"{label}.jpg")
Image.new("RGB", (40, 60), "red").save(output / "Concat_annotated.jpg")
atomic_write_json(output / "score.json", {"Ex 1": "1", "Ex 2": "1"})
answer = root / "answer.pdf"
answer.touch()
data = {
"01": {
label: {
"result": {"score": 2, "feedback": []},
"pdf_path": answer,
"coordinates": (0, 0),
}
for label in ("Ex 1", "Ex 2")
}
}
with (
patch.object(
reader.annotating,
"make_base_image",
return_value=(Image.new("RGB", (40, 20)), 0, 0),
),
patch.object(
reader.annotating,
"compose_label_image",
return_value=(Image.new("RGB", (40, 20), "blue"), 0),
),
):
for selected in ({"Ex 2"}, {"Ex 1"}):
status, _ = reader.apply_actions_and_regenerate_grouped(
workspace,
data,
"01",
[],
{},
["Ex 1", "Ex 2"],
annotation_dir="Anot",
selected_labels=selected,
)
self.assertEqual(status, ExitCode.SUCCESS)
if selected == {"Ex 2"}:
with Image.open(output / "Ex 1.jpg") as untouched:
self.assertGreater(untouched.getpixel((10, 10))[0], 240)
with Image.open(output / "Concat.jpg") as concat:
self.assertEqual(concat.size, (40, 40))
self.assertGreater(concat.getpixel((10, 30))[2], 240)
def test_latest_group_coordinates_win_numerically(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
group = root / "Par label" / "Ex 1"
group.mkdir(parents=True)
for number, height in ((2, 100), (10, 40)):
atomic_write_json(
group / f"Group_{number}.json", [["01", 0, height, "", "Ex 1"]]
)
Image.new("RGB", (20, height)).save(group / f"Group_{number}.jpg")
index, warnings = _coordinate_index(EvaluationWorkspace(root))
self.assertFalse(warnings)
self.assertEqual(index[("Ex 1", "01")].height, 40)
def test_unreadable_redo_is_rejected(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
atomic_write_json(root / "bnote.json", {"images": []})
with (
patch.object(
reader, "detect_checks_and_notes", return_value=([], None)
),
self.assertRaises(ValueError),
):
reader._scan_annotation_directory(root, required=True)
if __name__ == "__main__":
unittest.main()
+205
View File
@@ -0,0 +1,205 @@
from __future__ import annotations
import copy
import shutil
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from PIL import Image
from copienator import EvaluationWorkspace, ExitCode, atomic_write_json, read_json
from copienator.annotation_data import AnnotationLoadResult
from copienator.commands import annotating_with_checks as checks
from copienator.commands import export, import_annotations
from copienator.commands import reading_grouped_annotations as reader
from copienator_gui.refaire_sessions import begin_pass
class RefaireSessionTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.root = Path(self.temp.name) / "Exam"
self.root.mkdir()
self.workspace = EvaluationWorkspace(self.root)
atomic_write_json(self.workspace.refaire_file, [["Copie01", ["Ex 1"]]])
(self.root / "BRnot/Copie01").mkdir(parents=True)
(self.root / "BRnot/Copie01/Concat_annotated.pdf").write_bytes(b"legacy return")
self.state = {
"steps": {
"annotation": {"status": "success"},
"refaire_selection": {
"status": "success",
"values": {
"selection": [["Copie01", ["Ex 1"]]],
"annotation_dir": "Anot",
"layout": "grouped",
},
},
"refaire_merge": {"status": "success"},
},
"history": [],
}
atomic_write_json(self.workspace.gui_state_file, self.state)
def tearDown(self):
self.temp.cleanup()
def test_new_then_repeat_preserve_each_pass_and_reset_only_redo_progress(self):
state, first = begin_pass(self.workspace, self.state, keep_selection=False)
self.assertEqual(
state["steps"]["annotation"], self.state["steps"]["annotation"]
)
self.assertNotIn("refaire_merge", state["steps"])
self.assertEqual(state["steps"]["refaire_selection"]["values"]["selection"], [])
self.assertEqual(
state["steps"]["refaire_selection"]["values"]["layout"], "grouped"
)
self.assertEqual(read_json(self.workspace.refaire_file), [])
archives = [
path for path in (self.root / "Reprises").iterdir() if path.name != first
]
self.assertEqual(len(archives), 1)
self.assertEqual(
(archives[0] / "BRnot/Copie01/Concat_annotated.pdf").read_bytes(),
b"legacy return",
)
selection = [["Copie02", ["Ex 2"]]]
atomic_write_json(self.workspace.refaire_file, selection)
first_output = self.workspace.annotation_dir("refaire")
first_output.mkdir(parents=True)
(first_output / "review.pdf").write_bytes(b"first review")
state["steps"]["refaire_merge"] = {"status": "success"}
updated, second = begin_pass(self.workspace, state, keep_selection=True)
self.assertNotEqual(first, second)
self.assertEqual(read_json(self.workspace.refaire_file), selection)
self.assertEqual(
updated["steps"]["refaire_selection"]["values"]["selection"], selection
)
self.assertEqual((first_output / "review.pdf").read_bytes(), b"first review")
self.assertEqual(read_json(first_output.parent / "refaire.json"), selection)
self.assertNotIn("status", updated["steps"]["refaire_selection"])
self.assertEqual(
self.workspace.annotation_dir("refaire"),
self.root / "Reprises" / second / "BRnot",
)
reloaded = EvaluationWorkspace(self.root)
self.assertEqual(reloaded.refaire_session_id, second)
def test_failed_activation_leaves_current_pass_and_state_unchanged(self):
before = copy.deepcopy(self.state)
with (
patch(
"copienator_gui.refaire_sessions.staged_files",
side_effect=OSError("disk unavailable"),
),
self.assertRaises(OSError),
):
begin_pass(self.workspace, self.state, keep_selection=False)
self.assertIsNone(self.workspace.refaire_session_id)
self.assertEqual(read_json(self.workspace.gui_state_file), before)
self.assertEqual(
read_json(self.workspace.refaire_file), [["Copie01", ["Ex 1"]]]
)
self.assertEqual(self.state, before)
self.assertEqual(
(self.root / "BRnot/Copie01/Concat_annotated.pdf").read_bytes(),
b"legacy return",
)
def test_exports_are_unique_and_old_returns_cannot_enter_new_pass(self):
export_root = Path(self.temp.name) / "Export"
import_root = Path(self.temp.name) / "Import"
import_root.mkdir()
state, first = begin_pass(self.workspace, self.state, keep_selection=True)
for expected_id in (first, None):
if expected_id is None:
state, second = begin_pass(self.workspace, state, keep_selection=True)
expected_id = second
output = self.workspace.annotation_dir("refaire") / "Ex 1 G1"
output.mkdir(parents=True)
(output / "Concat.pdf").write_bytes(expected_id.encode())
with patch.object(export, "EXPORT_DIR", export_root):
self.assertEqual(
export.run(self.workspace, refaire=True), ExitCode.SUCCESS
)
exported = (
export_root / "Exam" / expected_id / f"{expected_id}__Ex 1 G1.pdf"
)
self.assertEqual(exported.read_bytes(), expected_id.encode())
if expected_id == first:
(import_root / exported.name).write_bytes(b"old return")
current = self.workspace.annotation_dir("refaire") / "Ex 1 G1"
with patch.object(import_annotations, "IMPORT_DIR", import_root):
self.assertEqual(
import_annotations.run(self.workspace, refaire=True), ExitCode.PARTIAL
)
self.assertFalse((current / "Concat_annotated.pdf").exists())
(import_root / f"{second}__Ex 1 G1.pdf").write_bytes(b"new return")
self.assertEqual(
import_annotations.run(self.workspace, refaire=True), ExitCode.SUCCESS
)
self.assertEqual((current / "Concat_annotated.pdf").read_bytes(), b"new return")
self.workspace.require_directories("BRnot")
def test_generation_and_merge_follow_the_active_pass(self):
for directory in ("Copies", "Par label", "Anot"):
(self.root / directory).mkdir()
(self.root / "labels").write_text("Ex 1\n")
atomic_write_json(self.root / "correction.json", {})
answer = self.root / "answer.pdf"
answer.touch()
loaded = AnnotationLoadResult(
{
"01": {
"Ex 1": {
"pdf_path": answer,
"result": {"score": 2, "feedback": []},
"coordinates": (0, 0),
}
}
},
[],
)
begin_pass(self.workspace, self.state, keep_selection=True)
with (
patch.object(checks, "load_annotation_data", return_value=loaded),
patch.object(reader, "load_annotation_data", return_value=loaded),
patch.object(
checks.annotating,
"make_base_image",
return_value=(Image.new("RGB", (100, 100), "white"), 0, 0),
),
patch.object(
checks.annotating,
"compose_label_image",
return_value=(Image.new("RGB", (100, 100), "white"), 0),
),
):
self.assertEqual(
checks.run(self.workspace, self.root, refaire=True, overwrite=True),
ExitCode.SUCCESS,
)
output = self.workspace.annotation_dir("refaire") / "Copie01"
shutil.copy2(output / "Concat.pdf", output / "Concat_annotated.pdf")
self.assertEqual(
reader.run(self.workspace, refaire=True, annotation_dir="Anot"),
ExitCode.SUCCESS,
)
self.assertEqual(
read_json(self.root / "Anot/Copie01/score.json"), {"Ex 1": "2"}
)
self.assertEqual(
(self.root / "BRnot/Copie01/Concat_annotated.pdf").read_bytes(),
b"legacy return",
)
def test_invalid_session_id_cannot_escape_workspace(self):
atomic_write_json(self.root / "refaire-session.json", {"id": "../elsewhere"})
with self.assertRaises(ValueError):
self.workspace.annotation_dir("refaire")
if __name__ == "__main__":
unittest.main()