feat(processed-audio-sources): phase 1 - audio filter framework
Add the foundation for audio processing filters, mirroring the existing picture filter/postprocessing template system: - AudioFilter base class, AudioFilterRegistry, AudioFilterOptionDef - AudioProcessingTemplate dataclass + SQLite-backed store - audio_filter_template meta-filter with recursive resolution - Full REST API: CRUD templates + filter registry discovery - Dependency injection wired in dependencies.py and main.py
This commit is contained in:
@@ -28,6 +28,8 @@ from .routes.assets import router as assets_router
|
||||
from .routes.home_assistant import router as home_assistant_router
|
||||
from .routes.mqtt import router as mqtt_router
|
||||
from .routes.game_integration import router as game_integration_router
|
||||
from .routes.audio_processing_templates import router as audio_processing_templates_router
|
||||
from .routes.audio_filters import router as audio_filters_router
|
||||
|
||||
router = APIRouter()
|
||||
router.include_router(system_router)
|
||||
@@ -56,5 +58,7 @@ router.include_router(assets_router)
|
||||
router.include_router(home_assistant_router)
|
||||
router.include_router(mqtt_router)
|
||||
router.include_router(game_integration_router)
|
||||
router.include_router(audio_processing_templates_router)
|
||||
router.include_router(audio_filters_router)
|
||||
|
||||
__all__ = ["router"]
|
||||
|
||||
@@ -37,6 +37,7 @@ from wled_controller.storage.game_integration_store import GameIntegrationStore
|
||||
from wled_controller.core.game_integration.event_bus import GameEventBus
|
||||
from wled_controller.storage.mqtt_source_store import MQTTSourceStore
|
||||
from wled_controller.core.mqtt.mqtt_manager import MQTTManager
|
||||
from wled_controller.storage.audio_processing_template_store import AudioProcessingTemplateStore
|
||||
|
||||
T = TypeVar("T")
|
||||
|
||||
@@ -163,6 +164,10 @@ def get_mqtt_manager() -> MQTTManager:
|
||||
return _get("mqtt_manager", "MQTT manager")
|
||||
|
||||
|
||||
def get_audio_processing_template_store() -> AudioProcessingTemplateStore:
|
||||
return _get("audio_processing_template_store", "Audio processing template store")
|
||||
|
||||
|
||||
def get_database() -> Database:
|
||||
return _get("database", "Database")
|
||||
|
||||
@@ -227,6 +232,7 @@ def init_dependencies(
|
||||
game_event_bus: GameEventBus | None = None,
|
||||
mqtt_store: MQTTSourceStore | None = None,
|
||||
mqtt_manager: MQTTManager | None = None,
|
||||
audio_processing_template_store: AudioProcessingTemplateStore | None = None,
|
||||
):
|
||||
"""Initialize global dependencies."""
|
||||
_deps.update(
|
||||
@@ -260,5 +266,6 @@ def init_dependencies(
|
||||
"game_event_bus": game_event_bus,
|
||||
"mqtt_store": mqtt_store,
|
||||
"mqtt_manager": mqtt_manager,
|
||||
"audio_processing_template_store": audio_processing_template_store,
|
||||
}
|
||||
)
|
||||
|
||||
@@ -0,0 +1,61 @@
|
||||
"""Audio filter registry endpoint."""
|
||||
|
||||
from fastapi import APIRouter, Depends
|
||||
|
||||
from wled_controller.api.auth import AuthRequired
|
||||
from wled_controller.api.dependencies import get_audio_processing_template_store
|
||||
from wled_controller.api.schemas.filters import (
|
||||
FilterOptionDefSchema,
|
||||
FilterTypeListResponse,
|
||||
FilterTypeResponse,
|
||||
)
|
||||
from wled_controller.core.audio.filters import AudioFilterRegistry
|
||||
from wled_controller.storage.audio_processing_template_store import AudioProcessingTemplateStore
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.get(
|
||||
"/api/v1/audio-filters",
|
||||
response_model=FilterTypeListResponse,
|
||||
tags=["Audio Filters"],
|
||||
)
|
||||
async def list_audio_filter_types(
|
||||
_auth: AuthRequired,
|
||||
apt_store: AudioProcessingTemplateStore = Depends(get_audio_processing_template_store),
|
||||
):
|
||||
"""List all available audio filter types and their options schemas."""
|
||||
all_filters = AudioFilterRegistry.get_all()
|
||||
|
||||
# Pre-build template choices for the audio_filter_template filter
|
||||
template_choices = None
|
||||
if apt_store:
|
||||
try:
|
||||
templates = apt_store.get_all_templates()
|
||||
template_choices = [{"value": t.id, "label": t.name} for t in templates]
|
||||
except Exception:
|
||||
template_choices = []
|
||||
|
||||
responses = []
|
||||
for filter_id, filter_cls in all_filters.items():
|
||||
schema = filter_cls.get_options_schema()
|
||||
opt_schemas = []
|
||||
for opt in schema:
|
||||
d = opt.to_dict()
|
||||
# Dynamically populate template_id choices for audio_filter_template
|
||||
if (
|
||||
filter_id == "audio_filter_template"
|
||||
and opt.key == "template_id"
|
||||
and template_choices is not None
|
||||
):
|
||||
d["choices"] = template_choices
|
||||
opt_schemas.append(FilterOptionDefSchema(**d))
|
||||
responses.append(
|
||||
FilterTypeResponse(
|
||||
filter_id=filter_id,
|
||||
filter_name=filter_cls.filter_name,
|
||||
options_schema=opt_schemas,
|
||||
)
|
||||
)
|
||||
|
||||
return FilterTypeListResponse(filters=responses, count=len(responses))
|
||||
@@ -0,0 +1,165 @@
|
||||
"""Audio processing template routes."""
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
|
||||
from wled_controller.api.auth import AuthRequired
|
||||
from wled_controller.api.dependencies import (
|
||||
fire_entity_event,
|
||||
get_audio_processing_template_store,
|
||||
)
|
||||
from wled_controller.api.schemas.audio_processing import (
|
||||
AudioProcessingTemplateCreate,
|
||||
AudioProcessingTemplateListResponse,
|
||||
AudioProcessingTemplateResponse,
|
||||
AudioProcessingTemplateUpdate,
|
||||
)
|
||||
from wled_controller.api.schemas.filters import FilterInstanceSchema
|
||||
from wled_controller.core.filters.filter_instance import FilterInstance
|
||||
from wled_controller.storage.audio_processing_template_store import AudioProcessingTemplateStore
|
||||
from wled_controller.storage.base_store import EntityNotFoundError
|
||||
from wled_controller.utils import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
def _apt_to_response(t) -> AudioProcessingTemplateResponse:
|
||||
"""Convert an AudioProcessingTemplate to its API response."""
|
||||
return AudioProcessingTemplateResponse(
|
||||
id=t.id,
|
||||
name=t.name,
|
||||
filters=[FilterInstanceSchema(filter_id=f.filter_id, options=f.options) for f in t.filters],
|
||||
created_at=t.created_at,
|
||||
updated_at=t.updated_at,
|
||||
description=t.description,
|
||||
tags=t.tags,
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/api/v1/audio-processing-templates",
|
||||
response_model=AudioProcessingTemplateListResponse,
|
||||
tags=["Audio Processing Templates"],
|
||||
)
|
||||
async def list_audio_processing_templates(
|
||||
_auth: AuthRequired,
|
||||
store: AudioProcessingTemplateStore = Depends(get_audio_processing_template_store),
|
||||
):
|
||||
"""List all audio processing templates."""
|
||||
templates = store.get_all_templates()
|
||||
responses = [_apt_to_response(t) for t in templates]
|
||||
return AudioProcessingTemplateListResponse(templates=responses, count=len(responses))
|
||||
|
||||
|
||||
@router.post(
|
||||
"/api/v1/audio-processing-templates",
|
||||
response_model=AudioProcessingTemplateResponse,
|
||||
tags=["Audio Processing Templates"],
|
||||
status_code=201,
|
||||
)
|
||||
async def create_audio_processing_template(
|
||||
data: AudioProcessingTemplateCreate,
|
||||
_auth: AuthRequired,
|
||||
store: AudioProcessingTemplateStore = Depends(get_audio_processing_template_store),
|
||||
):
|
||||
"""Create a new audio processing template."""
|
||||
try:
|
||||
filters = [FilterInstance(f.filter_id, f.options) for f in data.filters]
|
||||
template = store.create_template(
|
||||
name=data.name,
|
||||
filters=filters,
|
||||
description=data.description,
|
||||
tags=data.tags,
|
||||
)
|
||||
fire_entity_event("audio_processing_template", "created", template.id)
|
||||
return _apt_to_response(template)
|
||||
except EntityNotFoundError as e:
|
||||
raise HTTPException(status_code=404, detail=str(e))
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=400, detail=str(e))
|
||||
except Exception as e:
|
||||
logger.error("Failed to create audio processing template: %s", e, exc_info=True)
|
||||
raise HTTPException(status_code=500, detail="Internal server error")
|
||||
|
||||
|
||||
@router.get(
|
||||
"/api/v1/audio-processing-templates/{template_id}",
|
||||
response_model=AudioProcessingTemplateResponse,
|
||||
tags=["Audio Processing Templates"],
|
||||
)
|
||||
async def get_audio_processing_template(
|
||||
template_id: str,
|
||||
_auth: AuthRequired,
|
||||
store: AudioProcessingTemplateStore = Depends(get_audio_processing_template_store),
|
||||
):
|
||||
"""Get audio processing template by ID."""
|
||||
try:
|
||||
template = store.get_template(template_id)
|
||||
return _apt_to_response(template)
|
||||
except ValueError:
|
||||
raise HTTPException(
|
||||
status_code=404, detail=f"Audio processing template {template_id} not found"
|
||||
)
|
||||
|
||||
|
||||
@router.put(
|
||||
"/api/v1/audio-processing-templates/{template_id}",
|
||||
response_model=AudioProcessingTemplateResponse,
|
||||
tags=["Audio Processing Templates"],
|
||||
)
|
||||
async def update_audio_processing_template(
|
||||
template_id: str,
|
||||
data: AudioProcessingTemplateUpdate,
|
||||
_auth: AuthRequired,
|
||||
store: AudioProcessingTemplateStore = Depends(get_audio_processing_template_store),
|
||||
):
|
||||
"""Update an audio processing template."""
|
||||
try:
|
||||
filters = (
|
||||
[FilterInstance(f.filter_id, f.options) for f in data.filters]
|
||||
if data.filters is not None
|
||||
else None
|
||||
)
|
||||
template = store.update_template(
|
||||
template_id=template_id,
|
||||
name=data.name,
|
||||
filters=filters,
|
||||
description=data.description,
|
||||
tags=data.tags,
|
||||
)
|
||||
fire_entity_event("audio_processing_template", "updated", template_id)
|
||||
return _apt_to_response(template)
|
||||
except EntityNotFoundError as e:
|
||||
raise HTTPException(status_code=404, detail=str(e))
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=400, detail=str(e))
|
||||
except Exception as e:
|
||||
logger.error("Failed to update audio processing template: %s", e, exc_info=True)
|
||||
raise HTTPException(status_code=500, detail="Internal server error")
|
||||
|
||||
|
||||
@router.delete(
|
||||
"/api/v1/audio-processing-templates/{template_id}",
|
||||
status_code=204,
|
||||
tags=["Audio Processing Templates"],
|
||||
)
|
||||
async def delete_audio_processing_template(
|
||||
template_id: str,
|
||||
_auth: AuthRequired,
|
||||
store: AudioProcessingTemplateStore = Depends(get_audio_processing_template_store),
|
||||
):
|
||||
"""Delete an audio processing template."""
|
||||
try:
|
||||
# TODO: Phase 3 will add reference checks against ProcessedAudioSource
|
||||
store.delete_template(template_id)
|
||||
fire_entity_event("audio_processing_template", "deleted", template_id)
|
||||
except HTTPException:
|
||||
raise
|
||||
except EntityNotFoundError as e:
|
||||
raise HTTPException(status_code=404, detail=str(e))
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=400, detail=str(e))
|
||||
except Exception as e:
|
||||
logger.error("Failed to delete audio processing template: %s", e, exc_info=True)
|
||||
raise HTTPException(status_code=500, detail="Internal server error")
|
||||
@@ -0,0 +1,53 @@
|
||||
"""Audio processing template schemas."""
|
||||
|
||||
from datetime import datetime
|
||||
from typing import List, Optional
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from .filters import FilterInstanceSchema
|
||||
|
||||
|
||||
class AudioProcessingTemplateCreate(BaseModel):
|
||||
"""Request to create an audio processing template."""
|
||||
|
||||
name: str = Field(description="Template name", min_length=1, max_length=100)
|
||||
filters: List[FilterInstanceSchema] = Field(
|
||||
default_factory=list, description="Ordered list of audio filter instances"
|
||||
)
|
||||
description: Optional[str] = Field(None, description="Template description", max_length=500)
|
||||
tags: List[str] = Field(default_factory=list, description="User-defined tags")
|
||||
|
||||
|
||||
class AudioProcessingTemplateUpdate(BaseModel):
|
||||
"""Request to update an audio processing template."""
|
||||
|
||||
name: Optional[str] = Field(None, description="Template name", min_length=1, max_length=100)
|
||||
filters: Optional[List[FilterInstanceSchema]] = Field(
|
||||
None, description="Ordered list of audio filter instances"
|
||||
)
|
||||
description: Optional[str] = Field(None, description="Template description", max_length=500)
|
||||
tags: Optional[List[str]] = None
|
||||
|
||||
|
||||
class AudioProcessingTemplateResponse(BaseModel):
|
||||
"""Audio processing template information response."""
|
||||
|
||||
id: str = Field(description="Template ID")
|
||||
name: str = Field(description="Template name")
|
||||
filters: List[FilterInstanceSchema] = Field(
|
||||
description="Ordered list of audio filter instances"
|
||||
)
|
||||
tags: List[str] = Field(default_factory=list, description="User-defined tags")
|
||||
created_at: datetime = Field(description="Creation timestamp")
|
||||
updated_at: datetime = Field(description="Last update timestamp")
|
||||
description: Optional[str] = Field(None, description="Template description")
|
||||
|
||||
|
||||
class AudioProcessingTemplateListResponse(BaseModel):
|
||||
"""List of audio processing templates response."""
|
||||
|
||||
templates: List[AudioProcessingTemplateResponse] = Field(
|
||||
description="List of audio processing templates"
|
||||
)
|
||||
count: int = Field(description="Number of templates")
|
||||
@@ -0,0 +1,17 @@
|
||||
"""Audio filter system.
|
||||
|
||||
Provides a pluggable filter architecture for audio analysis postprocessing.
|
||||
Import this package to ensure all built-in filters are registered.
|
||||
"""
|
||||
|
||||
from wled_controller.core.audio.filters.base import AudioFilter, AudioFilterOptionDef
|
||||
from wled_controller.core.audio.filters.registry import AudioFilterRegistry
|
||||
|
||||
# Import individual filters to trigger auto-registration
|
||||
import wled_controller.core.audio.filters.audio_filter_template # noqa: F401
|
||||
|
||||
__all__ = [
|
||||
"AudioFilter",
|
||||
"AudioFilterOptionDef",
|
||||
"AudioFilterRegistry",
|
||||
]
|
||||
@@ -0,0 +1,39 @@
|
||||
"""Audio Filter Template meta-filter -- references another audio processing template.
|
||||
|
||||
This filter exists in the registry for UI discovery only. It is never
|
||||
instantiated at runtime: the audio processing pipeline expands it into the
|
||||
referenced template's filters when building the filter chain.
|
||||
"""
|
||||
|
||||
from typing import List
|
||||
|
||||
from wled_controller.core.audio.analysis import AudioAnalysis
|
||||
from wled_controller.core.audio.filters.base import AudioFilter, AudioFilterOptionDef
|
||||
from wled_controller.core.audio.filters.registry import AudioFilterRegistry
|
||||
|
||||
|
||||
@AudioFilterRegistry.register
|
||||
class AudioFilterTemplateFilter(AudioFilter):
|
||||
"""Include another audio filter template's chain at this position."""
|
||||
|
||||
filter_id = "audio_filter_template"
|
||||
filter_name = "Audio Filter Template"
|
||||
|
||||
@classmethod
|
||||
def get_options_schema(cls) -> List[AudioFilterOptionDef]:
|
||||
return [
|
||||
AudioFilterOptionDef(
|
||||
key="template_id",
|
||||
label="Template",
|
||||
option_type="select",
|
||||
default="",
|
||||
min_value=None,
|
||||
max_value=None,
|
||||
step=None,
|
||||
choices=[], # populated dynamically by GET /api/v1/audio-filters
|
||||
),
|
||||
]
|
||||
|
||||
def process(self, analysis: AudioAnalysis) -> AudioAnalysis:
|
||||
# Never called -- expanded at pipeline build time.
|
||||
return analysis
|
||||
@@ -0,0 +1,118 @@
|
||||
"""Base classes for the audio filter system."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from wled_controller.core.audio.analysis import AudioAnalysis
|
||||
|
||||
|
||||
@dataclass
|
||||
class AudioFilterOptionDef:
|
||||
"""Describes a single configurable option for an audio filter."""
|
||||
|
||||
key: str
|
||||
label: str
|
||||
option_type: str # "float" | "int" | "bool" | "select" | "string"
|
||||
default: Any
|
||||
min_value: Any
|
||||
max_value: Any
|
||||
step: Any
|
||||
choices: Optional[List[Dict[str, str]]] = None # for "select": [{value, label}]
|
||||
max_length: Optional[int] = None # for "string" type
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
d = {
|
||||
"key": self.key,
|
||||
"label": self.label,
|
||||
"type": self.option_type,
|
||||
"default": self.default,
|
||||
"min_value": self.min_value,
|
||||
"max_value": self.max_value,
|
||||
"step": self.step,
|
||||
}
|
||||
if self.choices is not None:
|
||||
d["choices"] = self.choices
|
||||
if self.max_length is not None:
|
||||
d["max_length"] = self.max_length
|
||||
return d
|
||||
|
||||
|
||||
class AudioFilter(ABC):
|
||||
"""Base class for all audio filters.
|
||||
|
||||
Each filter operates on an AudioAnalysis snapshot and returns
|
||||
a new (possibly transformed) AudioAnalysis.
|
||||
|
||||
Stateful filters (e.g. peak hold, envelope follower) must override
|
||||
``is_stateful`` to return True and implement ``reset()``.
|
||||
"""
|
||||
|
||||
filter_id: str = ""
|
||||
filter_name: str = ""
|
||||
|
||||
def __init__(self, options: Dict[str, Any]):
|
||||
"""Initialize filter with validated options."""
|
||||
self.options = self.validate_options(options)
|
||||
|
||||
@property
|
||||
def is_stateful(self) -> bool:
|
||||
"""Whether this filter maintains internal state across calls.
|
||||
|
||||
Stateful filters need per-stream instances and reset() support.
|
||||
"""
|
||||
return False
|
||||
|
||||
def reset(self) -> None:
|
||||
"""Reset internal state. Override in stateful filters."""
|
||||
|
||||
@classmethod
|
||||
@abstractmethod
|
||||
def get_options_schema(cls) -> List[AudioFilterOptionDef]:
|
||||
"""Return the list of configurable options for this filter type."""
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
def process(self, analysis: AudioAnalysis) -> AudioAnalysis:
|
||||
"""Process an audio analysis snapshot.
|
||||
|
||||
Args:
|
||||
analysis: Input AudioAnalysis snapshot.
|
||||
|
||||
Returns:
|
||||
New AudioAnalysis with transformations applied.
|
||||
"""
|
||||
...
|
||||
|
||||
@classmethod
|
||||
def validate_options(cls, options: dict) -> dict:
|
||||
"""Validate and clamp options against the schema. Returns cleaned dict."""
|
||||
schema = cls.get_options_schema()
|
||||
cleaned = {}
|
||||
for opt_def in schema:
|
||||
raw = options.get(opt_def.key, opt_def.default)
|
||||
if opt_def.option_type == "float":
|
||||
val = float(raw)
|
||||
elif opt_def.option_type == "int":
|
||||
val = int(raw)
|
||||
elif opt_def.option_type == "bool":
|
||||
val = bool(raw) if not isinstance(raw, bool) else raw
|
||||
elif opt_def.option_type == "select":
|
||||
val = str(raw) if raw is not None else opt_def.default
|
||||
elif opt_def.option_type == "string":
|
||||
val = str(raw) if raw is not None else opt_def.default
|
||||
else:
|
||||
val = raw
|
||||
# Clamp to range (skip for non-numeric types)
|
||||
if opt_def.option_type not in ("bool", "select", "string"):
|
||||
if opt_def.min_value is not None and val < opt_def.min_value:
|
||||
val = opt_def.min_value
|
||||
if opt_def.max_value is not None and val > opt_def.max_value:
|
||||
val = opt_def.max_value
|
||||
cleaned[opt_def.key] = val
|
||||
return cleaned
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"{self.__class__.__name__}({self.options})"
|
||||
@@ -0,0 +1,53 @@
|
||||
"""Audio filter registry for discovering and instantiating audio filters."""
|
||||
|
||||
from typing import Dict, Type
|
||||
|
||||
from wled_controller.core.audio.filters.base import AudioFilter
|
||||
from wled_controller.utils import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
class AudioFilterRegistry:
|
||||
"""Singleton registry of all available audio filter types."""
|
||||
|
||||
_filters: Dict[str, Type[AudioFilter]] = {}
|
||||
|
||||
@classmethod
|
||||
def register(cls, filter_cls: Type[AudioFilter]) -> Type[AudioFilter]:
|
||||
"""Register a filter class. Can be used as a decorator."""
|
||||
filter_id = filter_cls.filter_id
|
||||
if not filter_id:
|
||||
raise ValueError(f"Filter class {filter_cls.__name__} must define filter_id")
|
||||
if filter_id in cls._filters:
|
||||
logger.warning(f"Overwriting audio filter registration for '{filter_id}'")
|
||||
cls._filters[filter_id] = filter_cls
|
||||
logger.debug(f"Registered audio filter: {filter_id} ({filter_cls.__name__})")
|
||||
return filter_cls
|
||||
|
||||
@classmethod
|
||||
def get(cls, filter_id: str) -> Type[AudioFilter]:
|
||||
"""Get a filter class by ID.
|
||||
|
||||
Raises:
|
||||
ValueError: If filter_id is not registered.
|
||||
"""
|
||||
if filter_id not in cls._filters:
|
||||
raise ValueError(f"Unknown audio filter type: '{filter_id}'")
|
||||
return cls._filters[filter_id]
|
||||
|
||||
@classmethod
|
||||
def get_all(cls) -> Dict[str, Type[AudioFilter]]:
|
||||
"""Get all registered audio filter types."""
|
||||
return dict(cls._filters)
|
||||
|
||||
@classmethod
|
||||
def create_instance(cls, filter_id: str, options: dict) -> AudioFilter:
|
||||
"""Create a filter instance from a filter_id and options dict."""
|
||||
filter_cls = cls.get(filter_id)
|
||||
return filter_cls(options)
|
||||
|
||||
@classmethod
|
||||
def is_registered(cls, filter_id: str) -> bool:
|
||||
"""Check if a filter ID is registered."""
|
||||
return filter_id in cls._filters
|
||||
@@ -51,6 +51,8 @@ from wled_controller.core.game_integration.community_loader import register_comm
|
||||
from wled_controller.core.mqtt.mqtt_service import MQTTService
|
||||
from wled_controller.core.mqtt.mqtt_manager import MQTTManager
|
||||
from wled_controller.storage.mqtt_source_store import MQTTSourceStore
|
||||
from wled_controller.storage.audio_processing_template_store import AudioProcessingTemplateStore
|
||||
import wled_controller.core.audio.filters # noqa: F401 — trigger audio filter auto-registration
|
||||
from wled_controller.core.devices.mqtt_client import set_mqtt_service
|
||||
from wled_controller.core.backup.auto_backup import AutoBackupEngine
|
||||
from wled_controller.core.processing.os_notification_listener import OsNotificationListener
|
||||
@@ -104,6 +106,7 @@ weather_manager = WeatherManager(weather_source_store)
|
||||
ha_store = HomeAssistantStore(db)
|
||||
ha_manager = HomeAssistantManager(ha_store)
|
||||
mqtt_source_store = MQTTSourceStore(db)
|
||||
audio_processing_template_store = AudioProcessingTemplateStore(db)
|
||||
game_integration_store = GameIntegrationStore(db)
|
||||
game_event_bus = GameEventBus()
|
||||
register_community_adapters()
|
||||
@@ -231,6 +234,7 @@ async def lifespan(app: FastAPI):
|
||||
game_event_bus=game_event_bus,
|
||||
mqtt_store=mqtt_source_store,
|
||||
mqtt_manager=mqtt_manager,
|
||||
audio_processing_template_store=audio_processing_template_store,
|
||||
)
|
||||
|
||||
# Register devices in processor manager for health monitoring
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
"""Audio processing template data model."""
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime, timezone
|
||||
from typing import List, Optional
|
||||
|
||||
from wled_controller.core.filters.filter_instance import FilterInstance
|
||||
|
||||
|
||||
@dataclass
|
||||
class AudioProcessingTemplate:
|
||||
"""Audio processing settings template containing an ordered list of audio filters."""
|
||||
|
||||
id: str
|
||||
name: str
|
||||
filters: List[FilterInstance]
|
||||
created_at: datetime
|
||||
updated_at: datetime
|
||||
description: Optional[str] = None
|
||||
tags: List[str] = field(default_factory=list)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
"""Convert template to dictionary."""
|
||||
return {
|
||||
"id": self.id,
|
||||
"name": self.name,
|
||||
"filters": [f.to_dict() for f in self.filters],
|
||||
"created_at": self.created_at.isoformat(),
|
||||
"updated_at": self.updated_at.isoformat(),
|
||||
"description": self.description,
|
||||
"tags": self.tags,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "AudioProcessingTemplate":
|
||||
"""Create template from dictionary."""
|
||||
filters = [FilterInstance.from_dict(f) for f in data.get("filters", [])]
|
||||
|
||||
return cls(
|
||||
id=data["id"],
|
||||
name=data["name"],
|
||||
filters=filters,
|
||||
created_at=(
|
||||
datetime.fromisoformat(data["created_at"])
|
||||
if isinstance(data.get("created_at"), str)
|
||||
else data.get("created_at", datetime.now(timezone.utc))
|
||||
),
|
||||
updated_at=(
|
||||
datetime.fromisoformat(data["updated_at"])
|
||||
if isinstance(data.get("updated_at"), str)
|
||||
else data.get("updated_at", datetime.now(timezone.utc))
|
||||
),
|
||||
description=data.get("description"),
|
||||
tags=data.get("tags", []),
|
||||
)
|
||||
@@ -0,0 +1,125 @@
|
||||
"""Audio processing template storage using SQLite."""
|
||||
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from typing import List, Optional
|
||||
|
||||
from wled_controller.core.audio.filters.registry import AudioFilterRegistry
|
||||
from wled_controller.core.filters.filter_instance import FilterInstance
|
||||
from wled_controller.storage.audio_processing_template import AudioProcessingTemplate
|
||||
from wled_controller.storage.base_sqlite_store import BaseSqliteStore
|
||||
from wled_controller.storage.database import Database
|
||||
from wled_controller.utils import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
class AudioProcessingTemplateStore(BaseSqliteStore[AudioProcessingTemplate]):
|
||||
"""Storage for audio processing templates.
|
||||
|
||||
All templates are persisted to the database.
|
||||
"""
|
||||
|
||||
_table_name = "audio_processing_templates"
|
||||
_entity_name = "Audio processing template"
|
||||
_version = "1.0.0"
|
||||
|
||||
def __init__(self, db: Database):
|
||||
super().__init__(db, AudioProcessingTemplate.from_dict)
|
||||
|
||||
# Backward-compatible aliases
|
||||
get_all_templates = BaseSqliteStore.get_all
|
||||
get_template = BaseSqliteStore.get
|
||||
delete_template = BaseSqliteStore.delete
|
||||
|
||||
def create_template(
|
||||
self,
|
||||
name: str,
|
||||
filters: Optional[List[FilterInstance]] = None,
|
||||
description: Optional[str] = None,
|
||||
tags: Optional[List[str]] = None,
|
||||
) -> AudioProcessingTemplate:
|
||||
self._check_name_unique(name)
|
||||
|
||||
if filters is None:
|
||||
filters = []
|
||||
|
||||
# Validate filter IDs
|
||||
for fi in filters:
|
||||
if not AudioFilterRegistry.is_registered(fi.filter_id):
|
||||
raise ValueError(f"Unknown audio filter type: '{fi.filter_id}'")
|
||||
|
||||
template_id = f"apt_{uuid.uuid4().hex[:8]}"
|
||||
now = datetime.now(timezone.utc)
|
||||
|
||||
template = AudioProcessingTemplate(
|
||||
id=template_id,
|
||||
name=name,
|
||||
filters=filters,
|
||||
created_at=now,
|
||||
updated_at=now,
|
||||
description=description,
|
||||
tags=tags or [],
|
||||
)
|
||||
|
||||
self._items[template_id] = template
|
||||
self._save_item(template_id, template)
|
||||
|
||||
logger.info(f"Created audio processing template: {name} ({template_id})")
|
||||
return template
|
||||
|
||||
def update_template(
|
||||
self,
|
||||
template_id: str,
|
||||
name: Optional[str] = None,
|
||||
filters: Optional[List[FilterInstance]] = None,
|
||||
description: Optional[str] = None,
|
||||
tags: Optional[List[str]] = None,
|
||||
) -> AudioProcessingTemplate:
|
||||
template = self.get(template_id)
|
||||
|
||||
if name is not None:
|
||||
self._check_name_unique(name, exclude_id=template_id)
|
||||
template.name = name
|
||||
if filters is not None:
|
||||
# Validate filter IDs
|
||||
for fi in filters:
|
||||
if not AudioFilterRegistry.is_registered(fi.filter_id):
|
||||
raise ValueError(f"Unknown audio filter type: '{fi.filter_id}'")
|
||||
template.filters = filters
|
||||
if description is not None:
|
||||
template.description = description
|
||||
if tags is not None:
|
||||
template.tags = tags
|
||||
|
||||
template.updated_at = datetime.now(timezone.utc)
|
||||
self._save_item(template_id, template)
|
||||
|
||||
logger.info(f"Updated audio processing template: {template_id}")
|
||||
return template
|
||||
|
||||
def resolve_filter_instances(self, filter_instances, _visited=None):
|
||||
"""Recursively resolve filter instances, expanding audio_filter_template references.
|
||||
|
||||
Returns a flat list of FilterInstance objects with no audio_filter_template entries.
|
||||
"""
|
||||
if _visited is None:
|
||||
_visited = set()
|
||||
resolved = []
|
||||
for fi in filter_instances:
|
||||
if fi.filter_id == "audio_filter_template":
|
||||
template_id = fi.options.get("template_id", "")
|
||||
if not template_id or template_id in _visited:
|
||||
continue
|
||||
try:
|
||||
ref_template = self.get_template(template_id)
|
||||
_visited.add(template_id)
|
||||
resolved.extend(self.resolve_filter_instances(ref_template.filters, _visited))
|
||||
_visited.discard(template_id)
|
||||
except ValueError:
|
||||
logger.warning(
|
||||
f"Referenced audio filter template '{template_id}' not found, skipping"
|
||||
)
|
||||
else:
|
||||
resolved.append(fi)
|
||||
return resolved
|
||||
Reference in New Issue
Block a user