Add audio-reactive color strip sources, improve delete error messages

Add new "audio" color strip source type with three visualization modes
(spectrum analyzer, beat pulse, VU meter) supporting WASAPI loopback and
microphone input via PyAudioWPatch. Includes shared audio capture with
ref counting, real-time FFT spectrum analysis, and beat detection.

Improve all referential integrity 409 error messages across delete
endpoints to include specific names of referencing entities instead of
generic "one or more" messages.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-02-23 11:56:54 +03:00
parent 2657f46e5d
commit bbd2ac9910
24 changed files with 1247 additions and 86 deletions

View File

@@ -9,6 +9,7 @@ Current types:
StaticColorStripSource — constant solid color fills all LEDs
GradientColorStripSource — linear gradient across all LEDs from user-defined color stops
ColorCycleColorStripSource — smoothly cycles through a user-defined list of colors
AudioColorStripSource — audio-reactive visualization (spectrum, beat pulse, VU meter)
"""
from dataclasses import dataclass, field
@@ -72,6 +73,11 @@ class ColorStripSource:
"scale": None,
"mirror": None,
"layers": None,
"visualization_mode": None,
"audio_device_index": None,
"audio_loopback": None,
"sensitivity": None,
"color_peak": None,
}
@staticmethod
@@ -148,6 +154,26 @@ class ColorStripSource:
led_count=data.get("led_count") or 0,
)
if source_type == "audio":
raw_color = data.get("color")
color = raw_color if isinstance(raw_color, list) and len(raw_color) == 3 else [0, 255, 0]
raw_peak = data.get("color_peak")
color_peak = raw_peak if isinstance(raw_peak, list) and len(raw_peak) == 3 else [255, 0, 0]
return AudioColorStripSource(
id=sid, name=name, source_type="audio",
created_at=created_at, updated_at=updated_at, description=description,
visualization_mode=data.get("visualization_mode") or "spectrum",
audio_device_index=int(data.get("audio_device_index", -1)),
audio_loopback=bool(data.get("audio_loopback", True)),
sensitivity=float(data.get("sensitivity") or 1.0),
smoothing=float(data.get("smoothing") or 0.3),
palette=data.get("palette") or "rainbow",
color=color,
color_peak=color_peak,
led_count=data.get("led_count") or 0,
mirror=bool(data.get("mirror", False)),
)
if source_type == "effect":
raw_color = data.get("color")
color = (
@@ -328,6 +354,41 @@ class EffectColorStripSource(ColorStripSource):
return d
@dataclass
class AudioColorStripSource(ColorStripSource):
"""Color strip source driven by audio input (microphone or system audio).
visualization_mode selects the rendering algorithm:
spectrum, beat_pulse, vu_meter.
LED count auto-sizes from the connected device when led_count == 0.
"""
visualization_mode: str = "spectrum" # spectrum | beat_pulse | vu_meter
audio_device_index: int = -1 # -1 = default input device
audio_loopback: bool = True # True = WASAPI loopback (system audio)
sensitivity: float = 1.0 # gain multiplier (0.15.0)
smoothing: float = 0.3 # temporal smoothing (0.01.0)
palette: str = "rainbow" # named color palette
color: list = field(default_factory=lambda: [0, 255, 0]) # base RGB for VU meter
color_peak: list = field(default_factory=lambda: [255, 0, 0]) # peak RGB for VU meter
led_count: int = 0 # 0 = use device LED count
mirror: bool = False # mirror spectrum from center outward
def to_dict(self) -> dict:
d = super().to_dict()
d["visualization_mode"] = self.visualization_mode
d["audio_device_index"] = self.audio_device_index
d["audio_loopback"] = self.audio_loopback
d["sensitivity"] = self.sensitivity
d["smoothing"] = self.smoothing
d["palette"] = self.palette
d["color"] = list(self.color)
d["color_peak"] = list(self.color_peak)
d["led_count"] = self.led_count
d["mirror"] = self.mirror
return d
@dataclass
class CompositeColorStripSource(ColorStripSource):
"""Color strip source that composites (stacks) multiple other sources as layers.

