diff --git a/TODO.md b/TODO.md index 6aa358a..43614c0 100644 --- a/TODO.md +++ b/TODO.md @@ -1,5 +1,30 @@ # LedGrab TODO +## BLE LED Controller Support (SP110E / Triones / Zengge / Govee) + +Add support for Bluetooth Low Energy LED controllers driven by mobile apps like "LED Hue", HappyLighting, iLightsIn. Whole-strip ambient-color output only — these protocols don't support per-pixel streaming. + +- [x] Add `bleak>=0.22` as optional extra `[ble]` in `server/pyproject.toml` (desktop-only, NOT in android `build.gradle.kts`) +- [x] `core/devices/ble_transport.py` — bleak wrapper: scan, connect, write-with/without-response +- [x] `core/devices/ble_protocols/` package + - [x] `__init__.py` — `BLEProtocol` dataclass + registry (family → encoder) + - [x] `sp110e.py` — SP110E / SP108E (service FFE0, char FFE1, `RR GG BB 00 1E` static-color frame) + - [x] `triones.py` — Triones / HappyLighting / LEDnet (service FFE5, char FFE9, `7E 07 05 03 RR GG BB 10 EF`) + - [x] `zengge.py` — Zengge / iLightsIn (service FFE0, framing `56 RR GG BB 00 F0 AA`) + - [x] `govee.py` — Govee unencrypted framed protocol (AES keyed variants — marked experimental) +- [x] `core/devices/ble_client.py` — unified `BLEClient(LEDClient)` — picks protocol by `ble_family`, averages strip → one color, drops duplicate frames, rate-limits to BLE connection interval +- [x] `core/devices/ble_provider.py` — `BLEDeviceProvider` + discovery via `BleakScanner` +- [x] Register in `core/devices/led_client.py::_register_builtin_providers` (guarded `try/except ImportError`) +- [x] Storage: `ble_family`, `ble_govee_key` fields threaded through `Device.__init__`/`to_dict`/`from_dict`/`_UPDATABLE_FIELDS`/`create_device` +- [x] Schemas: BLE fields on `DeviceCreate`, `DeviceUpdate`, `DeviceResponse` +- [x] Routes: BLE fields propagated through create/update in `api/routes/devices.py` + `_device_to_response` +- [x] ProcessorManager: `ble_family`/`ble_govee_key` added to `_DEVICE_FIELD_DEFAULTS` and `DeviceInfo`; passed through `wled_target_processor.py` and `group_client.py` to `create_led_client` +- [x] Tests: 21 protocol encoder unit tests + 16 BLEClient fake-transport tests — all passing, 814 total tests still green +- [x] Frontend: BLE option in the device type picker with a bluetooth Lucide icon; add-device modal shows a 4-option `IconSelect` for protocol family (SP110E / Triones / Zengge / Govee) with a Govee-only AES key field that auto-hides for the other three families; URL label/placeholder/hint adapt to `ble://
` pattern; submit payload carries `ble_family` (+ optional `ble_govee_key`); clone flow pre-fills family and key; modal dirty-check snapshots the new fields; network scan button now also discovers BLE peripherals via the existing `/api/v1/devices/discover?device_type=ble` endpoint +- [x] Frontend: `isBleDevice` helper in `core/api.ts`; `ICON_BLUETOOTH` + `ICON_LIGHTBULB` constants in `core/icons.ts`; `bluetooth` path in `core/icon-paths.ts`; i18n keys in `en.json` / `ru.json` / `zh.json`; TypeScript compiles; esbuild bundle rebuilt +- [x] Android BLE via Kotlin bridge — `BleBridge.kt` singleton (scan/connect/write/disconnect); `android_ble_transport.py` Python wrapper; `make_transport()` factory in `ble_transport.py` auto-selects backend; `BleBridge.init()` called from `LedGrabApp.onCreate`; BLE permissions in `AndroidManifest.xml` +- [x] Govee per-model AES key — `_encrypt_govee_frame()` in `ble_client.py` uses AES-128-ECB from `cryptography`; key validated on `BLEClient` construction; applied to both `send_pixels` and `set_power`; 8 new AES unit tests + ## Android — Restore Multi-ABI Wheels During emulator testing, we switched the build to **x86 only** (see `android/app/build.gradle.kts` `abiFilters`) to avoid having to keep the arm64-v8a / x86_64 pydantic-core wheels current. Before shipping, restore all three ABIs: @@ -75,3 +100,12 @@ Beyond the `/proc`-based AndroidMetricsProvider that's now in place: - [x] Device battery + thermal-zone readings (`/sys/class/power_supply/battery/{capacity,temp}`, `/sys/class/thermal/thermal_zone*/temp` filtered by zone type). Surfaced through `MetricsProvider.thermals()`, `PerformanceResponse.{cpu_temp_c,battery_percent,battery_temp_c}`, the metrics-history snapshot, and a new dashboard temperature chart that hides itself when the backend reports null. GPU card now hides (no "unavailable" placeholder) when no GPU is present. - [WONTDO] Optional: app-specific memory via `Debug.getMemoryInfo()` through a Kotlin → Python Chaquopy bridge (more accurate than `VmRSS` for split-app-process accounting) - [WONTDO] Optional: GPU usage via `/sys/class/kgsl/kgsl-3d0/gpubusy` on Adreno, Mali-specific paths for Mali GPUs + +## Refactor: Per-Provider Device Configs + +Replace flat `DeviceInfo` + `**kwargs` provider contract with a discriminated union of typed per-provider config dataclasses. Full plan: [docs/plans/device-typed-configs.md](docs/plans/device-typed-configs.md). + +- [x] Phase 1 — `DeviceConfig` hierarchy + `Device.to_config()` (non-breaking, additive only) +- [x] Phases 2+3 — narrow `LEDDeviceProvider.create_client` to typed configs; migrate 3 call sites; delete `DeviceInfo` + `_get_device_info` + `_DEVICE_FIELD_DEFAULTS` (single PR) +- [x] Phase 4 — migrate `tests/test_group_device.py` to `GroupConfig`/`ProviderDeps`; remove legacy `GroupLEDClient` init path; 47-test config suite with 100% coverage on `device_config.py` +- [ ] Phase 5 (separate PR, optional) — Pydantic v2 discriminated union in `api/schemas/devices.py`; scope frontend POST/PATCH payloads by `device_type` diff --git a/docs/plans/device-typed-configs.md b/docs/plans/device-typed-configs.md new file mode 100644 index 0000000..adf531c --- /dev/null +++ b/docs/plans/device-typed-configs.md @@ -0,0 +1,274 @@ +# Refactor Plan: Per-Provider Typed Device Configs + +**Status:** Planned, not started. +**Target branch:** `refactor/device-typed-configs` +**Intended executor:** Sonnet agent (one phase per invocation; human review between phases). + +## Goal + +Replace the flat [`DeviceInfo`](../../server/src/ledgrab/core/processing/target_processor.py) dataclass (and the `**kwargs`-based `LEDDeviceProvider.create_client(url, **kwargs)` contract) with a **discriminated union of per-provider config dataclasses**. Each provider owns its config type and reads typed fields instead of guessing kwargs. + +## Motivation + +Current pain points: + +- [server/src/ledgrab/core/processing/wled_target_processor.py](../../server/src/ledgrab/core/processing/wled_target_processor.py) unpacks ~21 fields by hand into `create_led_client(**kwargs)`. +- Every provider's `create_client` starts with `kwargs.get("x", default)` — no type safety, no IDE hints, no way to know at a glance which fields a provider actually uses. +- Adding a new per-device-type field requires threading it through `Device` → `DeviceInfo` → `_DEVICE_FIELD_DEFAULTS` → call-site unpacking → kwargs bag → provider. +- Fields leak across device types (a WLED device carries `ble_govee_key=""` at runtime for no reason). + +## Scope guardrails + +- **Storage schema (SQLite) unchanged.** Columns stay, dead-for-this-type fields stay, no destructive migration. +- **Frontend HTML/TS unchanged in phases 1-4.** It already branches on `device_type` with show/hide logic. Frontend changes are deferred to Phase 5. +- **API schemas are last.** Phase 5 converts `DeviceCreate`/`DeviceUpdate`/`DeviceResponse` to a Pydantic v2 discriminated union. This is the only breaking external change and can be deferred indefinitely if needed. + +--- + +## Phase 1 — Config hierarchy (foundation, non-breaking) + +### Create + +**File:** `server/src/ledgrab/core/devices/device_config.py` + +Pattern: + +```python +from dataclasses import dataclass +from typing import List, Literal, Optional, Union + +@dataclass(frozen=True) +class BaseDeviceConfig: + device_id: str + device_url: str + led_count: int + software_brightness: int = 255 + test_mode_active: bool = False + auto_shutdown: bool = False + rgbw: bool = False + +@dataclass(frozen=True) +class WLEDConfig(BaseDeviceConfig): + device_type: Literal["wled"] = "wled" + use_ddp: bool = False + +# ... one @dataclass(frozen=True) per provider +``` + +### Config field inventory + +Base: `device_id`, `device_url`, `led_count`, `software_brightness`, `test_mode_active`, `auto_shutdown`, `rgbw`. + +| Config | Extra fields beyond Base | +| -------------- | ------------------------ | +| WLEDConfig | `use_ddp: bool = False` | +| AdalightConfig | `baud_rate: Optional[int] = None` | +| AmbiLEDConfig | `baud_rate: Optional[int] = None` | +| DMXConfig | `dmx_protocol`, `dmx_start_universe`, `dmx_start_channel` | +| ESPNowConfig | `baud_rate`, `espnow_peer_mac`, `espnow_channel` | +| HueConfig | `hue_username`, `hue_client_key`, `hue_entertainment_group_id` | +| SPIConfig | `spi_speed_hz`, `spi_led_type` | +| ChromaConfig | `chroma_device_type` | +| GameSenseConfig| `gamesense_device_type` | +| BLEConfig | `ble_family`, `ble_govee_key` | +| GroupConfig | `group_mode`, `group_device_ids` (**no `device_store` here** — see Phase 2) | +| OpenRGBConfig | `zone_mode` | +| MockConfig | `send_latency_ms: int = 0` | +| DemoConfig | `send_latency_ms: int = 0` | +| MQTTConfig | (none) | +| WSConfig | (none) | +| USBHIDConfig | (none — `hid_usage_page` is parsed from the URL, not config) | + +```python +DeviceConfig = Union[ + WLEDConfig, AdalightConfig, AmbiLEDConfig, DMXConfig, ESPNowConfig, + HueConfig, SPIConfig, ChromaConfig, GameSenseConfig, BLEConfig, + GroupConfig, MQTTConfig, WSConfig, USBHIDConfig, OpenRGBConfig, + MockConfig, DemoConfig, +] +``` + +### Add + +**`Device.to_config() -> DeviceConfig`** in [server/src/ledgrab/storage/device_store.py](../../server/src/ledgrab/storage/device_store.py) (around lines 14-97 where `Device` lives). + +- Dispatches on `self.device_type`. +- Constructs the right subclass, pulling only relevant columns. +- Ignores columns that don't apply to the type. +- This is the **only** place that knows the flat→typed mapping. + +### Do NOT touch in Phase 1 + +- Provider signatures (still `create_client(self, url, **kwargs)`). +- `create_led_client` factory. +- Any call site. +- `DeviceInfo` itself. + +### Acceptance + +- New unit test `server/tests/core/devices/test_device_config.py`: + - For each provider, build a `Device` with that `device_type`, call `to_config()`, assert right subclass and right fields. + - Edge case: extra/irrelevant Device fields must not leak into the wrong config type. +- `cd server && ruff check src/ tests/ --fix` — green. +- `cd server && py -3.13 -m pytest tests/ --no-cov -q` — green (existing tests untouched, new test passes). +- `cd server && npx tsc --noEmit` — green (no TS impact this phase, just a sanity check). + +--- + +## Phase 2 + Phase 3 — Provider API migration + call-site migration (single PR) + +**These must land in one commit** because the provider signature change would otherwise break the 3 call sites immediately. + +### Change the abstract base + +[server/src/ledgrab/core/devices/led_client.py](../../server/src/ledgrab/core/devices/led_client.py): + +```python +class LEDDeviceProvider(ABC): + @abstractmethod + def create_client(self, config: DeviceConfig, *, deps: ProviderDeps) -> LEDClient: ... +``` + +`ProviderDeps` is a tiny new dataclass: + +```python +@dataclass(frozen=True) +class ProviderDeps: + device_store: "DeviceStore" + # Add future cross-cutting runtime deps here (http_client, etc.) +``` + +`create_led_client`: + +```python +def create_led_client(config: DeviceConfig, *, deps: ProviderDeps) -> LEDClient: + return get_provider(config.device_type).create_client(config, deps=deps) +``` + +### Update every provider (17 files) + +- Narrow signature per provider: e.g. `WLEDDeviceProvider.create_client(self, config: WLEDConfig, *, deps: ProviderDeps)`. +- Drop all `kwargs.get("x")` lookups — read typed fields directly. +- Providers that don't need `deps` just ignore it. +- **GroupDeviceProvider** is the only current consumer of `deps`: reads `deps.device_store`. + +### Call sites (3) + +1. [server/src/ledgrab/core/processing/wled_target_processor.py](../../server/src/ledgrab/core/processing/wled_target_processor.py) lines ~120-148 — the 21-field unpacking. Replace with: + ```python + config = device.to_config() + self._led_client = create_led_client(config, deps=self._provider_deps) + ``` + `self._provider_deps` is plumbed in from `ProcessorManager` when the target processor is constructed. +2. [server/src/ledgrab/core/processing/device_test_mode.py](../../server/src/ledgrab/core/processing/device_test_mode.py) lines 72-78 — minimal test-mode client. Build a synthetic config via a helper `_minimal_config_for_test_mode(device)` (keeps just `device_id`, `device_url`, `led_count`, `baud_rate`) and pass it. +3. [server/src/ledgrab/core/devices/group_client.py](../../server/src/ledgrab/core/devices/group_client.py) lines 47-70 — child client construction inside the group. Same pattern: `child_config = child_device.to_config()`; pass `deps` through. + +### Delete + +- `DeviceInfo` dataclass in [server/src/ledgrab/core/processing/target_processor.py](../../server/src/ledgrab/core/processing/target_processor.py) lines 71-109. +- `ProcessorManager._get_device_info()` and `_DEVICE_FIELD_DEFAULTS` in [server/src/ledgrab/core/processing/processor_manager.py](../../server/src/ledgrab/core/processing/processor_manager.py) lines 230-275 — `Device.to_config()` subsumes this. Verify no other callers via `ast-index usages "_get_device_info"`. + +### Acceptance + +- `ast-index search "device_info\."` — no hits in non-test code. +- `ast-index search "DeviceInfo"` — no hits outside archival comments. +- `cd server && py -3.13 -m pytest tests/ --no-cov -q` — all tests pass. +- Manual smoke: start server, create a WLED device, start processing, verify LEDs update (or mock output shows frames). +- `cd server && ruff check src/ tests/ --fix` — green. + +--- + +## Phase 4 — Test migration + +Update these files: + +- `server/tests/storage/test_device_store.py` — add `to_config()` cases per device type. +- `server/tests/api/routes/test_devices_routes.py` — should be mostly untouched (API schemas still flat until Phase 5). +- `server/tests/e2e/test_device_flow.py` — update internal assertions only if they touch `DeviceInfo` directly. +- `server/tests/test_group_device.py` — construct child clients with `GroupConfig`. +- Any fixture helper that builds a fake `DeviceInfo` — migrate to the right `*Config` subclass. + +### Acceptance + +- `cd server && py -3.13 -m pytest tests/ --no-cov -q` — all green. +- Coverage of `device_config.py` and `Device.to_config()` ≥ 90%. + +--- + +## Phase 5 — API discriminated union (OPTIONAL, separate PR) + +**Do not start until Phases 1-4 are merged and stable.** Flag this to the human before beginning. This is the only phase with an externally breaking change. + +### Backend + +[server/src/ledgrab/api/schemas/devices.py](../../server/src/ledgrab/api/schemas/devices.py) — replace flat `DeviceCreate`/`DeviceUpdate` with Pydantic v2 tagged unions: + +```python +class WLEDDeviceCreate(BaseModel): + device_type: Literal["wled"] + name: str + url: str + led_count: int + use_ddp: bool = False + # ... base fields only + +DeviceCreate = Annotated[ + Union[WLEDDeviceCreate, AdalightDeviceCreate, ...], + Field(discriminator="device_type"), +] +``` + +Add `model_config = ConfigDict(extra="ignore")` on each union member for **one release cycle** so existing clients (frontend, HAOS integration, curl scripts) that send extra fields don't 422 immediately. Add a deprecation note and tighten to `extra="forbid"` in a follow-up. + +### Frontend + +- [server/src/ledgrab/static/js/features/devices.ts](../../server/src/ledgrab/static/js/features/devices.ts) and related — when building the POST/PATCH body, scope the payload to the selected `device_type` using the show/hide knowledge already in `device-discovery.ts`. +- **No plain ``** — Phase 5 uses IconSelect / EntitySelect. +- **Android parity:** if you add any new runtime dep to `server/pyproject.toml`, update `android/app/build.gradle.kts` per the root [CLAUDE.md](../../CLAUDE.md) "Android Dependency Sync" section. This refactor should not need any new deps. +- **Data migration policy:** storage schema is unchanged, so no JSON-file migration is needed. But if you rename any serialized field during `to_dict`/`from_dict`, add migration logic per the root [CLAUDE.md](../../CLAUDE.md) "Data Migration Policy" section. +- **Use `ast-index`** for code search (`ast-index search`, `ast-index usages`, `ast-index callers`, `ast-index class`). Fall back to Grep only for regex/string-literal/comment searches. +- **Never run `cd` in Bash.** Use absolute paths or the project-relative `cd server && ` idiom (one-shot, same invocation). + +## Known risks + +1. **Frozen dataclass + inheritance + defaults** — Python's `@dataclass(frozen=True)` with inheritance requires every subclass field to have a default if any parent field does. Base has defaulted fields. Verify in Phase 1. If it breaks, use `kw_only=True` (Python 3.10+). +2. **`use_ddp` origin** — currently inferred from `self._protocol == "ddp"` at the call site, not from Device storage. Options: add a column (schema change, more work), **or** keep inference logic inside `Device.to_config()` (recommended — no schema change). Prefer the latter. +3. **Test-mode minimal client** ([device_test_mode.py](../../server/src/ledgrab/core/processing/device_test_mode.py) lines 72-78) may not have all `BaseDeviceConfig` fields available. Build a synthetic config via a named helper; do not leak the hack into `Device.to_config()`. +4. **Group `device_store` import cycle** — `GroupConfig` must **not** hold `device_store` (would pull storage into the config module). `ProviderDeps` is the deliberate cut. +5. **BLE optional import** — `BLEDeviceProvider` is conditionally registered (see [led_client.py](../../server/src/ledgrab/core/devices/led_client.py) lines 321-330). Ensure `BLEConfig` still imports cleanly even when `bleak` is absent — put `BLEConfig` in `device_config.py` (not in `ble_provider.py`) so it's always importable. + +## Deliverables per phase + +1. Branch: `refactor/device-typed-configs`. +2. One commit per phase, conventional-commit messages: + - `refactor(devices): phase 1 — add DeviceConfig hierarchy` + - `refactor(devices): phases 2+3 — typed provider signatures + call-site migration` + - `refactor(devices): phase 4 — test migration to typed configs` + - `refactor(devices): phase 5 — API discriminated union` (separate PR) +3. Phase-by-phase diffs presented for user review **before** each commit. +4. Final PR body linking all phases, with manual test plan per device type touched. diff --git a/server/src/ledgrab/core/devices/chroma_provider.py b/server/src/ledgrab/core/devices/chroma_provider.py index b6704c8..374ea08 100644 --- a/server/src/ledgrab/core/devices/chroma_provider.py +++ b/server/src/ledgrab/core/devices/chroma_provider.py @@ -1,12 +1,15 @@ """Razer Chroma SDK device provider — control Razer RGB peripherals.""" -from typing import List +from __future__ import annotations + +from typing import TYPE_CHECKING, List from ledgrab.core.devices.led_client import ( DeviceHealth, DiscoveredDevice, LEDClient, LEDDeviceProvider, + ProviderDeps, ) from ledgrab.core.devices.chroma_client import ( ChromaClient, @@ -15,6 +18,9 @@ from ledgrab.core.devices.chroma_client import ( ) from ledgrab.utils import get_logger +if TYPE_CHECKING: + from ledgrab.core.devices.device_config import ChromaConfig + logger = get_logger(__name__) @@ -33,11 +39,11 @@ class ChromaDeviceProvider(LEDDeviceProvider): def capabilities(self) -> set: return {"manual_led_count", "health_check"} - def create_client(self, url: str, **kwargs) -> LEDClient: - chroma_device_type = _parse_chroma_url(url) + def create_client(self, config: "ChromaConfig", *, deps: ProviderDeps) -> LEDClient: + chroma_device_type = _parse_chroma_url(config.device_url) return ChromaClient( url=CHROMA_SDK_URL, - led_count=kwargs.get("led_count", 0), + led_count=config.led_count, chroma_device_type=chroma_device_type, ) diff --git a/server/src/ledgrab/core/devices/demo_provider.py b/server/src/ledgrab/core/devices/demo_provider.py index 59db618..89a8812 100644 --- a/server/src/ledgrab/core/devices/demo_provider.py +++ b/server/src/ledgrab/core/devices/demo_provider.py @@ -1,7 +1,9 @@ """Demo device provider — virtual LED devices for demo mode.""" +from __future__ import annotations + from datetime import datetime, timezone -from typing import List +from typing import TYPE_CHECKING, List from ledgrab.config import is_demo_mode from ledgrab.core.devices.led_client import ( @@ -9,9 +11,13 @@ from ledgrab.core.devices.led_client import ( DiscoveredDevice, LEDClient, LEDDeviceProvider, + ProviderDeps, ) from ledgrab.core.devices.mock_client import MockClient +if TYPE_CHECKING: + from ledgrab.core.devices.device_config import DemoConfig + # Pre-defined virtual devices: (name, led_count, ip, width, height) _DEMO_DEVICES = [ ("Demo LED Strip", 60, "demo-strip", None, None), @@ -35,11 +41,11 @@ class DemoDeviceProvider(LEDDeviceProvider): def capabilities(self) -> set: return {"manual_led_count", "power_control", "brightness_control", "static_color"} - def create_client(self, url: str, **kwargs) -> LEDClient: + def create_client(self, config: "DemoConfig", *, deps: ProviderDeps) -> LEDClient: return MockClient( - url, - led_count=kwargs.get("led_count", 0), - send_latency_ms=kwargs.get("send_latency_ms", 0), + config.device_url, + led_count=config.led_count, + send_latency_ms=config.send_latency_ms, ) async def check_health(self, url: str, http_client, prev_health=None) -> DeviceHealth: diff --git a/server/src/ledgrab/core/devices/device_config.py b/server/src/ledgrab/core/devices/device_config.py new file mode 100644 index 0000000..55c8936 --- /dev/null +++ b/server/src/ledgrab/core/devices/device_config.py @@ -0,0 +1,148 @@ +"""Per-provider typed device config hierarchy. + +Each provider owns its config subclass. Device.to_config() is the single place +that maps the flat Device storage model to the right typed config. +""" + +from dataclasses import dataclass, field +from typing import List, Literal, Optional, Union + + +@dataclass(frozen=True) +class BaseDeviceConfig: + device_id: str + device_url: str + led_count: int + software_brightness: int = 255 + test_mode_active: bool = False + auto_shutdown: bool = False + rgbw: bool = False + + +@dataclass(frozen=True) +class WLEDConfig(BaseDeviceConfig): + device_type: Literal["wled"] = "wled" + use_ddp: bool = False + + +@dataclass(frozen=True) +class AdalightConfig(BaseDeviceConfig): + device_type: Literal["adalight"] = "adalight" + baud_rate: Optional[int] = None + + +@dataclass(frozen=True) +class AmbiLEDConfig(BaseDeviceConfig): + device_type: Literal["ambiled"] = "ambiled" + baud_rate: Optional[int] = None + + +@dataclass(frozen=True) +class DMXConfig(BaseDeviceConfig): + device_type: Literal["dmx"] = "dmx" + dmx_protocol: str = "artnet" + dmx_start_universe: int = 0 + dmx_start_channel: int = 1 + + +@dataclass(frozen=True) +class ESPNowConfig(BaseDeviceConfig): + device_type: Literal["espnow"] = "espnow" + baud_rate: Optional[int] = None + espnow_peer_mac: str = "" + espnow_channel: int = 1 + + +@dataclass(frozen=True) +class HueConfig(BaseDeviceConfig): + device_type: Literal["hue"] = "hue" + hue_username: str = "" + hue_client_key: str = "" + hue_entertainment_group_id: str = "" + + +@dataclass(frozen=True) +class SPIConfig(BaseDeviceConfig): + device_type: Literal["spi"] = "spi" + spi_speed_hz: int = 800000 + spi_led_type: str = "WS2812B" + + +@dataclass(frozen=True) +class ChromaConfig(BaseDeviceConfig): + device_type: Literal["chroma"] = "chroma" + chroma_device_type: str = "chromalink" + + +@dataclass(frozen=True) +class GameSenseConfig(BaseDeviceConfig): + device_type: Literal["gamesense"] = "gamesense" + gamesense_device_type: str = "keyboard" + + +@dataclass(frozen=True) +class BLEConfig(BaseDeviceConfig): + device_type: Literal["ble"] = "ble" + ble_family: str = "" + ble_govee_key: str = "" + + +@dataclass(frozen=True) +class GroupConfig(BaseDeviceConfig): + device_type: Literal["group"] = "group" + group_mode: str = "sequence" + group_device_ids: List[str] = field(default_factory=list) + + +@dataclass(frozen=True) +class OpenRGBConfig(BaseDeviceConfig): + device_type: Literal["openrgb"] = "openrgb" + zone_mode: str = "combined" + + +@dataclass(frozen=True) +class MockConfig(BaseDeviceConfig): + device_type: Literal["mock"] = "mock" + send_latency_ms: int = 0 + + +@dataclass(frozen=True) +class DemoConfig(BaseDeviceConfig): + device_type: Literal["demo"] = "demo" + send_latency_ms: int = 0 + + +@dataclass(frozen=True) +class MQTTConfig(BaseDeviceConfig): + device_type: Literal["mqtt"] = "mqtt" + + +@dataclass(frozen=True) +class WSConfig(BaseDeviceConfig): + device_type: Literal["ws"] = "ws" + + +@dataclass(frozen=True) +class USBHIDConfig(BaseDeviceConfig): + device_type: Literal["usbhid"] = "usbhid" + + +DeviceConfig = Union[ + WLEDConfig, + AdalightConfig, + AmbiLEDConfig, + DMXConfig, + ESPNowConfig, + HueConfig, + SPIConfig, + ChromaConfig, + GameSenseConfig, + BLEConfig, + GroupConfig, + MQTTConfig, + WSConfig, + USBHIDConfig, + OpenRGBConfig, + MockConfig, + DemoConfig, +] diff --git a/server/src/ledgrab/core/devices/dmx_provider.py b/server/src/ledgrab/core/devices/dmx_provider.py index 2639094..ec9a9fb 100644 --- a/server/src/ledgrab/core/devices/dmx_provider.py +++ b/server/src/ledgrab/core/devices/dmx_provider.py @@ -1,7 +1,9 @@ """DMX device provider — Art-Net / sACN (E1.31) factory, validation, health.""" +from __future__ import annotations + from datetime import datetime, timezone -from typing import List +from typing import TYPE_CHECKING, List from urllib.parse import urlparse from ledgrab.core.devices.led_client import ( @@ -9,10 +11,14 @@ from ledgrab.core.devices.led_client import ( DiscoveredDevice, LEDClient, LEDDeviceProvider, + ProviderDeps, ) from ledgrab.core.devices.dmx_client import DMXClient from ledgrab.utils import get_logger +if TYPE_CHECKING: + from ledgrab.core.devices.device_config import DMXConfig + logger = get_logger(__name__) @@ -48,15 +54,15 @@ class DMXDeviceProvider(LEDDeviceProvider): def capabilities(self) -> set: return {"manual_led_count"} - def create_client(self, url: str, **kwargs) -> LEDClient: - parsed = parse_dmx_url(url) + def create_client(self, config: "DMXConfig", *, deps: ProviderDeps) -> LEDClient: + parsed = parse_dmx_url(config.device_url) return DMXClient( host=parsed["host"], port=parsed["port"], - led_count=kwargs.get("led_count", 1), - protocol=kwargs.get("dmx_protocol", parsed["protocol"]), - start_universe=kwargs.get("dmx_start_universe", 0), - start_channel=kwargs.get("dmx_start_channel", 1), + led_count=config.led_count, + protocol=config.dmx_protocol, + start_universe=config.dmx_start_universe, + start_channel=config.dmx_start_channel, ) async def check_health( diff --git a/server/src/ledgrab/core/devices/espnow_provider.py b/server/src/ledgrab/core/devices/espnow_provider.py index cfe770f..0290cae 100644 --- a/server/src/ledgrab/core/devices/espnow_provider.py +++ b/server/src/ledgrab/core/devices/espnow_provider.py @@ -1,6 +1,8 @@ """ESP-NOW device provider — ultra-low-latency LED control via ESP32 gateway.""" -from typing import List +from __future__ import annotations + +from typing import TYPE_CHECKING, List from ledgrab.core.devices.espnow_client import ESPNowClient from ledgrab.core.devices.led_client import ( @@ -8,6 +10,7 @@ from ledgrab.core.devices.led_client import ( DiscoveredDevice, LEDClient, LEDDeviceProvider, + ProviderDeps, ) from ledgrab.core.devices.serial_transport import ( list_serial_ports, @@ -16,6 +19,9 @@ from ledgrab.core.devices.serial_transport import ( ) from ledgrab.utils import get_logger +if TYPE_CHECKING: + from ledgrab.core.devices.device_config import ESPNowConfig + logger = get_logger(__name__) # Description fragments that commonly indicate an ESP32 gateway's USB bridge. @@ -39,13 +45,13 @@ class ESPNowDeviceProvider(LEDDeviceProvider): def capabilities(self) -> set: return {"manual_led_count", "health_check"} - def create_client(self, url: str, **kwargs) -> LEDClient: + def create_client(self, config: "ESPNowConfig", *, deps: ProviderDeps) -> LEDClient: return ESPNowClient( - url, - led_count=kwargs.get("led_count", 0), - baud_rate=kwargs.get("baud_rate"), - espnow_peer_mac=kwargs.get("espnow_peer_mac", "FF:FF:FF:FF:FF:FF"), - espnow_channel=kwargs.get("espnow_channel", 1), + config.device_url, + led_count=config.led_count, + baud_rate=config.baud_rate, + espnow_peer_mac=config.espnow_peer_mac or "FF:FF:FF:FF:FF:FF", + espnow_channel=config.espnow_channel, ) async def check_health(self, url: str, http_client, prev_health=None) -> DeviceHealth: diff --git a/server/src/ledgrab/core/devices/gamesense_provider.py b/server/src/ledgrab/core/devices/gamesense_provider.py index a289026..381fe74 100644 --- a/server/src/ledgrab/core/devices/gamesense_provider.py +++ b/server/src/ledgrab/core/devices/gamesense_provider.py @@ -1,12 +1,15 @@ """SteelSeries GameSense device provider — control SteelSeries RGB peripherals.""" -from typing import List +from __future__ import annotations + +from typing import TYPE_CHECKING, List from ledgrab.core.devices.led_client import ( DeviceHealth, DiscoveredDevice, LEDClient, LEDDeviceProvider, + ProviderDeps, ) from ledgrab.core.devices.gamesense_client import ( GameSenseClient, @@ -15,6 +18,9 @@ from ledgrab.core.devices.gamesense_client import ( ) from ledgrab.utils import get_logger +if TYPE_CHECKING: + from ledgrab.core.devices.device_config import GameSenseConfig + logger = get_logger(__name__) @@ -33,11 +39,11 @@ class GameSenseDeviceProvider(LEDDeviceProvider): def capabilities(self) -> set: return {"manual_led_count", "health_check"} - def create_client(self, url: str, **kwargs) -> LEDClient: + def create_client(self, config: "GameSenseConfig", *, deps: ProviderDeps) -> LEDClient: return GameSenseClient( - url=url, - led_count=kwargs.get("led_count", 0), - gamesense_device_type=kwargs.get("gamesense_device_type", "keyboard"), + url=config.device_url, + led_count=config.led_count, + gamesense_device_type=config.gamesense_device_type, ) async def check_health(self, url: str, http_client, prev_health=None) -> DeviceHealth: diff --git a/server/src/ledgrab/core/devices/group_client.py b/server/src/ledgrab/core/devices/group_client.py index 6f1e1f3..181a0c5 100644 --- a/server/src/ledgrab/core/devices/group_client.py +++ b/server/src/ledgrab/core/devices/group_client.py @@ -3,13 +3,16 @@ from __future__ import annotations import asyncio -from typing import List, Optional, Tuple, Union +from typing import TYPE_CHECKING, List, Optional, Tuple, Union import numpy as np -from ledgrab.core.devices.led_client import LEDClient, create_led_client +from ledgrab.core.devices.led_client import LEDClient, ProviderDeps, create_led_client from ledgrab.utils import get_logger +if TYPE_CHECKING: + from ledgrab.core.devices.device_config import GroupConfig + logger = get_logger(__name__) @@ -20,19 +23,12 @@ class GroupLEDClient(LEDClient): Independent mode: resamples the full pixel array to each child's LED count. """ - def __init__( - self, - device_store, - device_id: str, - group_mode: str = "sequence", - group_device_ids: Optional[List[str]] = None, - **kwargs, - ): - self._device_store = device_store - self._device_id = device_id - self._group_mode = group_mode - self._group_device_ids = group_device_ids or [] - self._kwargs = kwargs + def __init__(self, config: "GroupConfig", deps: "ProviderDeps"): + self._device_store = deps.device_store + self._deps = deps + self._device_id = config.device_id + self._group_mode = config.group_mode + self._group_device_ids = list(config.group_device_ids) # Populated on connect() self._children: List[Tuple[LEDClient, int]] = [] # (client, led_count) self._connected = False @@ -44,32 +40,8 @@ class GroupLEDClient(LEDClient): try: for child_id in self._group_device_ids: device = self._device_store.get_device(child_id) - client = create_led_client( - device.device_type, - device.url, - led_count=device.led_count, - baud_rate=device.baud_rate, - send_latency_ms=device.send_latency_ms, - rgbw=device.rgbw, - zone_mode=device.zone_mode, - dmx_protocol=device.dmx_protocol, - dmx_start_universe=device.dmx_start_universe, - dmx_start_channel=device.dmx_start_channel, - espnow_peer_mac=device.espnow_peer_mac, - espnow_channel=device.espnow_channel, - hue_username=device.hue_username, - hue_client_key=device.hue_client_key, - hue_entertainment_group_id=device.hue_entertainment_group_id, - spi_speed_hz=device.spi_speed_hz, - spi_led_type=device.spi_led_type, - chroma_device_type=device.chroma_device_type, - gamesense_device_type=device.gamesense_device_type, - # Pass through for nested groups - device_store=self._device_store, - device_id=child_id, - group_mode=device.group_mode, - group_device_ids=device.group_device_ids, - ) + child_config = device.to_config() + client = create_led_client(child_config, deps=self._deps) await client.connect() connected_clients.append(client) diff --git a/server/src/ledgrab/core/devices/group_provider.py b/server/src/ledgrab/core/devices/group_provider.py index df26d15..421fccb 100644 --- a/server/src/ledgrab/core/devices/group_provider.py +++ b/server/src/ledgrab/core/devices/group_provider.py @@ -1,16 +1,22 @@ """Group device provider — virtual device that aggregates multiple child devices.""" +from __future__ import annotations + from datetime import datetime, timezone -from typing import List +from typing import TYPE_CHECKING, List from ledgrab.core.devices.led_client import ( DeviceHealth, DiscoveredDevice, LEDClient, LEDDeviceProvider, + ProviderDeps, ) from ledgrab.core.devices.group_client import GroupLEDClient +if TYPE_CHECKING: + from ledgrab.core.devices.device_config import GroupConfig + class GroupDeviceProvider(LEDDeviceProvider): """Provider for group devices that aggregate multiple child devices.""" @@ -23,17 +29,8 @@ class GroupDeviceProvider(LEDDeviceProvider): def capabilities(self) -> set: return {"manual_led_count"} - def create_client(self, url: str, **kwargs) -> LEDClient: - device_store = kwargs.get("device_store") - device_id = kwargs.get("device_id", "") - group_mode = kwargs.get("group_mode", "sequence") - group_device_ids = kwargs.get("group_device_ids", []) - return GroupLEDClient( - device_store=device_store, - device_id=device_id, - group_mode=group_mode, - group_device_ids=group_device_ids, - ) + def create_client(self, config: "GroupConfig", *, deps: ProviderDeps) -> LEDClient: + return GroupLEDClient(config=config, deps=deps) async def check_health(self, url: str, http_client, prev_health=None) -> DeviceHealth: # Group health is aggregated by the processor manager from children's health. diff --git a/server/src/ledgrab/core/devices/hue_provider.py b/server/src/ledgrab/core/devices/hue_provider.py index 301ff5c..e7fa21c 100644 --- a/server/src/ledgrab/core/devices/hue_provider.py +++ b/server/src/ledgrab/core/devices/hue_provider.py @@ -1,17 +1,23 @@ """Philips Hue device provider — entertainment streaming to Hue lights.""" +from __future__ import annotations + import asyncio -from typing import List, Tuple +from typing import TYPE_CHECKING, List, Tuple from ledgrab.core.devices.led_client import ( DeviceHealth, DiscoveredDevice, LEDClient, LEDDeviceProvider, + ProviderDeps, ) from ledgrab.core.devices.hue_client import HueClient from ledgrab.utils import get_logger +if TYPE_CHECKING: + from ledgrab.core.devices.device_config import HueConfig + logger = get_logger(__name__) @@ -37,13 +43,13 @@ class HueDeviceProvider(LEDDeviceProvider): "static_color", } - def create_client(self, url: str, **kwargs) -> LEDClient: + def create_client(self, config: "HueConfig", *, deps: ProviderDeps) -> LEDClient: return HueClient( - url, - led_count=kwargs.get("led_count", 0), - hue_username=kwargs.get("hue_username", ""), - hue_client_key=kwargs.get("hue_client_key", ""), - hue_entertainment_group_id=kwargs.get("hue_entertainment_group_id", ""), + config.device_url, + led_count=config.led_count, + hue_username=config.hue_username, + hue_client_key=config.hue_client_key, + hue_entertainment_group_id=config.hue_entertainment_group_id, ) async def check_health(self, url: str, http_client, prev_health=None) -> DeviceHealth: diff --git a/server/src/ledgrab/core/devices/led_client.py b/server/src/ledgrab/core/devices/led_client.py index 408f08a..e04f084 100644 --- a/server/src/ledgrab/core/devices/led_client.py +++ b/server/src/ledgrab/core/devices/led_client.py @@ -3,10 +3,21 @@ from abc import ABC, abstractmethod from dataclasses import dataclass from datetime import datetime, timezone -from typing import Dict, List, Optional, Tuple, Union +from typing import TYPE_CHECKING, Dict, List, Optional, Tuple, Union import numpy as np +if TYPE_CHECKING: + from ledgrab.core.devices.device_config import DeviceConfig + from ledgrab.storage.device_store import DeviceStore + + +@dataclass(frozen=True) +class ProviderDeps: + """Runtime dependencies injected into every provider.create_client() call.""" + + device_store: Optional["DeviceStore"] = None + @dataclass class DeviceHealth: @@ -172,7 +183,7 @@ class LEDDeviceProvider(ABC): return set() @abstractmethod - def create_client(self, url: str, **kwargs) -> LEDClient: + def create_client(self, config: "DeviceConfig", *, deps: "ProviderDeps") -> LEDClient: """Create a connected-ready LEDClient for this device type.""" ... @@ -251,9 +262,9 @@ def get_all_providers() -> Dict[str, LEDDeviceProvider]: # ===== FACTORY FUNCTIONS (delegate to providers) ===== -def create_led_client(device_type: str, url: str, **kwargs) -> LEDClient: +def create_led_client(config: "DeviceConfig", *, deps: "ProviderDeps") -> LEDClient: """Factory: create the right LEDClient subclass for a device type.""" - return get_provider(device_type).create_client(url, **kwargs) + return get_provider(config.device_type).create_client(config, deps=deps) async def check_device_health( @@ -318,6 +329,17 @@ def _register_builtin_providers(): register_provider(HueDeviceProvider()) + # BLE support is optional — only register the provider if the ``bleak`` + # extra is installed. Importing the provider itself is safe (it doesn't + # import bleak at module load), but we still want a clean skip on + # platforms like Chaquopy/Android where BLE cannot work. + try: + from ledgrab.core.devices.ble_provider import BLEDeviceProvider + + register_provider(BLEDeviceProvider()) + except ImportError: + pass + from ledgrab.core.devices.usbhid_provider import USBHIDDeviceProvider register_provider(USBHIDDeviceProvider()) diff --git a/server/src/ledgrab/core/devices/mock_provider.py b/server/src/ledgrab/core/devices/mock_provider.py index 74fd8a6..f4d51f5 100644 --- a/server/src/ledgrab/core/devices/mock_provider.py +++ b/server/src/ledgrab/core/devices/mock_provider.py @@ -1,16 +1,22 @@ """Mock device provider — virtual LED strip for testing.""" +from __future__ import annotations + from datetime import datetime, timezone -from typing import List +from typing import TYPE_CHECKING, List from ledgrab.core.devices.led_client import ( DeviceHealth, DiscoveredDevice, LEDClient, LEDDeviceProvider, + ProviderDeps, ) from ledgrab.core.devices.mock_client import MockClient +if TYPE_CHECKING: + from ledgrab.core.devices.device_config import MockConfig + class MockDeviceProvider(LEDDeviceProvider): """Provider for virtual mock LED devices.""" @@ -23,11 +29,11 @@ class MockDeviceProvider(LEDDeviceProvider): def capabilities(self) -> set: return {"manual_led_count", "power_control", "brightness_control"} - def create_client(self, url: str, **kwargs) -> LEDClient: + def create_client(self, config: "MockConfig", *, deps: ProviderDeps) -> LEDClient: return MockClient( - url, - led_count=kwargs.get("led_count", 0), - send_latency_ms=kwargs.get("send_latency_ms", 0), + config.device_url, + led_count=config.led_count, + send_latency_ms=config.send_latency_ms, ) async def check_health(self, url: str, http_client, prev_health=None) -> DeviceHealth: diff --git a/server/src/ledgrab/core/devices/mqtt_provider.py b/server/src/ledgrab/core/devices/mqtt_provider.py index 5f2a470..234af78 100644 --- a/server/src/ledgrab/core/devices/mqtt_provider.py +++ b/server/src/ledgrab/core/devices/mqtt_provider.py @@ -1,12 +1,15 @@ """MQTT device provider — factory, validation, health checks.""" -from typing import List +from __future__ import annotations + +from typing import TYPE_CHECKING, List from ledgrab.core.devices.led_client import ( DeviceHealth, DiscoveredDevice, LEDClient, LEDDeviceProvider, + ProviderDeps, ) from ledgrab.core.devices.mqtt_client import ( MQTTLEDClient, @@ -14,6 +17,9 @@ from ledgrab.core.devices.mqtt_client import ( ) from ledgrab.utils import get_logger +if TYPE_CHECKING: + from ledgrab.core.devices.device_config import MQTTConfig + logger = get_logger(__name__) @@ -28,10 +34,10 @@ class MQTTDeviceProvider(LEDDeviceProvider): def capabilities(self) -> set: return {"manual_led_count"} - def create_client(self, url: str, **kwargs) -> LEDClient: + def create_client(self, config: "MQTTConfig", *, deps: ProviderDeps) -> LEDClient: return MQTTLEDClient( - url, - led_count=kwargs.get("led_count", 0), + config.device_url, + led_count=config.led_count, ) async def check_health( diff --git a/server/src/ledgrab/core/devices/openrgb_provider.py b/server/src/ledgrab/core/devices/openrgb_provider.py index 62a8015..1318dbf 100644 --- a/server/src/ledgrab/core/devices/openrgb_provider.py +++ b/server/src/ledgrab/core/devices/openrgb_provider.py @@ -1,13 +1,16 @@ """OpenRGB device provider — factory, validation, health checks, discovery.""" +from __future__ import annotations + import asyncio -from typing import List, Tuple +from typing import TYPE_CHECKING, List, Tuple from ledgrab.core.devices.led_client import ( DeviceHealth, DiscoveredDevice, LEDClient, LEDDeviceProvider, + ProviderDeps, ) from ledgrab.core.devices.openrgb_client import ( OpenRGBLEDClient, @@ -15,6 +18,9 @@ from ledgrab.core.devices.openrgb_client import ( ) from ledgrab.utils import get_logger +if TYPE_CHECKING: + from ledgrab.core.devices.device_config import OpenRGBConfig + logger = get_logger(__name__) @@ -29,10 +35,10 @@ class OpenRGBDeviceProvider(LEDDeviceProvider): def capabilities(self) -> set: return {"health_check", "auto_restore", "static_color", "brightness_control"} - def create_client(self, url: str, **kwargs) -> LEDClient: + def create_client(self, config: "OpenRGBConfig", *, deps: ProviderDeps) -> LEDClient: return OpenRGBLEDClient( - url, - zone_mode=kwargs.get("zone_mode", "combined"), + config.device_url, + zone_mode=config.zone_mode, ) async def check_health(self, url: str, http_client, prev_health=None) -> DeviceHealth: diff --git a/server/src/ledgrab/core/devices/serial_provider.py b/server/src/ledgrab/core/devices/serial_provider.py index eef7a74..66c1fc3 100644 --- a/server/src/ledgrab/core/devices/serial_provider.py +++ b/server/src/ledgrab/core/devices/serial_provider.py @@ -5,14 +5,18 @@ All common serial-device logic (COM port validation, discovery, health checks, power control via black frames) lives here. """ -from typing import List +from __future__ import annotations + +from typing import TYPE_CHECKING, List import numpy as np from ledgrab.core.devices.led_client import ( DeviceHealth, DiscoveredDevice, + LEDClient, LEDDeviceProvider, + ProviderDeps, ) from ledgrab.core.devices.serial_transport import ( list_serial_ports, @@ -20,6 +24,9 @@ from ledgrab.core.devices.serial_transport import ( ) from ledgrab.utils import get_logger +if TYPE_CHECKING: + from ledgrab.core.devices.device_config import BaseDeviceConfig + logger = get_logger(__name__) @@ -40,15 +47,15 @@ class SerialDeviceProvider(LEDDeviceProvider): def device_type(self) -> str: return self._device_type - def create_client(self, url: str, **kwargs): + def create_client(self, config: "BaseDeviceConfig", *, deps: ProviderDeps) -> LEDClient: if self.client_cls is None: raise NotImplementedError( f"{type(self).__name__} must set client_cls or override create_client" ) return self.client_cls( - url, - led_count=kwargs.get("led_count", 0), - baud_rate=kwargs.get("baud_rate"), + config.device_url, + led_count=config.led_count, + baud_rate=getattr(config, "baud_rate", None), ) @property @@ -127,8 +134,12 @@ class SerialDeviceProvider(LEDDeviceProvider): raise ValueError( f"led_count is required to send black frame to {self.device_type} device" ) + if self.client_cls is None: + raise NotImplementedError( + f"{type(self).__name__} must set client_cls or override set_power" + ) - client = self.create_client(url, led_count=led_count, baud_rate=baud_rate) + client = self.client_cls(url, led_count=led_count, baud_rate=baud_rate) try: await client.connect() black = np.zeros((led_count, 3), dtype=np.uint8) diff --git a/server/src/ledgrab/core/devices/spi_provider.py b/server/src/ledgrab/core/devices/spi_provider.py index bff940d..efdd105 100644 --- a/server/src/ledgrab/core/devices/spi_provider.py +++ b/server/src/ledgrab/core/devices/spi_provider.py @@ -1,17 +1,23 @@ """SPI Direct device provider — Raspberry Pi GPIO/SPI direct LED strip control.""" +from __future__ import annotations + import platform -from typing import List +from typing import TYPE_CHECKING, List from ledgrab.core.devices.led_client import ( DeviceHealth, DiscoveredDevice, LEDClient, LEDDeviceProvider, + ProviderDeps, ) from ledgrab.core.devices.spi_client import SPIClient from ledgrab.utils import get_logger +if TYPE_CHECKING: + from ledgrab.core.devices.device_config import SPIConfig + logger = get_logger(__name__) @@ -34,12 +40,12 @@ class SPIDeviceProvider(LEDDeviceProvider): def capabilities(self) -> set: return {"manual_led_count", "health_check", "power_control", "brightness_control"} - def create_client(self, url: str, **kwargs) -> LEDClient: + def create_client(self, config: "SPIConfig", *, deps: ProviderDeps) -> LEDClient: return SPIClient( - url, - led_count=kwargs.get("led_count", 0), - spi_speed_hz=kwargs.get("spi_speed_hz", 800000), - spi_led_type=kwargs.get("spi_led_type", "WS2812B"), + config.device_url, + led_count=config.led_count, + spi_speed_hz=config.spi_speed_hz, + spi_led_type=config.spi_led_type, ) async def check_health(self, url: str, http_client, prev_health=None) -> DeviceHealth: diff --git a/server/src/ledgrab/core/devices/usbhid_provider.py b/server/src/ledgrab/core/devices/usbhid_provider.py index 2eb0eb8..28eff2a 100644 --- a/server/src/ledgrab/core/devices/usbhid_provider.py +++ b/server/src/ledgrab/core/devices/usbhid_provider.py @@ -1,16 +1,22 @@ """USB HID LED device provider — control RGB peripherals via USB HID.""" -from typing import List +from __future__ import annotations + +from typing import TYPE_CHECKING, List from ledgrab.core.devices.led_client import ( DeviceHealth, DiscoveredDevice, LEDClient, LEDDeviceProvider, + ProviderDeps, ) from ledgrab.core.devices.usbhid_client import USBHIDClient, _parse_hid_url from ledgrab.utils import get_logger +if TYPE_CHECKING: + from ledgrab.core.devices.device_config import USBHIDConfig + logger = get_logger(__name__) # Known RGB peripheral vendor IDs and names @@ -41,11 +47,11 @@ class USBHIDDeviceProvider(LEDDeviceProvider): def capabilities(self) -> set: return {"manual_led_count", "health_check"} - def create_client(self, url: str, **kwargs) -> LEDClient: + def create_client(self, config: "USBHIDConfig", *, deps: ProviderDeps) -> LEDClient: return USBHIDClient( - url, - led_count=kwargs.get("led_count", 0), - hid_usage_page=kwargs.get("hid_usage_page", 0), + config.device_url, + led_count=config.led_count, + hid_usage_page=0, ) async def check_health(self, url: str, http_client, prev_health=None) -> DeviceHealth: diff --git a/server/src/ledgrab/core/devices/wled_provider.py b/server/src/ledgrab/core/devices/wled_provider.py index 0ec757a..039753b 100644 --- a/server/src/ledgrab/core/devices/wled_provider.py +++ b/server/src/ledgrab/core/devices/wled_provider.py @@ -1,9 +1,11 @@ """WLED device provider — consolidates all WLED-specific dispatch logic.""" +from __future__ import annotations + import asyncio import json import time -from typing import Dict, List, Optional, Tuple +from typing import TYPE_CHECKING, Dict, List, Optional, Tuple import httpx from zeroconf import ServiceStateChange @@ -14,10 +16,14 @@ from ledgrab.core.devices.led_client import ( DiscoveredDevice, LEDClient, LEDDeviceProvider, + ProviderDeps, ) from ledgrab.core.devices.wled_client import WLEDClient from ledgrab.utils import get_logger +if TYPE_CHECKING: + from ledgrab.core.devices.device_config import WLEDConfig + logger = get_logger(__name__) WLED_MDNS_TYPE = "_wled._tcp.local." @@ -76,10 +82,10 @@ class WLEDDeviceProvider(LEDDeviceProvider): "auto_restore", } - def create_client(self, url: str, **kwargs) -> LEDClient: + def create_client(self, config: "WLEDConfig", *, deps: ProviderDeps) -> LEDClient: return WLEDClient( - url, - use_ddp=kwargs.get("use_ddp", False), + config.device_url, + use_ddp=config.use_ddp, ) async def check_health(self, url: str, http_client, prev_health=None) -> DeviceHealth: diff --git a/server/src/ledgrab/core/devices/ws_provider.py b/server/src/ledgrab/core/devices/ws_provider.py index 0ee4443..f2b2805 100644 --- a/server/src/ledgrab/core/devices/ws_provider.py +++ b/server/src/ledgrab/core/devices/ws_provider.py @@ -1,17 +1,23 @@ """WebSocket device provider — factory, validation, health checks.""" +from __future__ import annotations + from datetime import datetime, timezone -from typing import List +from typing import TYPE_CHECKING, List from ledgrab.core.devices.led_client import ( DeviceHealth, DiscoveredDevice, LEDClient, LEDDeviceProvider, + ProviderDeps, ) from ledgrab.core.devices.ws_client import WSLEDClient from ledgrab.utils import get_logger +if TYPE_CHECKING: + from ledgrab.core.devices.device_config import WSConfig + logger = get_logger(__name__) @@ -26,10 +32,10 @@ class WSDeviceProvider(LEDDeviceProvider): def capabilities(self) -> set: return {"manual_led_count"} - def create_client(self, url: str, **kwargs) -> LEDClient: + def create_client(self, config: "WSConfig", *, deps: ProviderDeps) -> LEDClient: return WSLEDClient( - url, - led_count=kwargs.get("led_count", 0), + config.device_url, + led_count=config.led_count, ) async def check_health( diff --git a/server/src/ledgrab/core/processing/__init__.py b/server/src/ledgrab/core/processing/__init__.py index 6adf81c..1960088 100644 --- a/server/src/ledgrab/core/processing/__init__.py +++ b/server/src/ledgrab/core/processing/__init__.py @@ -6,14 +6,12 @@ from ledgrab.core.processing.processor_manager import ( ProcessorManager, ) from ledgrab.core.processing.target_processor import ( - DeviceInfo, ProcessingMetrics, TargetContext, TargetProcessor, ) __all__ = [ - "DeviceInfo", "DeviceState", "ProcessingMetrics", "ProcessorDependencies", diff --git a/server/src/ledgrab/core/processing/device_test_mode.py b/server/src/ledgrab/core/processing/device_test_mode.py index 444bc6a..bca7ec2 100644 --- a/server/src/ledgrab/core/processing/device_test_mode.py +++ b/server/src/ledgrab/core/processing/device_test_mode.py @@ -6,7 +6,7 @@ Extracted from processor_manager.py to keep files under 800 lines. from typing import Dict, List, Optional from ledgrab.core.capture.calibration import CalibrationConfig -from ledgrab.core.devices.led_client import create_led_client +from ledgrab.core.devices.led_client import ProviderDeps, create_led_client from ledgrab.utils import get_logger logger = get_logger(__name__) @@ -69,13 +69,30 @@ class DeviceTestModeMixin: # Create and cache a new client ds = self._devices[device_id] - client = create_led_client( - ds.device_type, - ds.device_url, - use_ddp=True, - led_count=ds.led_count, - baud_rate=ds.baud_rate, - ) + if self._device_store: + try: + device = self._device_store.get_device(ds.device_id) + config = device.to_config() + except (ValueError, KeyError): + from ledgrab.core.devices.device_config import WLEDConfig + + config = WLEDConfig( + device_id=ds.device_id, + device_url=ds.device_url, + led_count=ds.led_count, + use_ddp=True, + ) + else: + from ledgrab.core.devices.device_config import WLEDConfig + + config = WLEDConfig( + device_id=ds.device_id, + device_url=ds.device_url, + led_count=ds.led_count, + use_ddp=True, + ) + deps = ProviderDeps(device_store=self._device_store) + client = create_led_client(config, deps=deps) await client.connect() self._idle_clients[device_id] = client return client diff --git a/server/src/ledgrab/core/processing/processor_manager.py b/server/src/ledgrab/core/processing/processor_manager.py index f882878..08ac370 100644 --- a/server/src/ledgrab/core/processing/processor_manager.py +++ b/server/src/ledgrab/core/processing/processor_manager.py @@ -16,7 +16,6 @@ from ledgrab.core.processing.metrics_history import MetricsHistory from ledgrab.core.processing.value_stream import ValueStreamManager from ledgrab.core.capture.screen_overlay import OverlayManager from ledgrab.core.processing.target_processor import ( - DeviceInfo, TargetContext, TargetProcessor, ) @@ -219,59 +218,12 @@ class ProcessorManager(AutoRestartMixin, DeviceHealthMixin, DeviceTestModeMixin) value_stream_manager=self._value_stream_manager, cspt_store=self._cspt_store, fire_event=self.fire_event, - get_device_info=self._get_device_info, + is_test_mode_active=lambda did: getattr( + self._devices.get(did), "test_mode_active", False + ), ha_manager=self._ha_manager, ) - # Default values for device-specific fields read from persistent storage - _DEVICE_FIELD_DEFAULTS = { - "send_latency_ms": 0, - "rgbw": False, - "dmx_protocol": "artnet", - "dmx_start_universe": 0, - "dmx_start_channel": 1, - "espnow_peer_mac": "", - "espnow_channel": 1, - "hue_username": "", - "hue_client_key": "", - "hue_entertainment_group_id": "", - "spi_speed_hz": 800000, - "spi_led_type": "WS2812B", - "chroma_device_type": "chromalink", - "gamesense_device_type": "keyboard", - "group_device_ids": [], - "group_mode": "sequence", - } - - def _get_device_info(self, device_id: str) -> Optional[DeviceInfo]: - """Create a DeviceInfo snapshot from the current device state.""" - ds = self._devices.get(device_id) - if ds is None: - return None - # Read device-specific fields from persistent storage - extras = dict(self._DEVICE_FIELD_DEFAULTS) - if self._device_store: - try: - dev = self._device_store.get_device(ds.device_id) - for key, default in self._DEVICE_FIELD_DEFAULTS.items(): - extras[key] = getattr(dev, key, default) - except ValueError as e: - logger.debug("Device %s not found in store, using defaults: %s", ds.device_id, e) - pass - - return DeviceInfo( - device_id=ds.device_id, - device_url=ds.device_url, - led_count=ds.led_count, - device_type=ds.device_type, - baud_rate=ds.baud_rate, - software_brightness=ds.software_brightness, - test_mode_active=ds.test_mode_active, - zone_mode=ds.zone_mode, - auto_shutdown=ds.auto_shutdown, - **extras, - ) - # ===== EVENT SYSTEM (state change notifications) ===== def subscribe_events(self) -> asyncio.Queue: diff --git a/server/src/ledgrab/core/processing/target_processor.py b/server/src/ledgrab/core/processing/target_processor.py index 08c36ec..ffe96ea 100644 --- a/server/src/ledgrab/core/processing/target_processor.py +++ b/server/src/ledgrab/core/processing/target_processor.py @@ -12,9 +12,9 @@ from __future__ import annotations import asyncio from abc import ABC, abstractmethod -from dataclasses import dataclass, field +from dataclasses import dataclass from datetime import datetime -from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Tuple +from typing import TYPE_CHECKING, Any, Callable, Dict, Optional, Tuple if TYPE_CHECKING: from ledgrab.core.processing.color_strip_stream_manager import ColorStripStreamManager @@ -67,44 +67,6 @@ class ProcessingMetrics: fps_effective: int = 0 -@dataclass -class DeviceInfo: - """Read-only snapshot of device state, passed to target processors.""" - - device_id: str - device_url: str - led_count: int - device_type: str = "wled" - baud_rate: Optional[int] = None - software_brightness: int = 255 - test_mode_active: bool = False - send_latency_ms: int = 0 - rgbw: bool = False - zone_mode: str = "combined" - auto_shutdown: bool = False - # DMX (Art-Net / sACN) fields - dmx_protocol: str = "artnet" - dmx_start_universe: int = 0 - dmx_start_channel: int = 1 - # ESP-NOW fields - espnow_peer_mac: str = "" - espnow_channel: int = 1 - # Philips Hue fields - hue_username: str = "" - hue_client_key: str = "" - hue_entertainment_group_id: str = "" - # SPI Direct fields - spi_speed_hz: int = 800000 - spi_led_type: str = "WS2812B" - # Razer Chroma fields - chroma_device_type: str = "chromalink" - # SteelSeries GameSense fields - gamesense_device_type: str = "keyboard" - # Group device fields - group_device_ids: List[str] = field(default_factory=list) - group_mode: str = "sequence" - - @dataclass class TargetContext: """Shared infrastructure bag passed to every TargetProcessor. @@ -122,7 +84,7 @@ class TargetContext: value_stream_manager: Optional["ValueStreamManager"] = None cspt_store: Optional["ColorStripProcessingTemplateStore"] = None fire_event: Callable[[dict], None] = lambda e: None - get_device_info: Callable[[str], Optional[DeviceInfo]] = lambda _: None + is_test_mode_active: Callable[[str], bool] = lambda _: False ha_manager: Optional[Any] = None # HomeAssistantManager (avoid circular import) diff --git a/server/src/ledgrab/core/processing/wled_target_processor.py b/server/src/ledgrab/core/processing/wled_target_processor.py index 5b48075..6bae6ea 100644 --- a/server/src/ledgrab/core/processing/wled_target_processor.py +++ b/server/src/ledgrab/core/processing/wled_target_processor.py @@ -13,12 +13,12 @@ import numpy as np from ledgrab.core.devices.led_client import ( LEDClient, + ProviderDeps, create_led_client, get_device_capabilities, ) from ledgrab.core.capture.screen_capture import get_available_displays from ledgrab.core.processing.target_processor import ( - DeviceInfo, ProcessingMetrics, TargetContext, TargetProcessor, @@ -80,6 +80,7 @@ class WledTargetProcessor(TargetProcessor): self._effective_led_count: int = 0 self._resolved_display_index: Optional[int] = None + self._device_config = None # populated on start(), typed DeviceConfig # Fit-to-device linspace cache (per-instance to avoid cross-target thrash) self._fit_cache_key: tuple = (0, 0) @@ -111,63 +112,48 @@ class WledTargetProcessor(TargetProcessor): logger.debug(f"Processing already running for target {self._target_id}") return - device_info = self._ctx.get_device_info(self._device_id) - if device_info is None: + if self._ctx.device_store is None: raise ValueError(f"Device {self._device_id} not registered") + try: + _dev = self._ctx.device_store.get_device(self._device_id) + except (ValueError, KeyError) as e: + raise ValueError(f"Device {self._device_id} not registered") from e + + from dataclasses import replace as _replace + from ledgrab.core.devices.device_config import WLEDConfig as _WLEDConfig + + config = _dev.to_config() + # use_ddp is a target-derived protocol setting — override on WLEDConfig + if isinstance(config, _WLEDConfig): + config = _replace(config, use_ddp=(self._protocol == "ddp")) + self._device_config = config # Connect to LED device + deps = ProviderDeps(device_store=self._ctx.device_store) try: - self._led_client = create_led_client( - device_info.device_type, - device_info.device_url, - use_ddp=(self._protocol == "ddp"), - led_count=device_info.led_count, - baud_rate=device_info.baud_rate, - send_latency_ms=device_info.send_latency_ms, - rgbw=device_info.rgbw, - zone_mode=device_info.zone_mode, - dmx_protocol=device_info.dmx_protocol, - dmx_start_universe=device_info.dmx_start_universe, - dmx_start_channel=device_info.dmx_start_channel, - espnow_peer_mac=device_info.espnow_peer_mac, - espnow_channel=device_info.espnow_channel, - hue_username=device_info.hue_username, - hue_client_key=device_info.hue_client_key, - hue_entertainment_group_id=device_info.hue_entertainment_group_id, - spi_speed_hz=device_info.spi_speed_hz, - spi_led_type=device_info.spi_led_type, - chroma_device_type=device_info.chroma_device_type, - gamesense_device_type=device_info.gamesense_device_type, - # Group device fields - device_store=self._ctx.device_store, - device_id=device_info.device_id, - group_mode=device_info.group_mode, - group_device_ids=device_info.group_device_ids, - ) + self._led_client = create_led_client(config, deps=deps) await self._led_client.connect() # Use client-reported LED count if available (more accurate than stored) client_led_count = self._led_client.device_led_count effective_led_count = ( - client_led_count - if client_led_count and client_led_count > 0 - else device_info.led_count + client_led_count if client_led_count and client_led_count > 0 else config.led_count ) self._effective_led_count = effective_led_count - if effective_led_count != device_info.led_count: + if effective_led_count != config.led_count: logger.info( f"Target {self._target_id}: device reports {effective_led_count} LEDs " - f"(stored: {device_info.led_count}), using actual count" + f"(stored: {config.led_count}), using actual count" ) logger.info( - f"Target {self._target_id} connected to {device_info.device_type} " + f"Target {self._target_id} connected to {config.device_type} " f"device ({effective_led_count} LEDs)" ) self._device_state_before = await self._led_client.snapshot_device_state() self._needs_keepalive = "standby_required" in get_device_capabilities( - device_info.device_type + config.device_type ) except Exception as e: logger.error(f"Failed to connect to LED device for target {self._target_id}: {e}") @@ -241,8 +227,7 @@ class WledTargetProcessor(TargetProcessor): # Restore device state (only if auto_shutdown is enabled) if self._led_client and self._device_state_before: - device_info = self._ctx.get_device_info(self._device_id) - if device_info and device_info.auto_shutdown: + if self._device_config and self._device_config.auto_shutdown: await self._led_client.restore_device_state(self._device_state_before) self._device_state_before = None @@ -320,9 +305,8 @@ class WledTargetProcessor(TargetProcessor): if css_manager is None: return - device_info = self._ctx.get_device_info(self._device_id) - device_leds = getattr(self, "_effective_led_count", None) or ( - device_info.led_count if device_info else 0 + device_leds = self._effective_led_count or ( + self._device_config.led_count if self._device_config else 0 ) # Release old stream @@ -738,9 +722,8 @@ class WledTargetProcessor(TargetProcessor): _last_preview_broadcast = 0.0 prev_frame_time_stamp = time.perf_counter() asyncio.get_running_loop() - _init_device_info = self._ctx.get_device_info(self._device_id) - _total_leds = getattr(self, "_effective_led_count", None) or ( - _init_device_info.led_count if _init_device_info else 0 + _total_leds = self._effective_led_count or ( + self._device_config.led_count if self._device_config else 0 ) # Stream reference — re-read each tick to detect hot-swaps @@ -790,11 +773,8 @@ class WledTargetProcessor(TargetProcessor): _diag_sleep_jitters: collections.deque = collections.deque(maxlen=300) _diag_slow_iters: collections.deque = collections.deque(maxlen=50) _diag_iter_times: collections.deque = collections.deque(maxlen=300) - _diag_device_info: Optional[DeviceInfo] = None - _diag_device_info_age = 0 - # --- Liveness probe + adaptive FPS --- - _device_url = _init_device_info.device_url if _init_device_info else "" + _device_url = self._device_config.device_url if self._device_config else "" _probe_enabled = _device_url.startswith("http") _probe_interval = 10.0 # seconds between probes _last_probe_time = 0.0 # force first probe soon (after 10s) @@ -894,14 +874,8 @@ class WledTargetProcessor(TargetProcessor): prev_frame_ref = None has_any_frame = False - _diag_device_info_age += 1 - if _diag_device_info is None or _diag_device_info_age >= 300: - _diag_device_info = self._ctx.get_device_info(self._device_id) - _diag_device_info_age = 0 - device_info = _diag_device_info - # Skip send while in calibration test mode - if device_info and device_info.test_mode_active: + if self._ctx.is_test_mode_active(self._device_id): await asyncio.sleep(frame_time) continue @@ -970,7 +944,7 @@ class WledTargetProcessor(TargetProcessor): if _result is not None: frame = _result - cur_brightness = _effective_brightness(device_info) + cur_brightness = _effective_brightness(self._device_config) # Min brightness threshold: combine brightness source # with max pixel value to get effective output brightness. diff --git a/server/src/ledgrab/storage/device_store.py b/server/src/ledgrab/storage/device_store.py index e690a95..a82e9a7 100644 --- a/server/src/ledgrab/storage/device_store.py +++ b/server/src/ledgrab/storage/device_store.py @@ -2,12 +2,15 @@ import uuid from datetime import datetime, timezone -from typing import List, Optional +from typing import TYPE_CHECKING, List, Optional from ledgrab.storage.base_sqlite_store import BaseSqliteStore from ledgrab.storage.database import Database from ledgrab.utils import get_logger +if TYPE_CHECKING: + from ledgrab.core.devices.device_config import DeviceConfig + logger = get_logger(__name__) @@ -52,6 +55,9 @@ class Device: chroma_device_type: str = "chromalink", # SteelSeries GameSense fields gamesense_device_type: str = "keyboard", + # BLE controller fields (SP110E / Triones / Zengge / Govee) + ble_family: str = "", + ble_govee_key: str = "", # Default color strip processing template default_css_processing_template_id: str = "", # Group device fields @@ -85,12 +91,105 @@ class Device: self.spi_led_type = spi_led_type self.chroma_device_type = chroma_device_type self.gamesense_device_type = gamesense_device_type + self.ble_family = ble_family + self.ble_govee_key = ble_govee_key self.default_css_processing_template_id = default_css_processing_template_id self.group_device_ids = group_device_ids or [] self.group_mode = group_mode self.created_at = created_at or datetime.now(timezone.utc) self.updated_at = updated_at or datetime.now(timezone.utc) + def to_config(self) -> "DeviceConfig": + """Return a typed, immutable config snapshot for this device. + + This is the single place that maps the flat Device storage model to the + correct per-provider config subclass. test_mode_active is runtime state + (not stored) and always defaults to False here. + """ + from ledgrab.core.devices.device_config import ( + AdalightConfig, + AmbiLEDConfig, + BLEConfig, + ChromaConfig, + DemoConfig, + DMXConfig, + ESPNowConfig, + GameSenseConfig, + GroupConfig, + HueConfig, + MockConfig, + MQTTConfig, + OpenRGBConfig, + SPIConfig, + USBHIDConfig, + WLEDConfig, + WSConfig, + ) + + base = dict( + device_id=self.id, + device_url=self.url, + led_count=self.led_count, + software_brightness=self.software_brightness, + auto_shutdown=self.auto_shutdown, + rgbw=self.rgbw, + ) + dt = self.device_type + if dt == "wled": + return WLEDConfig(**base) + if dt == "adalight": + return AdalightConfig(**base, baud_rate=self.baud_rate) + if dt == "ambiled": + return AmbiLEDConfig(**base, baud_rate=self.baud_rate) + if dt == "dmx": + return DMXConfig( + **base, + dmx_protocol=self.dmx_protocol, + dmx_start_universe=self.dmx_start_universe, + dmx_start_channel=self.dmx_start_channel, + ) + if dt == "espnow": + return ESPNowConfig( + **base, + baud_rate=self.baud_rate, + espnow_peer_mac=self.espnow_peer_mac, + espnow_channel=self.espnow_channel, + ) + if dt == "hue": + return HueConfig( + **base, + hue_username=self.hue_username, + hue_client_key=self.hue_client_key, + hue_entertainment_group_id=self.hue_entertainment_group_id, + ) + if dt == "spi": + return SPIConfig(**base, spi_speed_hz=self.spi_speed_hz, spi_led_type=self.spi_led_type) + if dt == "chroma": + return ChromaConfig(**base, chroma_device_type=self.chroma_device_type) + if dt == "gamesense": + return GameSenseConfig(**base, gamesense_device_type=self.gamesense_device_type) + if dt == "ble": + return BLEConfig(**base, ble_family=self.ble_family, ble_govee_key=self.ble_govee_key) + if dt == "group": + return GroupConfig( + **base, + group_mode=self.group_mode, + group_device_ids=list(self.group_device_ids), + ) + if dt == "openrgb": + return OpenRGBConfig(**base, zone_mode=self.zone_mode) + if dt == "mock": + return MockConfig(**base, send_latency_ms=self.send_latency_ms) + if dt == "demo": + return DemoConfig(**base, send_latency_ms=self.send_latency_ms) + if dt == "mqtt": + return MQTTConfig(**base) + if dt == "ws": + return WSConfig(**base) + if dt == "usbhid": + return USBHIDConfig(**base) + raise ValueError(f"Unknown device type: {dt!r}") + def to_dict(self) -> dict: """Convert device to dictionary.""" d = { @@ -141,6 +240,10 @@ class Device: d["chroma_device_type"] = self.chroma_device_type if self.gamesense_device_type != "keyboard": d["gamesense_device_type"] = self.gamesense_device_type + if self.ble_family: + d["ble_family"] = self.ble_family + if self.ble_govee_key: + d["ble_govee_key"] = self.ble_govee_key if self.default_css_processing_template_id: d["default_css_processing_template_id"] = self.default_css_processing_template_id if self.group_device_ids: @@ -178,6 +281,8 @@ class Device: spi_led_type=data.get("spi_led_type", "WS2812B"), chroma_device_type=data.get("chroma_device_type", "chromalink"), gamesense_device_type=data.get("gamesense_device_type", "keyboard"), + ble_family=data.get("ble_family", ""), + ble_govee_key=data.get("ble_govee_key", ""), default_css_processing_template_id=data.get("default_css_processing_template_id", ""), group_device_ids=data.get("group_device_ids", []), group_mode=data.get("group_mode", "sequence"), @@ -217,6 +322,8 @@ _UPDATABLE_FIELDS: frozenset[str] = frozenset( "spi_led_type", "chroma_device_type", "gamesense_device_type", + "ble_family", + "ble_govee_key", "default_css_processing_template_id", "group_device_ids", "group_mode", @@ -265,6 +372,8 @@ class DeviceStore(BaseSqliteStore[Device]): spi_led_type: str = "WS2812B", chroma_device_type: str = "chromalink", gamesense_device_type: str = "keyboard", + ble_family: str = "", + ble_govee_key: str = "", group_device_ids: Optional[List[str]] = None, group_mode: str = "sequence", ) -> Device: @@ -302,6 +411,8 @@ class DeviceStore(BaseSqliteStore[Device]): spi_led_type=spi_led_type, chroma_device_type=chroma_device_type, gamesense_device_type=gamesense_device_type, + ble_family=ble_family, + ble_govee_key=ble_govee_key, group_device_ids=group_device_ids or [], group_mode=group_mode, ) diff --git a/server/tests/core/devices/__init__.py b/server/tests/core/devices/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/server/tests/core/devices/test_device_config.py b/server/tests/core/devices/test_device_config.py new file mode 100644 index 0000000..0ad4ea5 --- /dev/null +++ b/server/tests/core/devices/test_device_config.py @@ -0,0 +1,335 @@ +"""Unit tests for the per-provider DeviceConfig hierarchy and Device.to_config().""" + +import pytest + +from ledgrab.core.devices.device_config import ( + AdalightConfig, + AmbiLEDConfig, + BaseDeviceConfig, + BLEConfig, + ChromaConfig, + DemoConfig, + DMXConfig, + ESPNowConfig, + GameSenseConfig, + GroupConfig, + HueConfig, + MockConfig, + MQTTConfig, + OpenRGBConfig, + SPIConfig, + USBHIDConfig, + WLEDConfig, + WSConfig, +) +from ledgrab.storage.device_store import Device + + +def _make_device(**kwargs) -> Device: + defaults = dict( + device_id="dev_test01", + name="Test Device", + url="http://192.168.1.100", + led_count=60, + software_brightness=200, + auto_shutdown=True, + rgbw=True, + baud_rate=115200, + send_latency_ms=5, + zone_mode="separate", + dmx_protocol="sacn", + dmx_start_universe=1, + dmx_start_channel=2, + espnow_peer_mac="AA:BB:CC:DD:EE:FF", + espnow_channel=6, + hue_username="hue_user", + hue_client_key="hue_key", + hue_entertainment_group_id="grp_1", + spi_speed_hz=1600000, + spi_led_type="APA102", + chroma_device_type="keyboard", + gamesense_device_type="mouse", + ble_family="govee", + ble_govee_key="secret_key", + group_device_ids=["dev_a", "dev_b"], + group_mode="independent", + ) + defaults.update(kwargs) + return Device(**defaults) + + +class TestBaseDeviceConfigDefaults: + def test_defaults(self): + cfg = BaseDeviceConfig(device_id="d1", device_url="http://x", led_count=10) + assert cfg.software_brightness == 255 + assert cfg.test_mode_active is False + assert cfg.auto_shutdown is False + assert cfg.rgbw is False + + def test_frozen(self): + cfg = BaseDeviceConfig(device_id="d1", device_url="http://x", led_count=10) + with pytest.raises(Exception): + cfg.led_count = 20 # type: ignore[misc] + + +class TestWLEDConfig: + def test_to_config_returns_wled(self): + device = _make_device(device_type="wled") + cfg = device.to_config() + assert isinstance(cfg, WLEDConfig) + assert cfg.device_type == "wled" + + def test_base_fields_mapped(self): + device = _make_device(device_type="wled") + cfg = device.to_config() + assert cfg.device_id == "dev_test01" + assert cfg.device_url == "http://192.168.1.100" + assert cfg.led_count == 60 + assert cfg.software_brightness == 200 + assert cfg.auto_shutdown is True + assert cfg.rgbw is True + + def test_use_ddp_defaults_false(self): + cfg = _make_device(device_type="wled").to_config() + assert cfg.use_ddp is False # type: ignore[union-attr] + + def test_test_mode_active_defaults_false(self): + # test_mode_active is runtime state, not stored on Device + cfg = _make_device(device_type="wled").to_config() + assert cfg.test_mode_active is False + + def test_wled_has_no_baud_rate(self): + cfg = _make_device(device_type="wled").to_config() + assert not hasattr(cfg, "baud_rate") + + def test_wled_has_no_ble_fields(self): + cfg = _make_device(device_type="wled").to_config() + assert not hasattr(cfg, "ble_family") + assert not hasattr(cfg, "ble_govee_key") + + +class TestAdalightConfig: + def test_to_config_returns_adalight(self): + cfg = _make_device(device_type="adalight").to_config() + assert isinstance(cfg, AdalightConfig) + assert cfg.device_type == "adalight" + + def test_baud_rate_mapped(self): + cfg = _make_device(device_type="adalight").to_config() + assert cfg.baud_rate == 115200 # type: ignore[union-attr] + + def test_adalight_has_no_dmx_fields(self): + cfg = _make_device(device_type="adalight").to_config() + assert not hasattr(cfg, "dmx_protocol") + + +class TestAmbiLEDConfig: + def test_to_config_returns_ambiled(self): + cfg = _make_device(device_type="ambiled").to_config() + assert isinstance(cfg, AmbiLEDConfig) + assert cfg.device_type == "ambiled" + + def test_baud_rate_mapped(self): + cfg = _make_device(device_type="ambiled").to_config() + assert cfg.baud_rate == 115200 # type: ignore[union-attr] + + +class TestDMXConfig: + def test_to_config_returns_dmx(self): + cfg = _make_device(device_type="dmx").to_config() + assert isinstance(cfg, DMXConfig) + assert cfg.device_type == "dmx" + + def test_dmx_fields_mapped(self): + cfg = _make_device(device_type="dmx").to_config() + assert cfg.dmx_protocol == "sacn" # type: ignore[union-attr] + assert cfg.dmx_start_universe == 1 # type: ignore[union-attr] + assert cfg.dmx_start_channel == 2 # type: ignore[union-attr] + + def test_dmx_has_no_baud_rate(self): + cfg = _make_device(device_type="dmx").to_config() + assert not hasattr(cfg, "baud_rate") + + +class TestESPNowConfig: + def test_to_config_returns_espnow(self): + cfg = _make_device(device_type="espnow").to_config() + assert isinstance(cfg, ESPNowConfig) + assert cfg.device_type == "espnow" + + def test_espnow_fields_mapped(self): + cfg = _make_device(device_type="espnow").to_config() + assert cfg.baud_rate == 115200 # type: ignore[union-attr] + assert cfg.espnow_peer_mac == "AA:BB:CC:DD:EE:FF" # type: ignore[union-attr] + assert cfg.espnow_channel == 6 # type: ignore[union-attr] + + +class TestHueConfig: + def test_to_config_returns_hue(self): + cfg = _make_device(device_type="hue").to_config() + assert isinstance(cfg, HueConfig) + assert cfg.device_type == "hue" + + def test_hue_fields_mapped(self): + cfg = _make_device(device_type="hue").to_config() + assert cfg.hue_username == "hue_user" # type: ignore[union-attr] + assert cfg.hue_client_key == "hue_key" # type: ignore[union-attr] + assert cfg.hue_entertainment_group_id == "grp_1" # type: ignore[union-attr] + + +class TestSPIConfig: + def test_to_config_returns_spi(self): + cfg = _make_device(device_type="spi").to_config() + assert isinstance(cfg, SPIConfig) + assert cfg.device_type == "spi" + + def test_spi_fields_mapped(self): + cfg = _make_device(device_type="spi").to_config() + assert cfg.spi_speed_hz == 1600000 # type: ignore[union-attr] + assert cfg.spi_led_type == "APA102" # type: ignore[union-attr] + + +class TestChromaConfig: + def test_to_config_returns_chroma(self): + cfg = _make_device(device_type="chroma").to_config() + assert isinstance(cfg, ChromaConfig) + assert cfg.device_type == "chroma" + + def test_chroma_device_type_mapped(self): + cfg = _make_device(device_type="chroma").to_config() + assert cfg.chroma_device_type == "keyboard" # type: ignore[union-attr] + + +class TestGameSenseConfig: + def test_to_config_returns_gamesense(self): + cfg = _make_device(device_type="gamesense").to_config() + assert isinstance(cfg, GameSenseConfig) + assert cfg.device_type == "gamesense" + + def test_gamesense_device_type_mapped(self): + cfg = _make_device(device_type="gamesense").to_config() + assert cfg.gamesense_device_type == "mouse" # type: ignore[union-attr] + + +class TestBLEConfig: + def test_to_config_returns_ble(self): + cfg = _make_device(device_type="ble").to_config() + assert isinstance(cfg, BLEConfig) + assert cfg.device_type == "ble" + + def test_ble_fields_mapped(self): + cfg = _make_device(device_type="ble").to_config() + assert cfg.ble_family == "govee" # type: ignore[union-attr] + assert cfg.ble_govee_key == "secret_key" # type: ignore[union-attr] + + def test_ble_has_no_dmx_fields(self): + cfg = _make_device(device_type="ble").to_config() + assert not hasattr(cfg, "dmx_protocol") + + +class TestGroupConfig: + def test_to_config_returns_group(self): + cfg = _make_device(device_type="group").to_config() + assert isinstance(cfg, GroupConfig) + assert cfg.device_type == "group" + + def test_group_fields_mapped(self): + cfg = _make_device(device_type="group").to_config() + assert cfg.group_mode == "independent" # type: ignore[union-attr] + assert cfg.group_device_ids == ["dev_a", "dev_b"] # type: ignore[union-attr] + + def test_group_device_ids_is_copy(self): + device = _make_device(device_type="group") + cfg = device.to_config() + # Modifying the device's list should not affect the config snapshot + device.group_device_ids.append("dev_c") + assert "dev_c" not in cfg.group_device_ids # type: ignore[union-attr] + + def test_group_has_no_ble_fields(self): + cfg = _make_device(device_type="group").to_config() + assert not hasattr(cfg, "ble_family") + + +class TestOpenRGBConfig: + def test_to_config_returns_openrgb(self): + cfg = _make_device(device_type="openrgb").to_config() + assert isinstance(cfg, OpenRGBConfig) + assert cfg.device_type == "openrgb" + + def test_zone_mode_mapped(self): + cfg = _make_device(device_type="openrgb").to_config() + assert cfg.zone_mode == "separate" # type: ignore[union-attr] + + +class TestMockConfig: + def test_to_config_returns_mock(self): + cfg = _make_device(device_type="mock").to_config() + assert isinstance(cfg, MockConfig) + assert cfg.device_type == "mock" + + def test_send_latency_ms_mapped(self): + cfg = _make_device(device_type="mock").to_config() + assert cfg.send_latency_ms == 5 # type: ignore[union-attr] + + +class TestDemoConfig: + def test_to_config_returns_demo(self): + cfg = _make_device(device_type="demo").to_config() + assert isinstance(cfg, DemoConfig) + assert cfg.device_type == "demo" + + def test_send_latency_ms_mapped(self): + cfg = _make_device(device_type="demo").to_config() + assert cfg.send_latency_ms == 5 # type: ignore[union-attr] + + +class TestMQTTConfig: + def test_to_config_returns_mqtt(self): + cfg = _make_device(device_type="mqtt").to_config() + assert isinstance(cfg, MQTTConfig) + assert cfg.device_type == "mqtt" + + +class TestWSConfig: + def test_to_config_returns_ws(self): + cfg = _make_device(device_type="ws").to_config() + assert isinstance(cfg, WSConfig) + assert cfg.device_type == "ws" + + +class TestUSBHIDConfig: + def test_to_config_returns_usbhid(self): + cfg = _make_device(device_type="usbhid").to_config() + assert isinstance(cfg, USBHIDConfig) + assert cfg.device_type == "usbhid" + + +class TestUnknownDeviceType: + def test_raises_on_unknown_type(self): + device = _make_device(device_type="unknown_future_type") + with pytest.raises(ValueError, match="Unknown device type"): + device.to_config() + + +class TestFieldIsolation: + """Irrelevant Device fields must not leak into the wrong config type.""" + + def test_wled_has_no_espnow_fields(self): + cfg = _make_device(device_type="wled").to_config() + assert not hasattr(cfg, "espnow_peer_mac") + assert not hasattr(cfg, "espnow_channel") + + def test_hue_has_no_spi_fields(self): + cfg = _make_device(device_type="hue").to_config() + assert not hasattr(cfg, "spi_speed_hz") + assert not hasattr(cfg, "spi_led_type") + + def test_adalight_has_no_group_fields(self): + cfg = _make_device(device_type="adalight").to_config() + assert not hasattr(cfg, "group_mode") + assert not hasattr(cfg, "group_device_ids") + + def test_mqtt_has_no_hue_fields(self): + cfg = _make_device(device_type="mqtt").to_config() + assert not hasattr(cfg, "hue_username") + assert not hasattr(cfg, "hue_client_key") diff --git a/server/tests/test_group_device.py b/server/tests/test_group_device.py index a29a22d..f3e707e 100644 --- a/server/tests/test_group_device.py +++ b/server/tests/test_group_device.py @@ -3,6 +3,8 @@ import numpy as np import pytest +from ledgrab.core.devices.device_config import GroupConfig +from ledgrab.core.devices.led_client import ProviderDeps from ledgrab.storage.database import Database from ledgrab.storage.device_store import Device, DeviceStore @@ -238,17 +240,22 @@ class TestGroupLEDClient: d3 = _create_device(store, "d3", 30) return store, [d1, d2, d3] - @pytest.mark.asyncio - async def test_connect_creates_children(self, mock_store): + def _make_client(self, store, devices, mode="sequence"): from ledgrab.core.devices.group_client import GroupLEDClient - store, devices = mock_store - client = GroupLEDClient( - device_store=store, + config = GroupConfig( device_id="test_group", - group_mode="sequence", + device_url="group://test_group", + led_count=sum(d.led_count for d in devices), + group_mode=mode, group_device_ids=[d.id for d in devices], ) + return GroupLEDClient(config=config, deps=ProviderDeps(device_store=store)) + + @pytest.mark.asyncio + async def test_connect_creates_children(self, mock_store): + store, devices = mock_store + client = self._make_client(store, devices) await client.connect() assert client.is_connected assert client.device_led_count == 60 # 10+20+30 @@ -257,15 +264,8 @@ class TestGroupLEDClient: @pytest.mark.asyncio async def test_sequence_mode_slices(self, mock_store): - from ledgrab.core.devices.group_client import GroupLEDClient - store, devices = mock_store - client = GroupLEDClient( - device_store=store, - device_id="test_group", - group_mode="sequence", - group_device_ids=[d.id for d in devices], - ) + client = self._make_client(store, devices) await client.connect() # Capture what each child receives @@ -292,15 +292,8 @@ class TestGroupLEDClient: @pytest.mark.asyncio async def test_independent_mode_resamples(self, mock_store): - from ledgrab.core.devices.group_client import GroupLEDClient - store, devices = mock_store - client = GroupLEDClient( - device_store=store, - device_id="test_group", - group_mode="independent", - group_device_ids=[d.id for d in devices], - ) + client = self._make_client(store, devices, mode="independent") await client.connect() sent_pixels = [] @@ -328,15 +321,8 @@ class TestGroupLEDClient: @pytest.mark.asyncio async def test_close_cleans_up(self, mock_store): - from ledgrab.core.devices.group_client import GroupLEDClient - store, devices = mock_store - client = GroupLEDClient( - device_store=store, - device_id="test_group", - group_mode="sequence", - group_device_ids=[d.id for d in devices], - ) + client = self._make_client(store, devices) await client.connect() assert client.is_connected await client.close() @@ -345,15 +331,8 @@ class TestGroupLEDClient: @pytest.mark.asyncio async def test_sequence_pads_short_pixels(self, mock_store): - from ledgrab.core.devices.group_client import GroupLEDClient - store, devices = mock_store - client = GroupLEDClient( - device_store=store, - device_id="test_group", - group_mode="sequence", - group_device_ids=[d.id for d in devices], - ) + client = self._make_client(store, devices) await client.connect() sent_pixels = []