258 lines
7.6 KiB
Python
258 lines
7.6 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",
|
|
)
|
|
return path.is_dir() and any((path / marker).exists() for marker in markers)
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return self.root.name
|
|
|
|
@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 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.root / 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.root / 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"
|