5e10272c69
- DeviceRegistry with load/save/add/remove/instantiate and _DEVICE_CLASSES dict
- ShowStore with save/load/list_ids using aiofiles for async I/O
- FastAPI app with lifespan hook (startup: load registry/store, shutdown: save registry)
- API routes: GET/POST /api/devices/, DELETE /api/devices/{id}, GET/POST /api/shows/, GET /api/shows/{id}
- WebSocket stub at /ws with ConnectionManager and ack echo
- Frontend shell: DAW layout with header, sidebar (DEVICES + ANIMATIONS), TIMELINE, TRANSPORT
- Terminal aesthetic: dark theme, cyan accents, monospace fonts, no drop shadows
- .gitignore for runtime data (devices.json, shows/, __pycache__, .venv)
30 lines
920 B
Python
30 lines
920 B
Python
import aiofiles
|
|
from pathlib import Path
|
|
from lightsync.models.show import ShowModel
|
|
|
|
|
|
class ShowStore:
|
|
def __init__(self, shows_dir: Path):
|
|
self._dir = shows_dir
|
|
|
|
def ensure_dir(self) -> None:
|
|
self._dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
def _path_for(self, show_id: str) -> Path:
|
|
return self._dir / f"{show_id}.json"
|
|
|
|
async def save(self, show: ShowModel) -> None:
|
|
path = self._path_for(str(show.id))
|
|
async with aiofiles.open(path, "w") as f:
|
|
await f.write(show.model_dump_json(indent=2))
|
|
|
|
async def load(self, show_id: str) -> ShowModel | None:
|
|
path = self._path_for(show_id)
|
|
if not path.exists():
|
|
return None
|
|
async with aiofiles.open(path) as f:
|
|
return ShowModel.model_validate_json(await f.read())
|
|
|
|
def list_ids(self) -> list[str]:
|
|
return [p.stem for p in self._dir.glob("*.json")]
|