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)
56 lines
1.9 KiB
Python
56 lines
1.9 KiB
Python
import json
|
|
import aiofiles
|
|
from pathlib import Path
|
|
from lightsync.models.device import DeviceConfig
|
|
from lightsync.devices.base import BaseDevice
|
|
from lightsync.devices.sk6812 import SK6812Device
|
|
from lightsync.devices.ws2801 import WS2801Device
|
|
|
|
# Registry of strip_type -> device class (INF-02: add new type here only)
|
|
_DEVICE_CLASSES: dict[str, type[BaseDevice]] = {
|
|
"sk6812": SK6812Device,
|
|
"ws2801": WS2801Device,
|
|
}
|
|
|
|
|
|
class DeviceRegistry:
|
|
def __init__(self, path: Path):
|
|
self._path = path
|
|
self._devices: dict[str, DeviceConfig] = {} # id (str) -> config
|
|
|
|
async def load(self) -> None:
|
|
if self._path.exists():
|
|
async with aiofiles.open(self._path) as f:
|
|
raw = json.loads(await f.read())
|
|
self._devices = {
|
|
d["id"]: DeviceConfig.model_validate(d)
|
|
for d in raw.get("devices", [])
|
|
}
|
|
|
|
async def save(self) -> None:
|
|
data = {"devices": [d.model_dump(mode="json") for d in self._devices.values()]}
|
|
async with aiofiles.open(self._path, "w") as f:
|
|
await f.write(json.dumps(data, indent=2))
|
|
|
|
def add(self, config: DeviceConfig) -> None:
|
|
self._devices[str(config.id)] = config
|
|
|
|
def remove(self, device_id: str) -> bool:
|
|
return self._devices.pop(device_id, None) is not None
|
|
|
|
def get(self, device_id: str) -> DeviceConfig | None:
|
|
return self._devices.get(device_id)
|
|
|
|
def list_all(self) -> list[DeviceConfig]:
|
|
return list(self._devices.values())
|
|
|
|
def instantiate(self, device_id: str) -> BaseDevice | None:
|
|
"""Create a live BaseDevice instance from stored config."""
|
|
config = self.get(device_id)
|
|
if config is None:
|
|
return None
|
|
cls = _DEVICE_CLASSES.get(config.strip_type)
|
|
if cls is None:
|
|
raise ValueError(f"Unknown strip_type: {config.strip_type}")
|
|
return cls(config)
|