"""LED output target — sends color strip sources to an LED device.""" from dataclasses import dataclass from datetime import datetime, timezone from typing import List, Optional from wled_controller.storage.output_target import OutputTarget DEFAULT_STATE_CHECK_INTERVAL = 30 # seconds @dataclass class WledOutputTarget(OutputTarget): """LED output target — pairs an LED device with a ColorStripSource.""" device_id: str = "" color_strip_source_id: str = "" brightness_value_source_id: str = "" fps: int = 30 # target send FPS (1-90) keepalive_interval: float = 1.0 # seconds between keepalive sends when screen is static state_check_interval: int = DEFAULT_STATE_CHECK_INTERVAL min_brightness_threshold: int = 0 # brightness below this → 0 (disabled when 0) adaptive_fps: bool = False # auto-reduce FPS when device is unresponsive protocol: str = "ddp" # "ddp" (UDP) or "http" (JSON API) def register_with_manager(self, manager) -> None: """Register this WLED target with the processor manager.""" if self.device_id: manager.add_target( target_id=self.id, device_id=self.device_id, color_strip_source_id=self.color_strip_source_id, fps=self.fps, keepalive_interval=self.keepalive_interval, state_check_interval=self.state_check_interval, brightness_value_source_id=self.brightness_value_source_id, min_brightness_threshold=self.min_brightness_threshold, adaptive_fps=self.adaptive_fps, protocol=self.protocol, ) def sync_with_manager(self, manager, *, settings_changed: bool, css_changed: bool = False, device_changed: bool = False, brightness_vs_changed: bool = False) -> None: """Push changed fields to the processor manager.""" if settings_changed: manager.update_target_settings(self.id, { "fps": self.fps, "keepalive_interval": self.keepalive_interval, "state_check_interval": self.state_check_interval, "min_brightness_threshold": self.min_brightness_threshold, "adaptive_fps": self.adaptive_fps, }) if css_changed: manager.update_target_css(self.id, self.color_strip_source_id) if device_changed: manager.update_target_device(self.id, self.device_id) if brightness_vs_changed: manager.update_target_brightness_vs(self.id, self.brightness_value_source_id) def update_fields(self, *, name=None, device_id=None, color_strip_source_id=None, brightness_value_source_id=None, fps=None, keepalive_interval=None, state_check_interval=None, min_brightness_threshold=None, adaptive_fps=None, protocol=None, description=None, tags: Optional[List[str]] = None, **_kwargs) -> None: """Apply mutable field updates for WLED targets.""" super().update_fields(name=name, description=description, tags=tags) if device_id is not None: self.device_id = device_id if color_strip_source_id is not None: self.color_strip_source_id = color_strip_source_id if brightness_value_source_id is not None: self.brightness_value_source_id = brightness_value_source_id if fps is not None: self.fps = fps if keepalive_interval is not None: self.keepalive_interval = keepalive_interval if state_check_interval is not None: self.state_check_interval = state_check_interval if min_brightness_threshold is not None: self.min_brightness_threshold = min_brightness_threshold if adaptive_fps is not None: self.adaptive_fps = adaptive_fps if protocol is not None: self.protocol = protocol @property def has_picture_source(self) -> bool: return bool(self.color_strip_source_id) def to_dict(self) -> dict: """Convert to dictionary.""" d = super().to_dict() d["device_id"] = self.device_id d["color_strip_source_id"] = self.color_strip_source_id d["brightness_value_source_id"] = self.brightness_value_source_id d["fps"] = self.fps d["keepalive_interval"] = self.keepalive_interval d["state_check_interval"] = self.state_check_interval d["min_brightness_threshold"] = self.min_brightness_threshold d["adaptive_fps"] = self.adaptive_fps d["protocol"] = self.protocol return d @classmethod def from_dict(cls, data: dict) -> "WledOutputTarget": """Create from dictionary.""" return cls( id=data["id"], name=data["name"], target_type="led", device_id=data.get("device_id", ""), color_strip_source_id=data.get("color_strip_source_id", ""), brightness_value_source_id=data.get("brightness_value_source_id", ""), fps=data.get("fps", 30), keepalive_interval=data.get("keepalive_interval", 1.0), state_check_interval=data.get("state_check_interval", DEFAULT_STATE_CHECK_INTERVAL), min_brightness_threshold=data.get("min_brightness_threshold", 0), adaptive_fps=data.get("adaptive_fps", False), protocol=data.get("protocol", "ddp"), description=data.get("description"), tags=data.get("tags", []), created_at=datetime.fromisoformat(data.get("created_at", datetime.now(timezone.utc).isoformat())), updated_at=datetime.fromisoformat(data.get("updated_at", datetime.now(timezone.utc).isoformat())), )