"""Value source data model with inheritance-based source types. A ValueSource produces a scalar float (0.0–1.0) that can drive target parameters like brightness. Five types: StaticValueSource — constant float value AnimatedValueSource — periodic waveform (sine, triangle, square, sawtooth) AudioValueSource — audio-reactive scalar (RMS, peak, beat detection) AdaptiveValueSource — adapts to external conditions: adaptive_time — interpolates brightness along a 24-hour schedule adaptive_scene — derives brightness from a picture source's frame luminance """ from dataclasses import dataclass, field from datetime import datetime, timezone from typing import List, Optional @dataclass class ValueSource: """Base class for value source configurations.""" id: str name: str source_type: str # "static" | "animated" | "audio" | "adaptive_time" | "adaptive_scene" created_at: datetime updated_at: datetime description: Optional[str] = None tags: List[str] = field(default_factory=list) def to_dict(self) -> dict: """Convert source to dictionary. Subclasses extend this.""" return { "id": self.id, "name": self.name, "source_type": self.source_type, "created_at": self.created_at.isoformat(), "updated_at": self.updated_at.isoformat(), "description": self.description, "tags": self.tags, # Subclass fields default to None for forward compat "value": None, "waveform": None, "speed": None, "min_value": None, "max_value": None, "audio_source_id": None, "mode": None, "sensitivity": None, "smoothing": None, "auto_gain": None, "schedule": None, "picture_source_id": None, "scene_behavior": None, } @staticmethod def from_dict(data: dict) -> "ValueSource": """Factory: dispatch to the correct subclass based on source_type.""" source_type: str = data.get("source_type", "static") or "static" sid: str = data["id"] name: str = data["name"] description: str | None = data.get("description") tags: list = data.get("tags", []) raw_created = data.get("created_at") created_at: datetime = ( datetime.fromisoformat(raw_created) if isinstance(raw_created, str) else raw_created if isinstance(raw_created, datetime) else datetime.now(timezone.utc) ) raw_updated = data.get("updated_at") updated_at: datetime = ( datetime.fromisoformat(raw_updated) if isinstance(raw_updated, str) else raw_updated if isinstance(raw_updated, datetime) else datetime.now(timezone.utc) ) if source_type == "animated": return AnimatedValueSource( id=sid, name=name, source_type="animated", created_at=created_at, updated_at=updated_at, description=description, tags=tags, waveform=data.get("waveform") or "sine", speed=float(data.get("speed") or 10.0), min_value=float(data.get("min_value") or 0.0), max_value=float(data["max_value"]) if data.get("max_value") is not None else 1.0, ) if source_type == "audio": return AudioValueSource( id=sid, name=name, source_type="audio", created_at=created_at, updated_at=updated_at, description=description, tags=tags, audio_source_id=data.get("audio_source_id") or "", mode=data.get("mode") or "rms", sensitivity=float(data.get("sensitivity") or 1.0), smoothing=float(data.get("smoothing") or 0.3), min_value=float(data.get("min_value") or 0.0), max_value=float(data["max_value"]) if data.get("max_value") is not None else 1.0, auto_gain=bool(data.get("auto_gain", False)), ) if source_type == "adaptive_time": return AdaptiveValueSource( id=sid, name=name, source_type="adaptive_time", created_at=created_at, updated_at=updated_at, description=description, tags=tags, schedule=data.get("schedule") or [], min_value=float(data.get("min_value") or 0.0), max_value=float(data["max_value"]) if data.get("max_value") is not None else 1.0, ) if source_type == "adaptive_scene": return AdaptiveValueSource( id=sid, name=name, source_type="adaptive_scene", created_at=created_at, updated_at=updated_at, description=description, tags=tags, picture_source_id=data.get("picture_source_id") or "", scene_behavior=data.get("scene_behavior") or "complement", sensitivity=float(data.get("sensitivity") or 1.0), smoothing=float(data.get("smoothing") or 0.3), min_value=float(data.get("min_value") or 0.0), max_value=float(data["max_value"]) if data.get("max_value") is not None else 1.0, ) # Default: "static" type return StaticValueSource( id=sid, name=name, source_type="static", created_at=created_at, updated_at=updated_at, description=description, tags=tags, value=float(data["value"]) if data.get("value") is not None else 1.0, ) @dataclass class StaticValueSource(ValueSource): """Value source that outputs a constant float. Useful as a simple per-target brightness override. """ value: float = 1.0 # 0.0–1.0 def to_dict(self) -> dict: d = super().to_dict() d["value"] = self.value return d @dataclass class AnimatedValueSource(ValueSource): """Value source that cycles through a periodic waveform. Produces a smooth animation between min_value and max_value at the configured speed (cycles per minute). """ waveform: str = "sine" # sine | triangle | square | sawtooth speed: float = 10.0 # cycles per minute (1.0–120.0) min_value: float = 0.0 # minimum output (0.0–1.0) max_value: float = 1.0 # maximum output (0.0–1.0) def to_dict(self) -> dict: d = super().to_dict() d["waveform"] = self.waveform d["speed"] = self.speed d["min_value"] = self.min_value d["max_value"] = self.max_value return d @dataclass class AudioValueSource(ValueSource): """Value source driven by audio input. Converts audio analysis (RMS level, peak, or beat detection) into a scalar value for brightness modulation. """ audio_source_id: str = "" # references an audio source (mono or multichannel) mode: str = "rms" # rms | peak | beat sensitivity: float = 1.0 # gain multiplier (0.1–20.0) smoothing: float = 0.3 # temporal smoothing (0.0–1.0) min_value: float = 0.0 # minimum output (0.0–1.0) max_value: float = 1.0 # maximum output (0.0–1.0) auto_gain: bool = False # auto-normalize audio levels to full range def to_dict(self) -> dict: d = super().to_dict() d["audio_source_id"] = self.audio_source_id d["mode"] = self.mode d["sensitivity"] = self.sensitivity d["smoothing"] = self.smoothing d["min_value"] = self.min_value d["max_value"] = self.max_value d["auto_gain"] = self.auto_gain return d @dataclass class AdaptiveValueSource(ValueSource): """Value source that adapts to external conditions. source_type determines the sub-mode: adaptive_time — interpolates brightness along a 24-hour schedule adaptive_scene — derives brightness from a picture source's frame luminance """ schedule: List[dict] = field(default_factory=list) # [{time: "HH:MM", value: 0.0-1.0}] picture_source_id: str = "" # for scene mode scene_behavior: str = "complement" # "complement" | "match" sensitivity: float = 1.0 # gain multiplier (0.1-5.0) smoothing: float = 0.3 # temporal smoothing (0.0-1.0) min_value: float = 0.0 # output range min max_value: float = 1.0 # output range max def to_dict(self) -> dict: d = super().to_dict() d["schedule"] = self.schedule d["picture_source_id"] = self.picture_source_id d["scene_behavior"] = self.scene_behavior d["sensitivity"] = self.sensitivity d["smoothing"] = self.smoothing d["min_value"] = self.min_value d["max_value"] = self.max_value return d