"""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. Six 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 DaylightValueSource — brightness based on simulated or real-time daylight cycle """ from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Dict, List, Optional, Type @dataclass class ValueSource: """Base class for value source configurations.""" id: str name: str source_type: str # "static" | "animated" | "audio" | "adaptive_time" | "adaptive_scene" | "daylight" 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, "use_real_time": None, "latitude": None, } @staticmethod def from_dict(data: dict) -> "ValueSource": """Factory: dispatch to the correct subclass based on source_type.""" source_type = data.get("source_type", "static") or "static" subcls = _VALUE_SOURCE_MAP.get(source_type, StaticValueSource) return subcls.from_dict(data) def _parse_common_fields(data: dict) -> dict: """Extract common fields shared by all value source types.""" raw_created = data.get("created_at") created_at = ( 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.fromisoformat(raw_updated) if isinstance(raw_updated, str) else raw_updated if isinstance(raw_updated, datetime) else datetime.now(timezone.utc) ) return dict( id=data["id"], name=data["name"], description=data.get("description"), tags=data.get("tags", []), created_at=created_at, updated_at=updated_at, ) @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 @classmethod def from_dict(cls, data: dict) -> "StaticValueSource": common = _parse_common_fields(data) return cls( **common, source_type="static", value=float(data["value"]) if data.get("value") is not None else 1.0, ) @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 @classmethod def from_dict(cls, data: dict) -> "AnimatedValueSource": common = _parse_common_fields(data) return cls( **common, source_type="animated", 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, ) @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 @classmethod def from_dict(cls, data: dict) -> "AudioValueSource": common = _parse_common_fields(data) return cls( **common, source_type="audio", 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)), ) @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 @classmethod def from_dict(cls, data: dict) -> "AdaptiveValueSource": common = _parse_common_fields(data) source_type = data.get("source_type", "adaptive_time") return cls( **common, source_type=source_type, schedule=data.get("schedule") or [], 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, ) @dataclass class DaylightValueSource(ValueSource): """Value source that outputs brightness based on a daylight cycle. Uses the same daylight LUT as the CSS daylight stream to derive a scalar brightness from the simulated (or real-time) sky color luminance. """ speed: float = 1.0 # simulation speed (ignored when use_real_time) use_real_time: bool = False # use wall clock instead of simulation latitude: float = 50.0 # affects sunrise/sunset in real-time mode 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["speed"] = self.speed d["use_real_time"] = self.use_real_time d["latitude"] = self.latitude d["min_value"] = self.min_value d["max_value"] = self.max_value return d @classmethod def from_dict(cls, data: dict) -> "DaylightValueSource": common = _parse_common_fields(data) return cls( **common, source_type="daylight", speed=float(data.get("speed") or 1.0), use_real_time=bool(data.get("use_real_time", False)), latitude=float(data.get("latitude") or 50.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, ) # -- Source type registry -- # Maps source_type string to its subclass for factory dispatch. _VALUE_SOURCE_MAP: Dict[str, Type[ValueSource]] = { "static": StaticValueSource, "animated": AnimatedValueSource, "audio": AudioValueSource, "adaptive_time": AdaptiveValueSource, "adaptive_scene": AdaptiveValueSource, "daylight": DaylightValueSource, }