my_voice_assistant_v2/app/dependencies.py

from dataclasses import dataclass

from app.config import Settings, settings
from app.errors import UnknownComponentError
from app.audio.router import AudioRouter
from app.audio.endpoints.input.local_default import LocalDefaultInput
from app.audio.endpoints.input.bluetooth import BluetoothInput
from app.audio.endpoints.input.mobile_ws import MobileWebSocketInput
from app.audio.endpoints.input.mobile_webrtc import MobileWebRTCInput
from app.audio.endpoints.output.local_default import LocalDefaultOutput
from app.audio.endpoints.output.bluetooth import BluetoothOutput
from app.audio.endpoints.output.mobile_ws import MobileWebSocketOutput
from app.audio.endpoints.output.mobile_webrtc import MobileWebRTCOutput
from app.audio.endpoints.output.loopback import LoopbackOutput
from app.providers.stt.openrouter import OpenRouterSTTProvider
from app.providers.stt.faster_whisper import FasterWhisperProvider
from app.providers.llm.local_openai_compatible import LocalOpenAICompatibleLLM
from app.providers.llm.openrouter import OpenRouterLLMProvider
from app.providers.tts.openrouter import OpenRouterTTSProvider
from app.providers.tts.chatterbox import ChatterboxTTSProvider
from app.providers.tts.piper import PiperTTSProvider
from app.pipeline.input_cleaner import InputCleaner
from app.pipeline.spoken_response_adapter import SpokenResponseAdapter
from app.pipeline.tts_normalizer import TTSNormalizer
from app.core.orchestrator import Orchestrator
from app.core.session_manager import SessionManager

session_manager = SessionManager()

# ---------------------------------------------------------------------------
# Provider-Registries: Modul austauschbar via Name, ohne Kern-Code zu aendern.
# Ein neuer Provider = ein Eintrag. Unbekannter Name -> UnknownComponentError.
# ---------------------------------------------------------------------------
STT_REGISTRY = {
    "openrouter": lambda s: OpenRouterSTTProvider(s.openrouter_api_key, s.openrouter_stt_model),
    "faster-whisper": lambda s: FasterWhisperProvider(),
}

LLM_REGISTRY = {
    "openrouter": lambda s: OpenRouterLLMProvider(s.openrouter_api_key, s.openrouter_llm_model),
    "local-openai-compatible": lambda s: LocalOpenAICompatibleLLM(
        s.local_llm_base_url, s.local_llm_api_key, s.local_llm_model
    ),
}

TTS_REGISTRY = {
    "openrouter": lambda s: OpenRouterTTSProvider(
        s.openrouter_api_key, s.openrouter_tts_model, s.openrouter_tts_voice
    ),
    "chatterbox": lambda s: ChatterboxTTSProvider(),
    "piper": lambda s: PiperTTSProvider(),
}


def _from_registry(registry: dict, name: str, kind: str, cfg: Settings):
    try:
        factory = registry[name]
    except KeyError as exc:
        raise UnknownComponentError(
            f"Unbekannter {kind}-Provider {name!r}. Verfuegbar: {sorted(registry)}"
        ) from exc
    return factory(cfg)


def get_stt_provider(name: str | None = None, cfg: Settings = settings):
    return _from_registry(STT_REGISTRY, name or cfg.default_stt_provider, "STT", cfg)


def get_llm_provider(name: str | None = None, cfg: Settings = settings):
    return _from_registry(LLM_REGISTRY, name or cfg.default_llm_provider, "LLM", cfg)


def get_tts_provider(name: str | None = None, cfg: Settings = settings):
    return _from_registry(TTS_REGISTRY, name or cfg.default_tts_provider, "TTS", cfg)


# ---------------------------------------------------------------------------
# Audio-Router: Modul-Singleton, damit zustandsbehaftete Endpunkte
# (z. B. LoopbackOutput.chunks) ueber Requests hinweg stabil bleiben.
# ---------------------------------------------------------------------------
_audio_router: AudioRouter | None = None


def get_audio_router() -> AudioRouter:
    global _audio_router
    if _audio_router is None:
        _audio_router = AudioRouter(
            inputs=[
                LocalDefaultInput(),
                BluetoothInput(),
                MobileWebSocketInput(),
                MobileWebRTCInput(),
            ],
            outputs=[
                LocalDefaultOutput(),
                BluetoothOutput(),
                MobileWebSocketOutput(),
                MobileWebRTCOutput(),
                LoopbackOutput(),
            ],
        )
    return _audio_router


# ---------------------------------------------------------------------------
# Session-Routing und einheitliche Route-Aufloesung ueber alle Achsen.
# Praezedenz: Settings-Defaults < Session-Route < Request-Overrides.
# ---------------------------------------------------------------------------
ROUTE_KEYS = (
    "input_endpoint",
    "output_endpoint",
    "stt_provider",
    "llm_provider",
    "tts_provider",
    "language",
)


@dataclass
class ResolvedRoute:
    input_endpoint: str
    output_endpoint: str
    stt_provider: str
    llm_provider: str
    tts_provider: str
    language: str

    def as_dict(self) -> dict:
        return {
            "input_endpoint": self.input_endpoint,
            "output_endpoint": self.output_endpoint,
            "stt_provider": self.stt_provider,
            "llm_provider": self.llm_provider,
            "tts_provider": self.tts_provider,
            "language": self.language,
        }


def get_session_route(session_id: str | None) -> dict:
    """Liefert die gespeicherte Route einer Session (leeres dict ohne session_id)."""
    return session_manager.get(session_id) if session_id else {}


def resolve_route(
    session_id: str | None = None,
    overrides: dict | None = None,
    cfg: Settings = settings,
) -> ResolvedRoute:
    """Loest die effektive Route aus Defaults, Session und Request-Overrides auf."""
    resolved = {
        "input_endpoint": cfg.default_input_endpoint,
        "output_endpoint": cfg.default_output_endpoint,
        "stt_provider": cfg.default_stt_provider,
        "llm_provider": cfg.default_llm_provider,
        "tts_provider": cfg.default_tts_provider,
        "language": cfg.default_language,
    }

    session_route = get_session_route(session_id)
    request_overrides = overrides or {}

    for layer in (session_route, request_overrides):
        for key in ROUTE_KEYS:
            value = layer.get(key)
            if value is not None:
                resolved[key] = value

    return ResolvedRoute(**resolved)


def build_orchestrator(route: ResolvedRoute, cfg: Settings = settings) -> Orchestrator:
    return Orchestrator(
        stt=get_stt_provider(route.stt_provider, cfg),
        llm=get_llm_provider(route.llm_provider, cfg),
        tts=get_tts_provider(route.tts_provider, cfg),
        input_cleaner=InputCleaner(),
        spoken_adapter=SpokenResponseAdapter(),
        tts_normalizer=TTSNormalizer(),
    )


async def resolve_output_endpoint(route: ResolvedRoute):
    return await get_audio_router().select_output(route.output_endpoint)


async def resolve_input_endpoint(route: ResolvedRoute):
    return await get_audio_router().select_input(route.input_endpoint)