PyFlowery/pyflowery/pyflowery.py

import asyncio
from typing import AsyncGenerator, Tuple

from pyflowery.exceptions import ResponseError
from pyflowery.models import FloweryAPIConfig, Language, Voice
from pyflowery.rest_adapter import RestAdapter


class FloweryAPI:
    """Main class for interacting with the Flowery API

    Attributes:
        config (FloweryAPIConfig): Configuration object for the API
        adapter (RestAdapter): Adapter for making HTTP requests
    """

    def __init__(self, config: FloweryAPIConfig) -> None:
        self.config = config
        self.adapter = RestAdapter(config=config)
        self._voices_cache: Tuple[Voice, ...] = ()
        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            loop = None
        if loop and loop.is_running():
            self.config.logger.info("Async event loop is already running. Adding `_populate_voices_cache()` to the event loop.")
            asyncio.create_task(self._populate_voices_cache())
        else:
            asyncio.run(self._populate_voices_cache())

    async def _populate_voices_cache(self) -> None:
        """Populate the voices cache. This method is called automatically when the FloweryAPI object is created, and should not be called directly."""
        self._voices_cache = tuple([voice async for voice in self.fetch_voices()])  # pylint: disable=consider-using-generator
        if not self._voices_cache:
            raise ValueError("Failed to populate voices cache! Please report this issue at https://www.coastalcommits.com/cswimr/PyFlowery/issues.")
        self.config.logger.info("Voices cache populated!")

    def get_voices(self, voice_id: str | None = None, name: str | None = None) -> Tuple[Voice, ...] | None:
        """Get a set of voices from the cache.

        Args:
            voice_id (str): The ID of the voice
            name (str): The name of the voice

        Returns:
            Tuple[Voice] | None: A tuple of Voice objects if found, otherwise None
        """
        if voice_id:
            voice = next((voice for voice in self._voices_cache if voice.id == voice_id), None)
            return (voice,) if voice else None
        if name:
            voices = []
            for voice in self._voices_cache:
                if voice.name == name:
                    voices.append(voice)
            return tuple(voices) or None
        return self._voices_cache or None

    async def fetch_voice(self, voice_id: str) -> Voice:
        """Fetch a voice from the Flowery API. This method bypasses the cache and directly queries the Flowery API. You should usually use `get_voices()` instead.

        Args:
            voice_id (str): The ID of the voice

        Raises:
            ValueError: Raised when the voice is not found
            TooManyRequests: Raised when the Flowery API returns a 429 status code
            ClientError: Raised when the Flowery API returns a 4xx status code
            InternalServerError: Raised when the Flowery API returns a 5xx status code
            RetryLimitExceeded: Raised when the retry limit defined in the `FloweryAPIConfig` class (default 3) is exceeded

        Returns:
            Voice: The voice
        """
        async for voice in self.fetch_voices():
            if voice.id == voice_id:
                return voice
        raise ValueError(f"Voice with ID {voice_id} not found.")

    async def fetch_voices(self) -> AsyncGenerator[Voice, None]:
        """Fetch a list of voices from the Flowery API

        Raises:
            TooManyRequests: Raised when the Flowery API returns a 429 status code
            ClientError: Raised when the Flowery API returns a 4xx status code
            InternalServerError: Raised when the Flowery API returns a 5xx status code
            ResponseError: Raised when the Flowery API returns an empty response or a response with an unexpected format
            RetryLimitExceeded: Raised when the retry limit defined in the `FloweryAPIConfig` class (default 3) is exceeded

        Returns:
            AsyncGenerator[Voice, None]: A generator of Voices
        """
        request = await self.adapter.get(endpoint="/tts/voices")
        if request is not None:
            if isinstance(request.data, dict):
                for voice in request.data.get("voices", []):
                    yield Voice(
                        id=voice["id"],
                        name=voice["name"],
                        gender=voice["gender"],
                        source=voice["source"],
                        language=Language(**voice["language"]),
                    )
            else:
                raise ResponseError(f"Invalid response from Flowery API: {request.data!r}")
        else:
            raise ResponseError("Invalid response from Flowery API: Empty Response!}")

    async def fetch_tts(
        self, text: str, voice: Voice | str | None = None, translate: bool = False, silence: int = 0, audio_format: str = "mp3", speed: float = 1.0
    ) -> bytes:
        """Fetch a TTS audio file from the Flowery API

        Args:
            text (str): The text to convert to speech
            voice (Voice | str): The voice to use for the speech
            translate (bool): Whether to translate the text
            silence (int): Number of seconds of silence to add to the end of the audio
            audio_format (str): The audio format to return
            speed (float): The speed of the speech

        Raises:
            ValueError: Raised when the provided text is too long and `allow_truncation` in the `FloweryAPIConfig` class is set to `False` (default).
            TooManyRequests: Raised when the Flowery API returns a 429 status code
            ClientError: Raised when the Flowery API returns a 4xx status code
            InternalServerError: Raised when the Flowery API returns a 5xx status code
            ResponseError: Raised when the Flowery API returns an empty response or a response with an unexpected format
            RetryLimitExceeded: Raised when the retry limit defined in the `FloweryAPIConfig` class (default 3) is exceeded

        Returns:
            bytes: The audio file in bytes
        """
        if len(text) > 2048:
            if not self.config.allow_truncation:
                raise ValueError("Text must be less than or equal to 2048 characters")
            self.config.logger.warning("Text is too long, will be truncated to 2048 characters by the API")
        params = {
            "text": text,
            "translate": str(translate).lower(),
            "silence": silence,
            "audio_format": audio_format,
            "speed": speed,
        }
        if voice:
            params["voice"] = voice.id if isinstance(voice, Voice) else voice

        request = await self.adapter.get(endpoint="/tts", params=params, timeout=180)

        if request is not None:
            if isinstance(request.data, bytes):
                return request.data
            raise ResponseError(f"Invalid response from Flowery API: {request.data!r}")
        raise ResponseError("Invalid response from Flowery API: Empty Response!}")