83 lines
2.9 KiB
Python
83 lines
2.9 KiB
Python
import json
|
|
from pathlib import Path
|
|
from threading import Lock
|
|
|
|
from led_platform.core.clock import epoch_ms
|
|
from led_platform.domain import Scene, SceneState
|
|
|
|
|
|
class SceneStore:
|
|
def __init__(self, config_path: Path) -> None:
|
|
self.config_path = config_path
|
|
self._lock = Lock()
|
|
self._scenes: dict[str, Scene] = {}
|
|
self._state: SceneState | None = None
|
|
self.reload()
|
|
|
|
def reload(self) -> None:
|
|
data = json.loads(self.config_path.read_text(encoding="utf-8"))
|
|
scenes = [Scene(**item) for item in data["scenes"]]
|
|
if not scenes:
|
|
raise ValueError("At least one scene is required.")
|
|
default_scene_id = data.get("default_scene_id") or scenes[0].id
|
|
scene_map = {scene.id: scene for scene in scenes}
|
|
if default_scene_id not in scene_map:
|
|
raise ValueError(f"Default scene {default_scene_id!r} does not exist.")
|
|
default_scene = scene_map[default_scene_id]
|
|
with self._lock:
|
|
version = self._state.version + 1 if self._state else 1
|
|
self._scenes = scene_map
|
|
self._state = self._build_state(default_scene, version=version, apply_at_ms=None)
|
|
|
|
def list(self) -> list[Scene]:
|
|
with self._lock:
|
|
return list(self._scenes.values())
|
|
|
|
def get(self, scene_id: str) -> Scene:
|
|
with self._lock:
|
|
if scene_id not in self._scenes:
|
|
raise KeyError(scene_id)
|
|
return self._scenes[scene_id]
|
|
|
|
def state(self) -> SceneState:
|
|
with self._lock:
|
|
if self._state is None:
|
|
raise RuntimeError("Scene state is not initialized.")
|
|
return self._state
|
|
|
|
def switch(self, scene_id: str, apply_at_ms: int) -> SceneState:
|
|
with self._lock:
|
|
if scene_id not in self._scenes:
|
|
raise KeyError(scene_id)
|
|
version = self._state.version + 1 if self._state else 1
|
|
self._state = self._build_state(
|
|
self._scenes[scene_id],
|
|
version=version,
|
|
apply_at_ms=apply_at_ms,
|
|
)
|
|
return self._state
|
|
|
|
def action(self, action: str, apply_at_ms: int) -> SceneState:
|
|
with self._lock:
|
|
if self._state is None:
|
|
raise RuntimeError("Scene state is not initialized.")
|
|
self._state = self._state.model_copy(
|
|
update={
|
|
"version": self._state.version + 1,
|
|
"action": action,
|
|
"apply_at_ms": apply_at_ms,
|
|
}
|
|
)
|
|
return self._state
|
|
|
|
@staticmethod
|
|
def _build_state(scene: Scene, version: int, apply_at_ms: int | None) -> SceneState:
|
|
return SceneState(
|
|
active_scene_id=scene.id,
|
|
active_view=scene.view,
|
|
active_url=str(scene.url),
|
|
version=version,
|
|
scene_started_at_ms=apply_at_ms or epoch_ms(),
|
|
apply_at_ms=apply_at_ms,
|
|
)
|