View File

@@ -8,6 +8,7 @@ from typing import Dict, List, Optional
from wled_controller.core.capture.calibration import CalibrationConfig, calibration_to_dict
from wled_controller.storage.color_strip_source import (
AudioColorStripSource,
ColorCycleColorStripSource,
ColorStripSource,
CompositeColorStripSource,
@@ -118,6 +119,11 @@ class ColorStripStore:
scale: float = 1.0,
mirror: bool = False,
layers: Optional[list] = None,
visualization_mode: str = "spectrum",
audio_device_index: int = -1,
audio_loopback: bool = True,
sensitivity: float = 1.0,
color_peak: Optional[list] = None,
) -> ColorStripSource:
"""Create a new color strip source.
@@ -196,6 +202,27 @@ class ColorStripStore:
scale=float(scale) if scale else 1.0,
mirror=bool(mirror),
)
elif source_type == "audio":
rgb = color if isinstance(color, list) and len(color) == 3 else [0, 255, 0]
peak = color_peak if isinstance(color_peak, list) and len(color_peak) == 3 else [255, 0, 0]
source = AudioColorStripSource(
id=source_id,
name=name,
source_type="audio",
created_at=now,
updated_at=now,
description=description,
visualization_mode=visualization_mode or "spectrum",
audio_device_index=audio_device_index if audio_device_index is not None else -1,
audio_loopback=bool(audio_loopback),
sensitivity=float(sensitivity) if sensitivity else 1.0,
smoothing=float(smoothing) if smoothing else 0.3,
palette=palette or "rainbow",
color=rgb,
color_peak=peak,
led_count=led_count,
mirror=bool(mirror),
)
elif source_type == "composite":
source = CompositeColorStripSource(
id=source_id,
@@ -262,6 +289,11 @@ class ColorStripStore:
scale: Optional[float] = None,
mirror: Optional[bool] = None,
layers: Optional[list] = None,
visualization_mode: Optional[str] = None,
audio_device_index: Optional[int] = None,
audio_loopback: Optional[bool] = None,
sensitivity: Optional[float] = None,
color_peak: Optional[list] = None,
) -> ColorStripSource:
"""Update an existing color strip source.
@@ -342,6 +374,27 @@ class ColorStripStore:
source.scale = float(scale)
if mirror is not None:
source.mirror = bool(mirror)
elif isinstance(source, AudioColorStripSource):
if visualization_mode is not None:
source.visualization_mode = visualization_mode
if audio_device_index is not None:
source.audio_device_index = audio_device_index
if audio_loopback is not None:
source.audio_loopback = bool(audio_loopback)
if sensitivity is not None:
source.sensitivity = float(sensitivity)
if smoothing is not None:
source.smoothing = float(smoothing)
if palette is not None:
source.palette = palette
if color is not None and isinstance(color, list) and len(color) == 3:
source.color = color
if color_peak is not None and isinstance(color_peak, list) and len(color_peak) == 3:
source.color_peak = color_peak
if led_count is not None:
source.led_count = led_count
if mirror is not None:
source.mirror = bool(mirror)
elif isinstance(source, CompositeColorStripSource):
if layers is not None and isinstance(layers, list):
source.layers = layers
@@ -368,20 +421,13 @@ class ColorStripStore:
logger.info(f"Deleted color strip source: {source_id}")
def is_referenced_by_composite(self, source_id: str) -> bool:
"""Check if this source is referenced as a layer in any composite source."""
def get_composites_referencing(self, source_id: str) -> List[str]:
"""Return names of composite sources that reference a given source as a layer."""
names = []
for source in self._sources.values():
if isinstance(source, CompositeColorStripSource):
for layer in source.layers:
if layer.get("source_id") == source_id:
return True
return False
def is_referenced_by_target(self, source_id: str, target_store) -> bool:
"""Check if this source is referenced by any picture target."""
from wled_controller.storage.wled_picture_target import WledPictureTarget
for target in target_store.get_all_targets():
if isinstance(target, WledPictureTarget) and target.color_strip_source_id == source_id:
return True
return False
names.append(source.name)
break
return names

