my_voice_assistant_v3_jamulix/app/dependencies.py

from dataclasses import dataclass

from app.config import Settings, settings
from app.runtime_config import runtime_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.providers.fallback import (
    FallbackSTTProvider,
    FallbackLLMProvider,
    FallbackTTSProvider,
)
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.store import SQLiteStore, Store, User

# ---------------------------------------------------------------------------
# Persistenz-Store: Modul-Singleton (SQLite). Spaetere Backends implementieren
# dasselbe Store-Interface, ohne die App zu aendern.
# ---------------------------------------------------------------------------
_store: Store | None = None


def get_store() -> Store:
    global _store
    if _store is None:
        _store = SQLiteStore(settings.db_path)
    return _store

# ---------------------------------------------------------------------------
# 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,
        system_prompt=s.local_llm_system_prompt,
        disable_reasoning=s.local_llm_disable_reasoning,
        max_tokens=s.local_llm_max_tokens,
        temperature=s.local_llm_temperature,
        top_p=s.local_llm_top_p,
    ),
}

TTS_REGISTRY = {
    "openrouter": lambda s: OpenRouterTTSProvider(
        s.openrouter_api_key, s.openrouter_tts_model, s.openrouter_tts_voice
    ),
    "chatterbox": lambda s: ChatterboxTTSProvider(
        s.chatterbox_base_url,
        s.chatterbox_voice,
        s.chatterbox_lang,
        s.chatterbox_speed,
        s.tts_sample_rate,
        s.chatterbox_timeout,
        voices_dir=s.chatterbox_voices_dir,
    ),
    "piper": lambda s: PiperTTSProvider(
        s.piper_bin, s.piper_voices_dir, s.piper_voice, s.tts_sample_rate
    ),
}


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=None):
    cfg = cfg or runtime_settings
    return _from_registry(STT_REGISTRY, name or cfg.default_stt_provider, "STT", cfg)


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


def get_tts_provider(name: str | None = None, cfg=None):
    cfg = cfg or runtime_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",
    "language_mode",
)

# Stimm-Auswahl nach Sprache: (effektive) Sprache → beste Piper-Stimme.
# Gilt in BEIDEN Modi — die gesprochene Stimme folgt immer der Antwortsprache.
LANG_TO_PIPER_VOICE: dict[str, str] = {
    "de":  "de_DE-thorsten-high",
    "en":  "en_US-lessac-high",
    "fr":  "fr_FR-siwis-medium",
    "es":  "es_ES-sharvard-medium",
    "it":  "it_IT-paola-medium",
    "nl":  "nl_NL-mls-medium",
    "ru":  "ru_RU-irina-medium",
    "zh":  "zh_CN-huayan-medium",
    "cmn": "zh_CN-huayan-medium",
}


def piper_voice_for_language(tts_provider: str, language: str | None) -> str | None:
    """Liefert die zur Sprache passende Piper-Stimme (sonst None → Provider-Default).

    Nur für Piper sinnvoll: Cloud-TTS (OpenRouter) und Chatterbox haben eigene
    Stimmnamen (z. B. "Zephyr") und steuern die Sprache anders. Bei None nimmt der
    jeweilige Provider seine konfigurierte Default-Stimme.
    """
    if tts_provider == "piper" and language:
        return LANG_TO_PIPER_VOICE.get(language)
    return None


@dataclass
class ResolvedRoute:
    input_endpoint: str
    output_endpoint: str
    stt_provider: str
    llm_provider: str
    tts_provider: str
    language: str
    language_mode: str = "fix"

    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,
            "language_mode": self.language_mode,
        }


def get_session_route(session_id: str | None, user: User | None = None) -> dict:
    """Liefert die gespeicherte Route einer Session des Nutzers (leeres dict sonst).

    Gehoert die Session einem anderen Nutzer, wird SessionOwnershipError ausgeloest.
    """
    if not session_id:
        return {}
    session = get_store().get_session(session_id)
    if session is None:
        return {}
    if user is not None and session.user_id != user.id:
        from app.store import SessionOwnershipError

        raise SessionOwnershipError(
            f"Session {session_id!r} gehoert einem anderen Nutzer"
        )
    return session.data


def resolve_route(
    user: User | None = None,
    session_id: str | None = None,
    overrides: dict | None = None,
    cfg=None,
) -> ResolvedRoute:
    cfg = cfg or runtime_settings
    """Loest die effektive Route auf.

    Praezedenz (hoeher gewinnt): Defaults < Nutzer-Prefs < Session-Route < Request.
    """
    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,
        "language_mode": cfg.default_language_mode,
    }

    user_prefs = user.prefs if user is not None else {}
    session_route = get_session_route(session_id, user)
    request_overrides = overrides or {}

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

    return ResolvedRoute(**resolved)


_FALLBACK_CLASS = {
    "stt": FallbackSTTProvider,
    "llm": FallbackLLMProvider,
    "tts": FallbackTTSProvider,
}


def _provider_chain(registry, primary: str, fallback_csv: str, module: str, cfg: Settings):
    """Baut primaeren Provider + optionale Fallback-Kette (dedupliziert, Reihenfolge erhalten)."""
    names = [primary] + [n.strip() for n in (fallback_csv or "").split(",") if n.strip()]
    seen, ordered = set(), []
    for name in names:
        if name not in seen:
            seen.add(name)
            ordered.append(name)
    entries = [(name, _from_registry(registry, name, module.upper(), cfg)) for name in ordered]
    if len(entries) == 1:
        return entries[0][1]
    return _FALLBACK_CLASS[module](module, entries)


def _resolve_normalize_level(tts_provider: str, cfg: Settings) -> str:
    """auto -> piper bekommt 'full', Cloud-TTS 'light' (macht Zahlen/Abk. selbst gut)."""
    level = (cfg.tts_normalize_level or "auto").lower()
    if level == "auto":
        return "full" if tts_provider == "piper" else "light"
    return level


def build_orchestrator(route: ResolvedRoute, cfg=None) -> Orchestrator:
    cfg = cfg or runtime_settings
    return Orchestrator(
        stt=_provider_chain(STT_REGISTRY, route.stt_provider, cfg.stt_fallback, "stt", cfg),
        llm=_provider_chain(LLM_REGISTRY, route.llm_provider, cfg.llm_fallback, "llm", cfg),
        tts=_provider_chain(TTS_REGISTRY, route.tts_provider, cfg.tts_fallback, "tts", cfg),
        input_cleaner=InputCleaner(),
        spoken_adapter=SpokenResponseAdapter(),
        tts_normalizer=TTSNormalizer(),
        normalize_level=_resolve_normalize_level(route.tts_provider, cfg),
    )


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)