117 lines
4.3 KiB
Python
117 lines
4.3 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from threading import RLock
|
|
|
|
from app.actuators.models import (
|
|
ActuatorRecord,
|
|
LifecycleStatus,
|
|
ModelLifecycleState,
|
|
ReconciliationState,
|
|
model_id_for_actuator,
|
|
)
|
|
|
|
|
|
class ActuatorStore:
|
|
def __init__(self, root: str | Path) -> None:
|
|
self._root = Path(root).resolve()
|
|
self._actuators_root = self._root / "actuators"
|
|
self._actuators_root.mkdir(parents=True, exist_ok=True)
|
|
self._lock = RLock()
|
|
self._reconciliation_state_path = self._root / "reconciliation_state.json"
|
|
|
|
def list(self) -> list[ActuatorRecord]:
|
|
with self._lock:
|
|
return [self._load(path) for path in sorted(self._actuators_root.glob("*.json"))]
|
|
|
|
def get(self, actuator_entity_id: str) -> ActuatorRecord:
|
|
with self._lock:
|
|
target = self._target(actuator_entity_id)
|
|
if not target.exists():
|
|
raise KeyError("Aktuator-Konfiguration nicht gefunden.")
|
|
return self._load(target)
|
|
|
|
def upsert(self, record: ActuatorRecord) -> ActuatorRecord:
|
|
with self._lock:
|
|
self._persist(record)
|
|
return record
|
|
|
|
def configure(self, actuator_entity_id: str, *, enabled: bool = True) -> ActuatorRecord:
|
|
with self._lock:
|
|
target = self._target(actuator_entity_id)
|
|
if target.exists():
|
|
record = self._load(target)
|
|
updated = record.model_copy(
|
|
update={
|
|
"enabled": enabled,
|
|
"updated_at": datetime.now(timezone.utc),
|
|
}
|
|
)
|
|
self._persist(updated)
|
|
return updated
|
|
record = ActuatorRecord(
|
|
actuator_entity_id=actuator_entity_id,
|
|
enabled=enabled,
|
|
lifecycle=ModelLifecycleState(
|
|
model_id=model_id_for_actuator(actuator_entity_id),
|
|
status=LifecycleStatus.PENDING_ASSIGNMENT,
|
|
),
|
|
)
|
|
self._persist(record)
|
|
return record
|
|
|
|
def delete(self, actuator_entity_id: str) -> None:
|
|
with self._lock:
|
|
target = self._target(actuator_entity_id)
|
|
if target.exists():
|
|
target.unlink()
|
|
|
|
def load_reconciliation_state(self) -> ReconciliationState:
|
|
with self._lock:
|
|
if not self._reconciliation_state_path.exists():
|
|
return ReconciliationState()
|
|
try:
|
|
return ReconciliationState.model_validate_json(
|
|
self._reconciliation_state_path.read_text(encoding="utf-8")
|
|
)
|
|
except ValueError as exc:
|
|
raise ValueError("Ungültiger Reconciliation-Status.") from exc
|
|
|
|
def save_reconciliation_state(self, state: ReconciliationState) -> ReconciliationState:
|
|
with self._lock:
|
|
self._persist_reconciliation_state(state)
|
|
return state
|
|
|
|
def _target(self, actuator_entity_id: str) -> Path:
|
|
if "." not in actuator_entity_id:
|
|
raise ValueError("Ungültige actuator_entity_id.")
|
|
safe_name = actuator_entity_id.replace(".", "__")
|
|
return self._actuators_root / f"{safe_name}.json"
|
|
|
|
def _persist(self, record: ActuatorRecord) -> None:
|
|
target = self._target(record.actuator_entity_id)
|
|
temporary = target.with_suffix(".json.tmp")
|
|
temporary.write_text(
|
|
json.dumps(record.model_dump(mode="json"), ensure_ascii=True, sort_keys=True) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
os.replace(temporary, target)
|
|
|
|
def _persist_reconciliation_state(self, state: ReconciliationState) -> None:
|
|
temporary = self._reconciliation_state_path.with_suffix(".json.tmp")
|
|
temporary.write_text(
|
|
json.dumps(state.model_dump(mode="json"), ensure_ascii=True, sort_keys=True) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
os.replace(temporary, self._reconciliation_state_path)
|
|
|
|
@staticmethod
|
|
def _load(path: Path) -> ActuatorRecord:
|
|
try:
|
|
return ActuatorRecord.model_validate_json(path.read_text(encoding="utf-8"))
|
|
except ValueError as exc:
|
|
raise ValueError(f"Ungültige Aktuator-Konfiguration: {path.name}") from exc
|