alexei.dolgolyov/wled-screen-controller-mixed
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add audio capture engine template system with multi-backend support

Browse Source 
Introduces an engine+template abstraction for audio capture, mirroring the
existing screen capture engine pattern. This enables multiple audio backends
(WASAPI for Windows, sounddevice for cross-platform) with per-source
engine configuration via reusable templates.

Backend:
- AudioCaptureEngine ABC with WasapiEngine and SounddeviceEngine implementations
- AudioEngineRegistry for engine discovery and factory creation
- AudioAnalyzer class decouples FFT/RMS/beat analysis from engine-specific capture
- ManagedAudioStream wraps engine stream + analyzer in background thread
- AudioCaptureTemplate model and AudioTemplateStore with JSON CRUD
- AudioCaptureManager keyed by (engine_type, device_index, is_loopback)
- Auto-migration: default template created on startup, assigned to existing sources
- Full REST API: CRUD for audio templates + engine listing with availability flags
- audio_template_id added to MultichannelAudioSource model and API schemas

Frontend:
- Audio template cards in Streams > Audio tab with engine badge and config details
- Audio template editor modal with engine selector and dynamic config fields
- Audio template dropdown in multichannel audio source editor
- Template name crosslink badge on multichannel audio source cards
- Confirm modal z-index fix (always stacks above editor modals)
- i18n keys for EN and RU

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Admin
2026-02-26 13:55:46 +03:00
parent cbbaa852ed
commit bae2166bc2
35 changed files with 2163 additions and 402 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
server/src/wled_controller/storage/audio_source.py
View File

@@ -33,6 +33,7 @@ class AudioSource:
# Subclass fields default to None for forward compat
"device_index": None,
"is_loopback": None,
"audio_template_id": None,
"audio_source_id": None,
"channel": None,
}
@@ -74,6 +75,7 @@ class AudioSource:
created_at=created_at, updated_at=updated_at, description=description,
device_index=int(data.get("device_index", -1)),
is_loopback=bool(data.get("is_loopback", True)),
audio_template_id=data.get("audio_template_id"),
)
@@ -87,11 +89,13 @@ class MultichannelAudioSource(AudioSource):
device_index: int = -1 # -1 = default device
is_loopback: bool = True # True = WASAPI loopback (system audio)
audio_template_id: Optional[str] = None # references AudioCaptureTemplate
def to_dict(self) -> dict:
d = super().to_dict()
d["device_index"] = self.device_index
d["is_loopback"] = self.is_loopback
d["audio_template_id"] = self.audio_template_id
return d

32
server/src/wled_controller/storage/audio_source_store.py
View File

