317 lines
10 KiB
Python
317 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
|
|
class WorkspaceNotFoundError(FileNotFoundError):
|
|
pass
|
|
|
|
|
|
class WorkspaceValidationError(ValueError):
|
|
def __init__(self, root: Path, missing: list[str]) -> None:
|
|
self.root = root
|
|
self.missing = missing
|
|
super().__init__(
|
|
f"Invalid evaluation workspace {root}: missing {', '.join(missing)}"
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class EvaluationWorkspace:
|
|
root: Path
|
|
repository: Path | None = None
|
|
|
|
def __post_init__(self) -> None:
|
|
object.__setattr__(self, "root", Path(self.root).expanduser().resolve())
|
|
if self.repository is not None:
|
|
object.__setattr__(
|
|
self, "repository", Path(self.repository).expanduser().resolve()
|
|
)
|
|
|
|
@classmethod
|
|
def discover(
|
|
cls,
|
|
target: str | Path,
|
|
*,
|
|
repository: str | Path | None = None,
|
|
) -> EvaluationWorkspace:
|
|
candidate = Path(target).expanduser().resolve()
|
|
if candidate.is_file():
|
|
candidate = candidate.parent
|
|
repository_path = Path(repository).expanduser().resolve() if repository else None
|
|
|
|
for directory in (candidate, *candidate.parents):
|
|
if cls.looks_like_evaluation(directory):
|
|
return cls(directory, repository_path)
|
|
if repository_path is not None and directory == repository_path:
|
|
break
|
|
raise WorkspaceNotFoundError(
|
|
f"Could not find an evaluation workspace from: {target}"
|
|
)
|
|
|
|
@staticmethod
|
|
def looks_like_evaluation(path: Path) -> bool:
|
|
markers = (
|
|
"enonce.pdf",
|
|
"enonce.tex",
|
|
"labels",
|
|
"Copies",
|
|
"correction.json",
|
|
".copienator",
|
|
".copienator-gui.json",
|
|
)
|
|
if not path.is_dir():
|
|
return False
|
|
workspace = EvaluationWorkspace(path)
|
|
inputs = (
|
|
candidate
|
|
for name in ("enonce.pdf", "enonce.tex", "correction.tex")
|
|
for candidate in workspace.input_candidates(name)
|
|
)
|
|
return any((path / marker).exists() for marker in markers) or any(
|
|
candidate.exists() for candidate in inputs
|
|
)
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return self.root.name
|
|
|
|
def input_candidates(self, canonical_name: str) -> tuple[Path, ...]:
|
|
"""Return preferred paths for one of the three statement inputs."""
|
|
aliases = {
|
|
"enonce.pdf": ("enonce.pdf", "énonce.pdf", "énoncé.pdf", f"{self.name}.pdf"),
|
|
"enonce.tex": ("enonce.tex", "énonce.tex", "énoncé.tex", f"{self.name}.tex"),
|
|
"correction.tex": ("correction.tex", "corrige.tex", "corrigé.tex", f"{self.name}c.tex"),
|
|
}
|
|
names = aliases.get(canonical_name, (canonical_name,))
|
|
candidates = tuple(self.root / name for name in names)
|
|
if canonical_name in aliases and self.root.is_dir():
|
|
folder_alias = names[-1].replace(" ", "")
|
|
candidates += tuple(
|
|
path
|
|
for path in sorted(self.root.iterdir())
|
|
if path.name.replace(" ", "") == folder_alias and path not in candidates
|
|
)
|
|
return candidates
|
|
|
|
def resolve(self, relative_path: str) -> Path:
|
|
"""Resolve a logical workspace path, including statement aliases."""
|
|
candidates = self.input_candidates(relative_path)
|
|
return next((path for path in candidates if path.exists()), candidates[0])
|
|
|
|
def resolve_file(self, relative_path: str) -> Path:
|
|
candidates = self.input_candidates(relative_path)
|
|
return next((path for path in candidates if path.is_file()), candidates[0])
|
|
|
|
@property
|
|
def statement_pdf_file(self) -> Path:
|
|
return self.resolve_file("enonce.pdf")
|
|
|
|
@property
|
|
def statement_source_file(self) -> Path:
|
|
return self.resolve_file("enonce.tex")
|
|
|
|
@property
|
|
def correction_source_file(self) -> Path:
|
|
return self.resolve_file("correction.tex")
|
|
|
|
def is_statement_pdf(self, path: Path) -> bool:
|
|
return path.parent == self.root and path.name.replace(" ", "") in {
|
|
candidate.name.replace(" ", "")
|
|
for candidate in self.input_candidates("enonce.pdf")
|
|
}
|
|
|
|
@property
|
|
def metadata_dir(self) -> Path:
|
|
return self.root / ".copienator"
|
|
|
|
@property
|
|
def logs_dir(self) -> Path:
|
|
return self.metadata_dir / "logs"
|
|
|
|
@property
|
|
def runs_dir(self) -> Path:
|
|
return self.metadata_dir / "runs"
|
|
|
|
@property
|
|
def state_database(self) -> Path:
|
|
return self.metadata_dir / "state.sqlite3"
|
|
|
|
@property
|
|
def gui_state_file(self) -> Path:
|
|
return self.root / ".copienator-gui.json"
|
|
|
|
@property
|
|
def labels_file(self) -> Path:
|
|
return self.root / "labels"
|
|
|
|
@property
|
|
def label_groups_file(self) -> Path:
|
|
return self.root / "label_groups"
|
|
|
|
@property
|
|
def gemini_exam_items_file(self) -> Path:
|
|
return self.root / "Tmp" / "exam_items.txt"
|
|
|
|
@property
|
|
def correction_file(self) -> Path:
|
|
return self.root / "correction.json"
|
|
|
|
@property
|
|
def correction_progress_file(self) -> Path:
|
|
return self.root / "correction_progress.json"
|
|
|
|
@property
|
|
def correction_pending_responses_file(self) -> Path:
|
|
return self.root / "correction_pending_responses.json"
|
|
|
|
@property
|
|
def batch_jobs_file(self) -> Path:
|
|
return self.root / "batch_jobs.json"
|
|
|
|
@property
|
|
def batched_correction_result_file(self) -> Path:
|
|
return self.root / "batched_correction_result.jsonl"
|
|
|
|
@property
|
|
def manual_resolutions_file(self) -> Path:
|
|
return self.root / "manual_resolutions.txt"
|
|
|
|
@property
|
|
def refaire_file(self) -> Path:
|
|
return self.root / "refaire.json"
|
|
|
|
@property
|
|
def copies_dir(self) -> Path:
|
|
return self.root / "Copies"
|
|
|
|
@property
|
|
def original_copies_dir(self) -> Path:
|
|
return self.root / "Copies Originales"
|
|
|
|
@property
|
|
def cutleft_dir(self) -> Path:
|
|
return self.root / "Cutleft"
|
|
|
|
@property
|
|
def groups_dir(self) -> Path:
|
|
return self.root / "Par label"
|
|
|
|
@property
|
|
def text_dir(self) -> Path:
|
|
return self.root / "Text"
|
|
|
|
@property
|
|
def solution_dir(self) -> Path:
|
|
return self.root / "Sol"
|
|
|
|
@property
|
|
def rendered_text_dir(self) -> Path:
|
|
return self.root / "Text2"
|
|
|
|
@property
|
|
def rendered_solution_dir(self) -> Path:
|
|
return self.root / "Sol2"
|
|
|
|
@property
|
|
def rubric_dir(self) -> Path:
|
|
return self.root / "Persp"
|
|
|
|
@property
|
|
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",
|
|
"grouped": "BGnot",
|
|
"refaire": "BRnot",
|
|
}
|
|
try:
|
|
return self.root / directories[mode]
|
|
except KeyError as exc:
|
|
choices = ", ".join(directories)
|
|
raise ValueError(
|
|
f"Unknown annotation mode {mode!r}; expected one of: {choices}"
|
|
) from exc
|
|
|
|
def names_file(self) -> Path:
|
|
local = self.root / "names"
|
|
if local.exists() or self.repository is None:
|
|
return local
|
|
return self.repository / "names"
|
|
|
|
def read_labels(self) -> list[str]:
|
|
return [
|
|
line.strip()
|
|
for line in self.labels_file.read_text(encoding="utf-8").splitlines()
|
|
if line.strip()
|
|
]
|
|
|
|
def require(self, *relative_paths: str) -> None:
|
|
missing = [
|
|
relative_path
|
|
for relative_path in relative_paths
|
|
if not self.resolve(relative_path).exists()
|
|
]
|
|
if missing:
|
|
raise WorkspaceValidationError(self.root, missing)
|
|
|
|
def require_files(self, *relative_paths: str) -> None:
|
|
missing = [
|
|
relative_path
|
|
for relative_path in relative_paths
|
|
if not self.resolve_file(relative_path).is_file()
|
|
]
|
|
if missing:
|
|
raise WorkspaceValidationError(self.root, missing)
|
|
|
|
def require_directories(self, *relative_paths: str) -> None:
|
|
missing = [
|
|
relative_path
|
|
for relative_path in relative_paths
|
|
if not (self.annotation_dir("refaire") if relative_path == "BRnot" else self.root / relative_path).is_dir()
|
|
]
|
|
if missing:
|
|
raise WorkspaceValidationError(self.root, missing)
|
|
|
|
def ensure_control_directories(self) -> None:
|
|
self.logs_dir.mkdir(parents=True, exist_ok=True)
|
|
self.runs_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
def command_argument(self) -> str:
|
|
if self.repository is not None:
|
|
try:
|
|
return str(self.root.relative_to(self.repository))
|
|
except ValueError:
|
|
pass
|
|
return str(self.root)
|
|
|
|
def log_path(self, step_id: str, when: datetime | None = None) -> Path:
|
|
timestamp = (when or datetime.now().astimezone()).strftime("%Y%m%d-%H%M%S")
|
|
safe_step = re.sub(r"[^A-Za-z0-9_.-]+", "_", step_id)
|
|
return self.logs_dir / f"{timestamp}-{safe_step}.log"
|