Add value sources for dynamic brightness control on LED targets

Introduces a new Value Source entity that produces a scalar float (0.0-1.0)
for dynamic brightness modulation. Three subtypes: Static (constant),
Animated (sine/triangle/square/sawtooth waveform), and Audio-reactive
(RMS/peak/beat from mono audio source). Value sources can be optionally
attached to LED targets to control brightness each frame.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

server/src/wled_controller/storage/picture_target_store.py

@@ -102,6 +102,7 @@ class PictureTargetStore:
         target_type: str,
         device_id: str = "",
         color_strip_source_id: str = "",
+        brightness_value_source_id: str = "",
         fps: int = 30,
         keepalive_interval: float = 1.0,
         state_check_interval: int = DEFAULT_STATE_CHECK_INTERVAL,
@@ -132,6 +133,7 @@ class PictureTargetStore:
                 target_type="led",
                 device_id=device_id,
                 color_strip_source_id=color_strip_source_id,
+                brightness_value_source_id=brightness_value_source_id,
                 fps=fps,
                 keepalive_interval=keepalive_interval,
                 state_check_interval=state_check_interval,
@@ -165,6 +167,7 @@ class PictureTargetStore:
         name: Optional[str] = None,
         device_id: Optional[str] = None,
         color_strip_source_id: Optional[str] = None,
+        brightness_value_source_id: Optional[str] = None,
         fps: Optional[int] = None,
         keepalive_interval: Optional[float] = None,
         state_check_interval: Optional[int] = None,
@@ -191,6 +194,7 @@ class PictureTargetStore:
             name=name,
             device_id=device_id,
             color_strip_source_id=color_strip_source_id,
+            brightness_value_source_id=brightness_value_source_id,
             fps=fps,
             keepalive_interval=keepalive_interval,
             state_check_interval=state_check_interval,

server/src/wled_controller/storage/value_source.py

