"""Audio source data model with inheritance-based source types. An AudioSource represents a reusable audio input configuration: MultichannelAudioSource — wraps a physical audio device (index + loopback flag) MonoAudioSource — extracts a single channel from a multichannel source """ from dataclasses import dataclass, field from datetime import datetime, timezone from typing import List, Optional @dataclass class AudioSource: """Base class for audio source configurations.""" id: str name: str source_type: str # "multichannel" | "mono" created_at: datetime updated_at: datetime description: Optional[str] = None tags: List[str] = field(default_factory=list) 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, "tags": self.tags, # Subclass fields default to None for forward compat "device_index": None, "is_loopback": None, "audio_template_id": None, "audio_source_id": None, "channel": None, } @staticmethod def from_dict(data: dict) -> "AudioSource": """Factory: dispatch to the correct subclass based on source_type.""" source_type: str = data.get("source_type", "multichannel") or "multichannel" sid: str = data["id"] name: str = data["name"] description: str | None = data.get("description") tags: list = data.get("tags", []) 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.now(timezone.utc) ) 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.now(timezone.utc) ) if source_type == "mono": return MonoAudioSource( id=sid, name=name, source_type="mono", created_at=created_at, updated_at=updated_at, description=description, tags=tags, audio_source_id=data.get("audio_source_id") or "", channel=data.get("channel") or "mono", ) # Default: multichannel return MultichannelAudioSource( id=sid, name=name, source_type="multichannel", created_at=created_at, updated_at=updated_at, description=description, tags=tags, device_index=int(data.get("device_index", -1)), is_loopback=bool(data.get("is_loopback", True)), audio_template_id=data.get("audio_template_id"), ) @dataclass class MultichannelAudioSource(AudioSource): """Audio source wrapping a physical audio device. Captures all channels from the device. For WASAPI loopback devices (system audio output), set is_loopback=True. """ 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 @dataclass class MonoAudioSource(AudioSource): """Audio source that extracts a single channel from a multichannel source. References a MultichannelAudioSource and selects which channel to use: mono (L+R mix), left, or right. """ audio_source_id: str = "" # references a MultichannelAudioSource channel: str = "mono" # mono | left | right def to_dict(self) -> dict: d = super().to_dict() d["audio_source_id"] = self.audio_source_id d["channel"] = self.channel return d