voiceagent/src/voice_assistant/providers.py

"""
Dynamic provider loading and composition for the Voice Assistant.

Purpose:
    Resolve provider names from the application configuration and
    instantiate implementations that satisfy the corresponding adapter
    contract.

Responsibilities:
    - Resolve provider implementations from ProviderCatalog metadata.
    - Import provider classes dynamically.
    - Validate that the loaded implementation matches the expected adapter.
    - Instantiate STT, Agent, TTS, and VAD providers using provider-specific options.

Non-responsibilities:
    - Loading YAML files.
    - Reading secrets directly.
    - Managing WebSocket connections.
    - Managing turn state.
    - Executing provider functionality.

Architecture:
    AppConfig + ProviderCatalog
              |
              v
        Provider Loader
              |
              v
    STTAdapter / AgentAdapter / TTSAdapter / VADAdapter
              |
              v
       concrete provider
"""

from importlib import import_module
from typing import Any

from voice_assistant.agent.adapter import AgentAdapter
from voice_assistant.config.models import (
    AppConfig,
    ComponentConfig,
    ProviderCatalog,
)
from voice_assistant.stt.adapter import STTAdapter
from voice_assistant.tts.adapter import TTSAdapter
from voice_assistant.vad.adapter import VADAdapter


PROVIDER_CONTRACTS: dict[str, type] = {
    "stt": STTAdapter,
    "agent": AgentAdapter,
    "tts": TTSAdapter,
    "vad": VADAdapter,
}


def load_provider_class(
    catalog: ProviderCatalog,
    component: str,
    provider: str,
) -> type:
    """
    Load and validate a provider class from the provider catalog.

    The catalog stores the import path, while this function ensures that
    the resolved class implements the adapter contract associated with the
    requested component.
    """
    contract = PROVIDER_CONTRACTS.get(component)

    if contract is None:
        raise ValueError(
            f"Unsupported provider component: {component!r}."
        )

    component_providers = catalog.providers.get(component)

    if component_providers is None:
        raise ValueError(
            f"No providers are registered for component {component!r}."
        )

    definition = component_providers.get(provider)

    if definition is None:
        raise ValueError(
            f"Provider {provider!r} is not registered for "
            f"component {component!r}."
        )

    module_name, class_name = definition.implementation.split(
        ":",
        maxsplit=1,
    )

    try:
        module = import_module(module_name)
    except ImportError as exc:
        raise RuntimeError(
            f"Could not import provider module "
            f"{module_name!r} for {component!r}/{provider!r}."
        ) from exc

    try:
        provider_class = getattr(module, class_name)
    except AttributeError as exc:
        raise RuntimeError(
            f"Provider class {class_name!r} was not found in "
            f"module {module_name!r}."
        ) from exc

    if not isinstance(provider_class, type):
        raise TypeError(
            f"Provider {component!r}/{provider!r} does not resolve "
            f"to a class."
        )

    if not issubclass(provider_class, contract):
        raise TypeError(
            f"Provider {component!r}/{provider!r} must implement "
            f"{contract.__name__}."
        )

    return provider_class


def _create_provider(
    catalog: ProviderCatalog,
    component: str,
    config: ComponentConfig,
) -> Any:
    """
    Instantiate the provider selected by a component configuration.

    Provider-specific constructor arguments are kept inside `options`
    so the common configuration model does not need to know the API of
    every vendor.
    """
    provider_class = load_provider_class(
        catalog=catalog,
        component=component,
        provider=config.provider,
    )

    try:
        return provider_class(**config.options)
    except TypeError as exc:
        raise TypeError(
            f"Invalid options for provider "
            f"{component!r}/{config.provider!r}: {exc}"
        ) from exc


def create_stt_adapter(
    config: AppConfig,
    catalog: ProviderCatalog,
) -> STTAdapter:
    """Create the configured STT adapter."""
    return _create_provider(
        catalog=catalog,
        component="stt",
        config=config.stt,
    )


def create_agent_adapter(
    config: AppConfig,
    catalog: ProviderCatalog,
) -> AgentAdapter:
    """Create the configured Agent adapter."""
    return _create_provider(
        catalog=catalog,
        component="agent",
        config=config.agent,
    )


def create_tts_adapter(
    config: AppConfig,
    catalog: ProviderCatalog,
) -> TTSAdapter:
    """Create the configured TTS adapter."""
    return _create_provider(
        catalog=catalog,
        component="tts",
        config=config.tts,
    )




def create_vad_adapter(
    config: AppConfig,
    catalog: ProviderCatalog,
) -> VADAdapter:
    """Create the VAD selected by the active environment configuration."""
    if config.vad is None:
        raise ValueError("VAD provider is not configured for this environment.")

    return _create_provider(
        catalog=catalog,
        component="vad",
        config=config.vad,
    )