alexei.dolgolyov/ledgrab
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 14 Wiki Activity

refactor(devices): per-provider typed configs (phases 1-4)

Browse Source 
Phase 1 — DeviceConfig hierarchy (device_config.py):
- 17 @dataclass(frozen=True) subclasses (WLEDConfig, AdalightConfig, …) sharing
  BaseDeviceConfig; DeviceConfig = Union[all 17]
- Device.to_config() in device_store.py: single flat→typed dispatch point

Phase 2+3 — Typed provider signatures + call-site migration:
- ProviderDeps(device_store) frozen dataclass in led_client.py
- LEDDeviceProvider.create_client(config, *, deps) abstract signature
- create_led_client(config, *, deps) factory dispatches via config.device_type
- All 17 providers narrowed to their specific config type; drop kwargs.get()
- GroupLEDClient.connect() uses device.to_config() + create_led_client()
- wled_target_processor: replaced 21-field DeviceInfo unpacking with to_config()
  + dataclasses.replace(config, use_ddp=…) for DDP override
- device_test_mode: build typed config via to_config() + ProviderDeps
- Deleted DeviceInfo dataclass, _get_device_info(), _DEVICE_FIELD_DEFAULTS
- TargetContext: replaced get_device_info callback with is_test_mode_active

Phase 4 — Test migration:
- 47-case test suite in tests/core/devices/test_device_config.py (100% coverage)
- test_group_device.py TestGroupLEDClient migrated to GroupConfig + ProviderDeps
- Removed legacy keyword-arg init path from GroupLEDClient
This commit is contained in:
Admin
2026-04-18 01:24:27 +03:00
parent 123da1b5c4
commit d3a6416a1d
29 changed files with 1192 additions and 328 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/ledgrab/core/devices/chroma_provider.py
+10 -4
View File
@@ -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,
)
server/src/ledgrab/core/devices/demo_provider.py
+11 -5
View File
@@ -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:
server/src/ledgrab/core/devices/device_config.py
+148
View File
@@ -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,
]
server/src/ledgrab/core/devices/dmx_provider.py
+13 -7
View File
@@ -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(
server/src/ledgrab/core/devices/espnow_provider.py
+13 -7
View File
@@ -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:
server/src/ledgrab/core/devices/gamesense_provider.py
+11 -5
View File
@@ -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:
server/src/ledgrab/core/devices/group_client.py
+13 -41
View File
@@ -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)
server/src/ledgrab/core/devices/group_provider.py
+9 -12
View File
@@ -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.
server/src/ledgrab/core/devices/hue_provider.py
+13 -7
View File
@@ -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:
server/src/ledgrab/core/devices/led_client.py
+26 -4
View File
@@ -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())
server/src/ledgrab/core/devices/mock_provider.py
+11 -5
View File
@@ -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:
server/src/ledgrab/core/devices/mqtt_provider.py
+10 -4
View File
@@ -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(
server/src/ledgrab/core/devices/openrgb_provider.py
+10 -4
View File
@@ -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:
server/src/ledgrab/core/devices/serial_provider.py
+17 -6
View File
@@ -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)
server/src/ledgrab/core/devices/spi_provider.py
+12 -6
View File
@@ -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:
server/src/ledgrab/core/devices/usbhid_provider.py
+11 -5
View File
@@ -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:
server/src/ledgrab/core/devices/wled_provider.py
+10 -4
View File
@@ -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:
server/src/ledgrab/core/devices/ws_provider.py
+10 -4
View File
@@ -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(
server/src/ledgrab/core/processing/__init__.py
-2
View File
@@ -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",
server/src/ledgrab/core/processing/device_test_mode.py
+25 -8
View File
@@ -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
server/src/ledgrab/core/processing/processor_manager.py
+3 -51
View File
@@ -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:
server/src/ledgrab/core/processing/target_processor.py
+3 -41
View File
@@ -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)
server/src/ledgrab/core/processing/wled_target_processor.py
+31 -57
View File
@@ -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.
server/src/ledgrab/storage/device_store.py
+112 -1
View File
@@ -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,
)