65 lines
2.2 KiB
Python
65 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from copienator import EvaluationWorkspace, atomic_write_json, read_json
|
|
|
|
|
|
class StateStore:
|
|
def __init__(self) -> None:
|
|
self.evaluation: Path | None = None
|
|
self.workspace: EvaluationWorkspace | None = None
|
|
self.data: dict[str, Any] = self._empty_data()
|
|
|
|
@staticmethod
|
|
def _empty_data() -> dict[str, Any]:
|
|
return {"version": 1, "steps": {}, "history": []}
|
|
|
|
def load(self, evaluation: Path) -> None:
|
|
self.workspace = EvaluationWorkspace(evaluation)
|
|
self.evaluation = self.workspace.root
|
|
path = self.workspace.gui_state_file
|
|
try:
|
|
loaded = read_json(path, default=self._empty_data())
|
|
self.data = loaded if isinstance(loaded, dict) else self._empty_data()
|
|
except (OSError, json.JSONDecodeError):
|
|
self.data = self._empty_data()
|
|
self.data.setdefault("version", 1)
|
|
self.data.setdefault("steps", {})
|
|
self.data.setdefault("history", [])
|
|
|
|
def save(self) -> None:
|
|
if not self.workspace:
|
|
return
|
|
atomic_write_json(self.workspace.gui_state_file, self.data)
|
|
|
|
def step(self, step_id: str) -> dict[str, Any]:
|
|
return self.data["steps"].setdefault(step_id, {})
|
|
|
|
def update_step(self, step_id: str, **changes: Any) -> None:
|
|
entry = self.step(step_id)
|
|
entry.update(changes)
|
|
entry["updated_at"] = datetime.now(timezone.utc).isoformat()
|
|
self.save()
|
|
|
|
def add_history(self, entry: dict[str, Any]) -> None:
|
|
entry = dict(entry)
|
|
entry["timestamp"] = datetime.now(timezone.utc).isoformat()
|
|
self.data["history"].append(entry)
|
|
self.data["history"] = self.data["history"][-200:]
|
|
self.save()
|
|
|
|
def invalidate_after(self, ordered_ids: list[str], current_id: str) -> None:
|
|
try:
|
|
index = ordered_ids.index(current_id)
|
|
except ValueError:
|
|
return
|
|
for step_id in ordered_ids[index + 1 :]:
|
|
entry = self.step(step_id)
|
|
if entry.get("status") in {"success", "detected", "skipped"}:
|
|
entry["status"] = "stale"
|
|
self.save()
|