View File

@@ -207,19 +207,11 @@ class PatternTemplateStore:
logger.info(f"Deleted pattern template: {template_id}")
def is_referenced_by(self, template_id: str, picture_target_store) -> bool:
"""Check if this template is referenced by any key colors target.
Args:
template_id: Template ID to check
picture_target_store: PictureTargetStore instance
Returns:
True if any KC target references this template
"""
def get_targets_referencing(self, template_id: str, picture_target_store) -> List[str]:
"""Return names of KC targets that reference this template."""
from wled_controller.storage.key_colors_picture_target import KeyColorsPictureTarget
for target in picture_target_store.get_all_targets():
if isinstance(target, KeyColorsPictureTarget) and target.settings.pattern_template_id == template_id:
return True
return False
return [
target.name for target in picture_target_store.get_all_targets()
if isinstance(target, KeyColorsPictureTarget) and target.settings.pattern_template_id == template_id
]

View File

@@ -301,17 +301,9 @@ class PictureSourceStore:
logger.info(f"Deleted picture source: {stream_id}")
def is_referenced_by_target(self, stream_id: str, target_store) -> bool:
"""Check if this stream is referenced by any picture target.
Args:
stream_id: Stream ID to check
target_store: PictureTargetStore instance
Returns:
True if any target references this stream
"""
return target_store.is_referenced_by_source(stream_id)
def get_targets_referencing(self, stream_id: str, target_store) -> List[str]:
"""Return names of targets that reference this stream."""
return target_store.get_targets_referencing_source(stream_id)
def resolve_stream_chain(self, stream_id: str) -> dict:
"""Resolve a stream chain to get the terminal stream and collected postprocessing templates.

View File

@@ -251,19 +251,19 @@ class PictureTargetStore:
if isinstance(t, WledPictureTarget) and t.device_id == device_id
]
def is_referenced_by_source(self, source_id: str) -> bool:
"""Check if any KC target directly references a picture source."""
for target in self._targets.values():
if isinstance(target, KeyColorsPictureTarget) and target.picture_source_id == source_id:
return True
return False
def get_targets_referencing_source(self, source_id: str) -> List[str]:
"""Return names of KC targets that reference a picture source."""
return [
target.name for target in self._targets.values()
if isinstance(target, KeyColorsPictureTarget) and target.picture_source_id == source_id
]
def is_referenced_by_color_strip_source(self, css_id: str) -> bool:
"""Check if any WLED target references a color strip source."""
for target in self._targets.values():
if isinstance(target, WledPictureTarget) and target.color_strip_source_id == css_id:
return True
return False
def get_targets_referencing_css(self, css_id: str) -> List[str]:
"""Return names of LED targets that reference a color strip source."""
return [
target.name for target in self._targets.values()
if isinstance(target, WledPictureTarget) and target.color_strip_source_id == css_id
]
def count(self) -> int:
"""Get number of targets."""

View File

@@ -220,17 +220,9 @@ class PostprocessingTemplateStore:
logger.info(f"Deleted postprocessing template: {template_id}")
def is_referenced_by(self, template_id: str, picture_source_store) -> bool:
"""Check if this template is referenced by any picture source.
Args:
template_id: Template ID to check
picture_source_store: PictureSourceStore instance
Returns:
True if any picture source references this template
"""
for stream in picture_source_store.get_all_streams():
if isinstance(stream, ProcessedPictureSource) and stream.postprocessing_template_id == template_id:
return True
return False
def get_sources_referencing(self, template_id: str, picture_source_store) -> List[str]:
"""Return names of picture sources that reference this template."""
return [
stream.name for stream in picture_source_store.get_all_streams()
if isinstance(stream, ProcessedPictureSource) and stream.postprocessing_template_id == template_id
]