feat: add band_extract audio source type for frequency band filtering
Lint & Test / test (push) Failing after 29s

New audio source type that filters a parent source to a specific frequency
band (bass 20-250Hz, mid 250-4kHz, treble 4k-20kHz, or custom range).
Supports chaining with frequency range intersection and cycle detection.
Band filtering applied in both CSS audio streams and test WebSocket.
This commit is contained in:
2026-03-24 19:36:11 +03:00
parent a62e2f474d
commit ae0a5cb160
18 changed files with 512 additions and 66 deletions
@@ -42,6 +42,9 @@ def _to_response(source: AudioSource) -> AudioSourceResponse:
audio_template_id=getattr(source, "audio_template_id", None),
audio_source_id=getattr(source, "audio_source_id", None),
channel=getattr(source, "channel", None),
band=getattr(source, "band", None),
freq_low=getattr(source, "freq_low", None),
freq_high=getattr(source, "freq_high", None),
description=source.description,
tags=source.tags,
created_at=source.created_at,
@@ -52,7 +55,7 @@ def _to_response(source: AudioSource) -> AudioSourceResponse:
@router.get("/api/v1/audio-sources", response_model=AudioSourceListResponse, tags=["Audio Sources"])
async def list_audio_sources(
_auth: AuthRequired,
source_type: Optional[str] = Query(None, description="Filter by source_type: multichannel or mono"),
source_type: Optional[str] = Query(None, description="Filter by source_type: multichannel, mono, or band_extract"),
store: AudioSourceStore = Depends(get_audio_source_store),
):
"""List all audio sources, optionally filtered by type."""
@@ -83,6 +86,9 @@ async def create_audio_source(
description=data.description,
audio_template_id=data.audio_template_id,
tags=data.tags,
band=data.band,
freq_low=data.freq_low,
freq_high=data.freq_high,
)
fire_entity_event("audio_source", "created", source.id)
return _to_response(source)
@@ -126,6 +132,9 @@ async def update_audio_source(
description=data.description,
audio_template_id=data.audio_template_id,
tags=data.tags,
band=data.band,
freq_low=data.freq_low,
freq_high=data.freq_high,
)
fire_entity_event("audio_source", "updated", source_id)
return _to_response(source)
@@ -182,17 +191,28 @@ async def test_audio_source_ws(
await websocket.close(code=4001, reason="Unauthorized")
return
# Resolve source → device info
# Resolve source → device info + optional band filter
store = get_audio_source_store()
template_store = get_audio_template_store()
manager = get_processor_manager()
try:
device_index, is_loopback, channel, audio_template_id = store.resolve_audio_source(source_id)
resolved = store.resolve_audio_source(source_id)
except ValueError as e:
await websocket.close(code=4004, reason=str(e))
return
device_index = resolved.device_index
is_loopback = resolved.is_loopback
channel = resolved.channel
audio_template_id = resolved.audio_template_id
# Precompute band mask if this is a band_extract source
band_mask = None
if resolved.freq_low is not None and resolved.freq_high is not None:
from wled_controller.core.audio.band_filter import compute_band_mask
band_mask = compute_band_mask(resolved.freq_low, resolved.freq_high)
# Resolve template → engine_type + config
engine_type = None
engine_config = None
@@ -233,6 +253,11 @@ async def test_audio_source_ws(
spectrum = analysis.spectrum
rms = analysis.rms
# Apply band filter if present
if band_mask is not None:
from wled_controller.core.audio.band_filter import apply_band_filter
spectrum, rms = apply_band_filter(spectrum, rms, band_mask)
await websocket.send_json({
"spectrum": spectrum.tolist(),
"rms": round(rms, 4),
@@ -10,14 +10,18 @@ class AudioSourceCreate(BaseModel):
"""Request to create an audio source."""
name: str = Field(description="Source name", min_length=1, max_length=100)
source_type: Literal["multichannel", "mono"] = Field(description="Source type")
source_type: Literal["multichannel", "mono", "band_extract"] = Field(description="Source type")
# multichannel fields
device_index: Optional[int] = Field(None, description="Audio device index (-1 = default)")
is_loopback: Optional[bool] = Field(None, description="True for system audio (WASAPI loopback)")
audio_template_id: Optional[str] = Field(None, description="Audio capture template ID")
# mono fields
audio_source_id: Optional[str] = Field(None, description="Parent multichannel audio source ID")
audio_source_id: Optional[str] = Field(None, description="Parent audio source ID")
channel: Optional[str] = Field(None, description="Channel: mono|left|right")
# band_extract fields
band: Optional[str] = Field(None, description="Band preset: bass|mid|treble|custom")
freq_low: Optional[float] = Field(None, description="Low frequency bound (Hz)", ge=20, le=20000)
freq_high: Optional[float] = Field(None, description="High frequency bound (Hz)", ge=20, le=20000)
description: Optional[str] = Field(None, description="Optional description", max_length=500)
tags: List[str] = Field(default_factory=list, description="User-defined tags")
@@ -29,8 +33,11 @@ class AudioSourceUpdate(BaseModel):
device_index: Optional[int] = Field(None, description="Audio device index (-1 = default)")
is_loopback: Optional[bool] = Field(None, description="True for system audio (WASAPI loopback)")
audio_template_id: Optional[str] = Field(None, description="Audio capture template ID")
audio_source_id: Optional[str] = Field(None, description="Parent multichannel audio source ID")
audio_source_id: Optional[str] = Field(None, description="Parent audio source ID")
channel: Optional[str] = Field(None, description="Channel: mono|left|right")
band: Optional[str] = Field(None, description="Band preset: bass|mid|treble|custom")
freq_low: Optional[float] = Field(None, description="Low frequency bound (Hz)", ge=20, le=20000)
freq_high: Optional[float] = Field(None, description="High frequency bound (Hz)", ge=20, le=20000)
description: Optional[str] = Field(None, description="Optional description", max_length=500)
tags: Optional[List[str]] = None
@@ -40,12 +47,15 @@ class AudioSourceResponse(BaseModel):
id: str = Field(description="Source ID")
name: str = Field(description="Source name")
source_type: str = Field(description="Source type: multichannel or mono")
source_type: str = Field(description="Source type: multichannel, mono, or band_extract")
device_index: Optional[int] = Field(None, description="Audio device index")
is_loopback: Optional[bool] = Field(None, description="WASAPI loopback mode")
audio_template_id: Optional[str] = Field(None, description="Audio capture template ID")
audio_source_id: Optional[str] = Field(None, description="Parent multichannel source ID")
audio_source_id: Optional[str] = Field(None, description="Parent audio source ID")
channel: Optional[str] = Field(None, description="Channel: mono|left|right")
band: Optional[str] = Field(None, description="Band preset: bass|mid|treble|custom")
freq_low: Optional[float] = Field(None, description="Low frequency bound (Hz)")
freq_high: Optional[float] = Field(None, description="High frequency bound (Hz)")
description: Optional[str] = Field(None, description="Description")
tags: List[str] = Field(default_factory=list, description="User-defined tags")
created_at: datetime = Field(description="Creation timestamp")