feat(01-01): create device registry, show store, FastAPI app, API routes, and frontend shell
- 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)
This commit is contained in:
55
lightsync/devices/registry.py
Normal file
55
lightsync/devices/registry.py
Normal file
@@ -0,0 +1,55 @@
|
||||
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)
|
||||
Reference in New Issue
Block a user