alexei.dolgolyov
88b3ecd5e1
- Add real-time value source test: WebSocket endpoint streams get_value() at ~20Hz, frontend renders scrolling time-series chart with min/max/current stats - Add auto-gain for audio value sources: rolling peak normalization with slow decay, sensitivity range increased to 0.1-20.0 - Always show brightness overlay on LED preview when brightness source is set - Refactor ValueStreamManager to shared ref-counted streams (value streams produce scalars, not LED-count-dependent, so sharing is correct) - Simplify acquire/release API: remove consumer_id parameter since streams are no longer consumer-dependent Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
221 lines
8.5 KiB
Python
221 lines
8.5 KiB
Python
"""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
|
||
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
|
||
|
||
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,
|
||
"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")
|
||
|
||
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),
|
||
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,
|
||
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,
|
||
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,
|
||
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
|