@@ -99,6 +99,7 @@ class AudioSourceStore:
audio_source_id: Optional[str] = None,
channel: Optional[str] = None,
description: Optional[str] = None,
audio_template_id: Optional[str] = None,
) -> AudioSource:
if not name or not name.strip():
raise ValueError("Name is required")
@@ -135,6 +136,7 @@ class AudioSourceStore:
created_at=now, updated_at=now, description=description,
device_index=device_index if device_index is not None else -1,
is_loopback=bool(is_loopback) if is_loopback is not None else True,
audio_template_id=audio_template_id,
)
self._sources[sid] = source
@@ -152,6 +154,7 @@ class AudioSourceStore:
audio_source_id: Optional[str] = None,
channel: Optional[str] = None,
description: Optional[str] = None,
audio_template_id: Optional[str] = None,
) -> AudioSource:
if source_id not in self._sources:
raise ValueError(f"Audio source not found: {source_id}")
@@ -172,6 +175,8 @@ class AudioSourceStore:
source.device_index = device_index
if is_loopback is not None:
source.is_loopback = bool(is_loopback)
if audio_template_id is not None:
source.audio_template_id = audio_template_id
elif isinstance(source, MonoAudioSource):
if audio_source_id is not None:
parent = self._sources.get(audio_source_id)
@@ -210,8 +215,8 @@ class AudioSourceStore:
# ── Resolution ───────────────────────────────────────────────────
def resolve_audio_source(self, source_id: str) -> Tuple[int, bool, str]:
"""Resolve any audio source to (device_index, is_loopback, channel).
def resolve_audio_source(self, source_id: str) -> Tuple[int, bool, str, Optional[str]]:
"""Resolve any audio source to (device_index, is_loopback, channel, audio_template_id).
Accepts both MultichannelAudioSource (defaults to "mono" channel)
and MonoAudioSource (follows reference chain to parent multichannel).
@@ -222,7 +227,7 @@ class AudioSourceStore:
source = self.get_source(source_id)
if isinstance(source, MultichannelAudioSource):
return source.device_index, source.is_loopback, "mono"
return source.device_index, source.is_loopback, "mono", source.audio_template_id
if isinstance(source, MonoAudioSource):
parent = self.get_source(source.audio_source_id)
@@ -230,11 +235,11 @@ class AudioSourceStore:
raise ValueError(
f"Mono source {source_id} references non-multichannel source {source.audio_source_id}"
)
return parent.device_index, parent.is_loopback, source.channel
return parent.device_index, parent.is_loopback, source.channel, parent.audio_template_id
raise ValueError(f"Audio source {source_id} is not a valid audio source")
def resolve_mono_source(self, mono_id: str) -> Tuple[int, bool, str]:
def resolve_mono_source(self, mono_id: str) -> Tuple[int, bool, str, Optional[str]]:
"""Backward-compatible wrapper for resolve_audio_source()."""
return self.resolve_audio_source(mono_id)
@@ -330,3 +335,20 @@ class AudioSourceStore:
logger.info(f"Migration complete: migrated {migrated} audio CSS entities")
else:
logger.debug("No audio CSS entities needed migration")
def migrate_add_default_template(self, audio_template_store) -> None:
"""Assign the default audio template to multichannel sources that have no template."""
default_id = audio_template_store.get_default_template_id()
if not default_id:
logger.debug("No default audio template — skipping template migration")
return
migrated = 0
for source in self._sources.values():
if isinstance(source, MultichannelAudioSource) and not source.audio_template_id:
source.audio_template_id = default_id
migrated += 1
if migrated > 0:
self._save()
logger.info(f"Assigned default audio template to {migrated} multichannel sources")

47
server/src/wled_controller/storage/audio_template.py Normal file
View File

@@ -0,0 +1,47 @@
"""Audio capture template data model."""
from dataclasses import dataclass
from datetime import datetime
from typing import Any, Dict, Optional
@dataclass
class AudioCaptureTemplate:
"""Represents an audio capture template configuration."""
id: str
name: str
engine_type: str
engine_config: Dict[str, Any]
created_at: datetime
updated_at: datetime
description: Optional[str] = None
def to_dict(self) -> dict:
"""Convert template to dictionary."""
return {
"id": self.id,
"name": self.name,
"engine_type": self.engine_type,
"engine_config": self.engine_config,
"created_at": self.created_at.isoformat(),
"updated_at": self.updated_at.isoformat(),
"description": self.description,
}
@classmethod
def from_dict(cls, data: dict) -> "AudioCaptureTemplate":
"""Create template from dictionary."""
return cls(
id=data["id"],
name=data["name"],
engine_type=data["engine_type"],
engine_config=data.get("engine_config", {}),
created_at=datetime.fromisoformat(data["created_at"])
if isinstance(data.get("created_at"), str)
else data.get("created_at", datetime.utcnow()),
updated_at=datetime.fromisoformat(data["updated_at"])
if isinstance(data.get("updated_at"), str)
else data.get("updated_at", datetime.utcnow()),
description=data.get("description"),
)

212
server/src/wled_controller/storage/audio_template_store.py Normal file
View File

