src/integrations/hume_api.py

"""
hume_api.py

This file defines the interaction with the Hume TTS API, focusing on converting text to audio.
It includes functionality for input validation, API request handling, and processing API responses.

Key Features:
- Encapsulates all logic related to the Hume TTS API.
- Implements retry logic for handling transient API errors.
- Handles received audio and processes it for playback on the web.
- Provides detailed logging for debugging and error tracking.

Classes:
- HumeException: Custom exception for TTS API-related errors.
- HumeConfig: Immutable configuration for interacting with the TTS API.

Functions:
- text_to_speech_with_hume: Converts text to speech using the Hume TTS API with input validation and retry logic.
"""

# Standard Library Imports
from dataclasses import dataclass
import logging
from typing import Optional
# Third-Party Library Imports
import requests
from tenacity import retry, stop_after_attempt, wait_fixed, before_log, after_log
# Local Application Imports
from src.config import logger
from src.utils import validate_env_var, truncate_text


@dataclass(frozen=True)
class HumeConfig:
    """Immutable configuration for interacting with the TTS API."""
    tts_endpoint_url: str = "https://api.hume.ai/v0/tts"
    api_key: str = validate_env_var("HUME_API_KEY")
    voice: str = "KORA"
    audio_format: str = 'wav'
    headers: dict = None

    def __post_init__(self):
        # Dynamically set headers after initialization
        object.__setattr__(self, "headers", { 
            'X-Hume-Api-Key': f"{self.api_key}",
            'Content-Type': 'application/json',
        })


class HumeException(Exception):
    """Custom exception for errors related to the Hume TTS API."""
    def __init__(self, message: str, original_exception: Optional[Exception] = None):
        super().__init__(message)
        self.original_exception = original_exception


# Initialize the Hume client
hume_config = HumeConfig()


@retry(
    stop=stop_after_attempt(3),
    wait=wait_fixed(2),
    before=before_log(logger, logging.DEBUG),
    after=after_log(logger, logging.DEBUG),
)
def text_to_speech_with_hume(prompt: str, text: str) -> bytes:
    """
    Converts text to speech using the Hume TTS API and processes raw binary audio data.

    Args:
        prompt (str): The original user prompt (for debugging).
        text (str): The generated text to be converted to speech.

    Returns:
        bytes: The raw binary audio data for playback.

    Raises:
        HumeException: If there is an error communicating with the Hume TTS API.
    """
    logger.debug(f"Preparing TTS request for prompt: {truncate_text(prompt)}")
    logger.debug(f"Generated text for TTS: {truncate_text(text)}")

    request_body = {
        "text": text,
        "voice": {"name": hume_config.voice},
        # "voice_description": prompt, # <-- breaking request!?
        # "format": hume_config.audio_format, # <-- breaking request!?
    }

    try:
        response = requests.post(
            url=hume_config.tts_endpoint_url,
            headers=hume_config.headers,
            json=request_body,
        )

        # Log the status and content type for debugging
        logger.debug(f"Hume TTS API Response Status: {response.status_code}")

        if response.status_code != 200:
            logger.error(f"Hume TTS API Error: {response.status_code} - {response.text[:200]}... (truncated)")
            raise HumeException(f"Hume TTS API responded with status {response.status_code}: {response.text}")

        # If Content-Type is audio, return the binary audio data
        if response.headers.get("Content-Type", "").startswith("audio/"):
            audio_data = response.content  # Raw binary audio data
            logger.debug(f"Received binary audio data: {len(audio_data)} bytes")
            return audio_data

        # Unexpected content type
        logger.error(f"Unexpected Content-Type: {response.headers.get('Content-Type', 'Unknown')}")
        raise HumeException(f"Unexpected Content-Type: {response.headers.get('Content-Type', 'Unknown')}")

    except requests.exceptions.RequestException as e:
        logger.exception("Request to Hume TTS API failed.")
        raise HumeException(
            message=f"Failed to communicate with Hume TTS API: {e}",
            original_exception=e,
        )
    except Exception as e:
        logger.exception(f"Unexpected error: {e}")
        raise HumeException(
            message=f"Unexpected error while processing the Hume TTS response: {e}",
            original_exception=e,
        )