@@ -0,0 +1,154 @@
+"""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.  Three types:
+  StaticValueSource    — constant float value
+  AnimatedValueSource  — periodic waveform (sine, triangle, square, sawtooth)
+  AudioValueSource     — audio-reactive scalar (RMS, peak, beat detection)
+"""
+
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Optional
+
+
+@dataclass
+class ValueSource:
+    """Base class for value source configurations."""
+
+    id: str
+    name: str
+    source_type: str  # "static" | "animated" | "audio"
+    created_at: datetime
+    updated_at: datetime
+    description: Optional[str] = None
+
+    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,
+            # 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,
+        }
+
+    @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")
+
+        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.utcnow()
+        )
+        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.utcnow()
+        )
+
+        if source_type == "animated":
+            return AnimatedValueSource(
+                id=sid, name=name, source_type="animated",
+                created_at=created_at, updated_at=updated_at, description=description,
+                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,
+                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),
+            )
+
+        # Default: "static" type
+        return StaticValueSource(
+            id=sid, name=name, source_type="static",
+            created_at=created_at, updated_at=updated_at, description=description,
+            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 a MonoAudioSource
+    mode: str = "rms"            # rms | peak | beat
+    sensitivity: float = 1.0     # gain multiplier (0.1–5.0)
+    smoothing: float = 0.3       # temporal smoothing (0.0–1.0)
+
+    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
+        return d

server/src/wled_controller/storage/value_source_store.py

@@ -0,0 +1,213 @@
+"""Value source storage using JSON files."""
+
+import json
+import uuid
+from datetime import datetime
+from pathlib import Path
+from typing import Dict, List, Optional
+
+from wled_controller.storage.value_source import (
+    AnimatedValueSource,
+    AudioValueSource,
+    StaticValueSource,
+    ValueSource,
+)
+from wled_controller.utils import get_logger
+
+logger = get_logger(__name__)
+
+
+class ValueSourceStore:
+    """Persistent storage for value sources."""
+
+    def __init__(self, file_path: str):
+        self.file_path = Path(file_path)
+        self._sources: Dict[str, ValueSource] = {}
+        self._load()
+
+    def _load(self) -> None:
+        if not self.file_path.exists():
+            logger.info("Value source store file not found — starting empty")
+            return
+
+        try:
+            with open(self.file_path, "r", encoding="utf-8") as f:
+                data = json.load(f)
+
+            sources_data = data.get("value_sources", {})
+            loaded = 0
+            for source_id, source_dict in sources_data.items():
+                try:
+                    source = ValueSource.from_dict(source_dict)
+                    self._sources[source_id] = source
+                    loaded += 1
+                except Exception as e:
+                    logger.error(
+                        f"Failed to load value source {source_id}: {e}",
+                        exc_info=True,
+                    )
+
+            if loaded > 0:
+                logger.info(f"Loaded {loaded} value sources from storage")
+
+        except Exception as e:
+            logger.error(f"Failed to load value sources from {self.file_path}: {e}")
+            raise
+
+        logger.info(f"Value source store initialized with {len(self._sources)} sources")
+
+    def _save(self) -> None:
+        try:
+            self.file_path.parent.mkdir(parents=True, exist_ok=True)
+
+            sources_dict = {
+                sid: source.to_dict()
+                for sid, source in self._sources.items()
+            }
+
+            data = {
+                "version": "1.0.0",
+                "value_sources": sources_dict,
+            }
+
+            with open(self.file_path, "w", encoding="utf-8") as f:
+                json.dump(data, f, indent=2, ensure_ascii=False)
+
+        except Exception as e:
+            logger.error(f"Failed to save value sources to {self.file_path}: {e}")
+            raise
+
+    # ── CRUD ─────────────────────────────────────────────────────────
+
+    def get_all_sources(self) -> List[ValueSource]:
+        return list(self._sources.values())
+
+    def get_source(self, source_id: str) -> ValueSource:
+        if source_id not in self._sources:
+            raise ValueError(f"Value source not found: {source_id}")
+        return self._sources[source_id]
+
+    def create_source(
+        self,
+        name: str,
+        source_type: str,
+        value: Optional[float] = None,
+        waveform: Optional[str] = None,
+        speed: Optional[float] = None,
+        min_value: Optional[float] = None,
+        max_value: Optional[float] = None,
+        audio_source_id: Optional[str] = None,
+        mode: Optional[str] = None,
+        sensitivity: Optional[float] = None,
+        smoothing: Optional[float] = None,
+        description: Optional[str] = None,
+    ) -> ValueSource:
+        if not name or not name.strip():
+            raise ValueError("Name is required")
+
+        if source_type not in ("static", "animated", "audio"):
+            raise ValueError(f"Invalid source type: {source_type}")
+
+        for source in self._sources.values():
+            if source.name == name:
+                raise ValueError(f"Value source with name '{name}' already exists")
+
+        sid = f"vs_{uuid.uuid4().hex[:8]}"
+        now = datetime.utcnow()
+
+        if source_type == "static":
+            source: ValueSource = StaticValueSource(
+                id=sid, name=name, source_type="static",
+                created_at=now, updated_at=now, description=description,
+                value=value if value is not None else 1.0,
+            )
+        elif source_type == "animated":
+            source = AnimatedValueSource(
+                id=sid, name=name, source_type="animated",
+                created_at=now, updated_at=now, description=description,
+                waveform=waveform or "sine",
+                speed=speed if speed is not None else 10.0,
+                min_value=min_value if min_value is not None else 0.0,
+                max_value=max_value if max_value is not None else 1.0,
+            )
+        elif source_type == "audio":
+            source = AudioValueSource(
+                id=sid, name=name, source_type="audio",
+                created_at=now, updated_at=now, description=description,
+                audio_source_id=audio_source_id or "",
+                mode=mode or "rms",
+                sensitivity=sensitivity if sensitivity is not None else 1.0,
+                smoothing=smoothing if smoothing is not None else 0.3,
+            )
+
+        self._sources[sid] = source
+        self._save()
+
+        logger.info(f"Created value source: {name} ({sid}, type={source_type})")
+        return source
+
+    def update_source(
+        self,
+        source_id: str,
+        name: Optional[str] = None,
+        value: Optional[float] = None,
+        waveform: Optional[str] = None,
+        speed: Optional[float] = None,
+        min_value: Optional[float] = None,
+        max_value: Optional[float] = None,
+        audio_source_id: Optional[str] = None,
+        mode: Optional[str] = None,
+        sensitivity: Optional[float] = None,
+        smoothing: Optional[float] = None,
+        description: Optional[str] = None,
+    ) -> ValueSource:
+        if source_id not in self._sources:
+            raise ValueError(f"Value source not found: {source_id}")
+
+        source = self._sources[source_id]
+
+        if name is not None:
+            for other in self._sources.values():
+                if other.id != source_id and other.name == name:
+                    raise ValueError(f"Value source with name '{name}' already exists")
+            source.name = name
+
+        if description is not None:
+            source.description = description
+
+        if isinstance(source, StaticValueSource):
+            if value is not None:
+                source.value = value
+        elif isinstance(source, AnimatedValueSource):
+            if waveform is not None:
+                source.waveform = waveform
+            if speed is not None:
+                source.speed = speed
+            if min_value is not None:
+                source.min_value = min_value
+            if max_value is not None:
+                source.max_value = max_value
+        elif isinstance(source, AudioValueSource):
+            if audio_source_id is not None:
+                source.audio_source_id = audio_source_id
+            if mode is not None:
+                source.mode = mode
+            if sensitivity is not None:
+                source.sensitivity = sensitivity
+            if smoothing is not None:
+                source.smoothing = smoothing
+
+        source.updated_at = datetime.utcnow()
+        self._save()
+
+        logger.info(f"Updated value source: {source_id}")
+        return source
+
+    def delete_source(self, source_id: str) -> None:
+        if source_id not in self._sources:
+            raise ValueError(f"Value source not found: {source_id}")
+
+        del self._sources[source_id]
+        self._save()
+
+        logger.info(f"Deleted value source: {source_id}")

server/src/wled_controller/storage/wled_picture_target.py

@@ -15,6 +15,7 @@ class WledPictureTarget(PictureTarget):
 
     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
@@ -29,11 +30,13 @@ class WledPictureTarget(PictureTarget):
                 fps=self.fps,
                 keepalive_interval=self.keepalive_interval,
                 state_check_interval=self.state_check_interval,
+                brightness_value_source_id=self.brightness_value_source_id,
             )
 
     def sync_with_manager(self, manager, *, settings_changed: bool,
                           css_changed: bool = False,
-                          device_changed: bool = False) -> None:
+                          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, {
@@ -45,8 +48,11 @@ class WledPictureTarget(PictureTarget):
             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,
                       description=None, **_kwargs) -> None:
         """Apply mutable field updates for WLED targets."""
@@ -55,6 +61,8 @@ class WledPictureTarget(PictureTarget):
             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:
@@ -71,6 +79,7 @@ class WledPictureTarget(PictureTarget):
         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
@@ -95,6 +104,7 @@ class WledPictureTarget(PictureTarget):
             target_type="led",
             device_id=data.get("device_id", ""),
             color_strip_source_id=css_id,
+            brightness_value_source_id=data.get("brightness_value_source_id", ""),
             fps=data.get("fps", 30),
             keepalive_interval=data.get("keepalive_interval", data.get("standby_interval", 1.0)),
             state_check_interval=data.get("state_check_interval", DEFAULT_STATE_CHECK_INTERVAL),