@@ -0,0 +1,212 @@
"""Audio template 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.core.audio.factory import AudioEngineRegistry
from wled_controller.storage.audio_template import AudioCaptureTemplate
from wled_controller.utils import get_logger
logger = get_logger(__name__)
class AudioTemplateStore:
"""Storage for audio capture templates.
All templates are persisted to the JSON file.
On startup, if no templates exist, one is auto-created using the
highest-priority available engine.
"""
def __init__(self, file_path: str):
self.file_path = Path(file_path)
self._templates: Dict[str, AudioCaptureTemplate] = {}
self._load()
self._ensure_initial_template()
def _ensure_initial_template(self) -> None:
"""Auto-create a template if none exist, using the best available engine."""
if self._templates:
return
best_engine = AudioEngineRegistry.get_best_available_engine()
if not best_engine:
logger.warning("No audio engines available, cannot create initial template")
return
engine_class = AudioEngineRegistry.get_engine(best_engine)
default_config = engine_class.get_default_config()
now = datetime.utcnow()
template_id = f"atpl_{uuid.uuid4().hex[:8]}"
template = AudioCaptureTemplate(
id=template_id,
name="Default Audio",
engine_type=best_engine,
engine_config=default_config,
created_at=now,
updated_at=now,
description=f"Default audio template using {best_engine.upper()} engine",
)
self._templates[template_id] = template
self._save()
logger.info(
f"Auto-created initial audio template: {template.name} "
f"({template_id}, engine={best_engine})"
)
def _load(self) -> None:
"""Load templates from file."""
if not self.file_path.exists():
return
try:
with open(self.file_path, "r", encoding="utf-8") as f:
data = json.load(f)
templates_data = data.get("templates", {})
loaded = 0
for template_id, template_dict in templates_data.items():
try:
template = AudioCaptureTemplate.from_dict(template_dict)
self._templates[template_id] = template
loaded += 1
except Exception as e:
logger.error(
f"Failed to load audio template {template_id}: {e}",
exc_info=True,
)
if loaded > 0:
logger.info(f"Loaded {loaded} audio templates from storage")
except Exception as e:
logger.error(f"Failed to load audio templates from {self.file_path}: {e}")
raise
logger.info(f"Audio template store initialized with {len(self._templates)} templates")
def _save(self) -> None:
"""Save all templates to file."""
try:
self.file_path.parent.mkdir(parents=True, exist_ok=True)
templates_dict = {
template_id: template.to_dict()
for template_id, template in self._templates.items()
}
data = {
"version": "1.0.0",
"templates": templates_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 audio templates to {self.file_path}: {e}")
raise
def get_all_templates(self) -> List[AudioCaptureTemplate]:
return list(self._templates.values())
def get_template(self, template_id: str) -> AudioCaptureTemplate:
if template_id not in self._templates:
raise ValueError(f"Audio template not found: {template_id}")
return self._templates[template_id]
def get_default_template_id(self) -> Optional[str]:
"""Return the ID of the first template, or None if none exist."""
if self._templates:
return next(iter(self._templates))
return None
def create_template(
self,
name: str,
engine_type: str,
engine_config: Dict[str, any],
description: Optional[str] = None,
) -> AudioCaptureTemplate:
for template in self._templates.values():
if template.name == name:
raise ValueError(f"Audio template with name '{name}' already exists")
template_id = f"atpl_{uuid.uuid4().hex[:8]}"
now = datetime.utcnow()
template = AudioCaptureTemplate(
id=template_id,
name=name,
engine_type=engine_type,
engine_config=engine_config,
created_at=now,
updated_at=now,
description=description,
)
self._templates[template_id] = template
self._save()
logger.info(f"Created audio template: {name} ({template_id})")
return template
def update_template(
self,
template_id: str,
name: Optional[str] = None,
engine_type: Optional[str] = None,
engine_config: Optional[Dict[str, any]] = None,
description: Optional[str] = None,
) -> AudioCaptureTemplate:
if template_id not in self._templates:
raise ValueError(f"Audio template not found: {template_id}")
template = self._templates[template_id]
if name is not None:
template.name = name
if engine_type is not None:
template.engine_type = engine_type
if engine_config is not None:
template.engine_config = engine_config
if description is not None:
template.description = description
template.updated_at = datetime.utcnow()
self._save()
logger.info(f"Updated audio template: {template_id}")
return template
def delete_template(self, template_id: str, audio_source_store=None) -> None:
"""Delete a template.
Args:
template_id: Template ID
audio_source_store: Optional AudioSourceStore for reference check
Raises:
ValueError: If template not found or still referenced
"""
if template_id not in self._templates:
raise ValueError(f"Audio template not found: {template_id}")
# Check if any multichannel audio source references this template
if audio_source_store is not None:
from wled_controller.storage.audio_source import MultichannelAudioSource
for source in audio_source_store.get_all_sources():
if (
isinstance(source, MultichannelAudioSource)
and getattr(source, "audio_template_id", None) == template_id
):
raise ValueError(
f"Cannot delete audio template '{template_id}': "
f"referenced by audio source '{source.name}' ({source.id})"
)
del self._templates[template_id]
self._save()
logger.info(f"Deleted audio template: {template_id}")