From 1b3b67779c70fab223d5058109dbe978ae11fdac Mon Sep 17 00:00:00 2001 From: Mike Seese Date: Mon, 5 Jan 2026 10:57:27 -0800 Subject: [PATCH] switch hathora services to use `InputParams` pattern --- src/pipecat/services/hathora/stt.py | 40 ++++++++++++++--------- src/pipecat/services/hathora/tts.py | 50 +++++++++++++++++------------ 2 files changed, 54 insertions(+), 36 deletions(-) diff --git a/src/pipecat/services/hathora/stt.py b/src/pipecat/services/hathora/stt.py index f87f15fdf..a0829b455 100644 --- a/src/pipecat/services/hathora/stt.py +++ b/src/pipecat/services/hathora/stt.py @@ -11,6 +11,7 @@ import os from typing import AsyncGenerator, Optional import aiohttp +from pydantic import BaseModel from pipecat.frames.frames import ( ErrorFrame, @@ -31,14 +32,27 @@ class HathoraSTTService(SegmentedSTTService): [Documentation](https://models.hathora.dev) """ + class InputParams(BaseModel): + """Optional input parameters for Hathora STT configuration. + + Parameters: + language: Language code (if supported by model). + model_config: Some models support additional config, refer to + [docs](https://models.hathora.dev) for each model to see + what is supported. + base_url: Base API URL for the Hathora STT service. + """ + + language: Optional[str] = None + model_config: Optional[list[ConfigOption]] = None + base_url: str = "https://api.models.hathora.dev/inference/v1/stt", + def __init__( self, *, model: str, - language: Optional[str] = None, - model_config: Optional[list[ConfigOption]] = None, - base_url: str = "https://api.models.hathora.dev/inference/v1/stt", api_key: Optional[str] = None, + params: Optional[InputParams] = None, **kwargs, ): """Initialize the Hathora STT service. @@ -46,23 +60,17 @@ class HathoraSTTService(SegmentedSTTService): Args: model: Model to use; find available models [here](https://models.hathora.dev). - language: Language code (if supported by model). - model_config: Some models support additional config, refer to - [docs](https://models.hathora.dev) for each model to see - what is supported. - base_url: Base API URL for the Hathora STT service. api_key: API key for authentication with the Hathora service; provision one [here](https://models.hathora.dev/tokens). + params: Configuration parameters. **kwargs: Additional arguments passed to the parent class. """ super().__init__( **kwargs, ) self._model = model - self._language = language - self._model_config = model_config - self._base_url = base_url self._api_key = api_key or os.getenv("HATHORA_API_KEY") + self._params = params or HathoraSTTService.InputParams() @traced_stt async def _handle_transcription( @@ -84,17 +92,17 @@ class HathoraSTTService(SegmentedSTTService): await self.start_processing_metrics() await self.start_ttfb_metrics() - url = f"{self._base_url}" + url = f"{self._params.base_url}" payload = { "model": self._model, } - if self._language is not None: - payload["language"] = self._language - if self._model_config is not None: + if self._params.language is not None: + payload["language"] = self._params.language + if self._params.model_config is not None: payload["model_config"] = [ - {"name": option.name, "value": option.value} for option in self._model_config + {"name": option.name, "value": option.value} for option in self._params.model_config ] base64_audio = base64.b64encode(audio).decode("utf-8") diff --git a/src/pipecat/services/hathora/tts.py b/src/pipecat/services/hathora/tts.py index 2a82c7dbc..f1d35c9fb 100644 --- a/src/pipecat/services/hathora/tts.py +++ b/src/pipecat/services/hathora/tts.py @@ -12,6 +12,7 @@ import wave from typing import AsyncGenerator, Optional, Tuple import aiohttp +from pydantic import BaseModel from pipecat.frames.frames import ( ErrorFrame, @@ -50,15 +51,28 @@ class HathoraTTSService(TTSService): [Documentation](https://models.hathora.dev) """ + class InputParams(BaseModel): + """Optional input parameters for Hathora TTS configuration. + + Parameters: + speed: Speech speed multiplier (if supported by model). + model_config: Some models support additional config, refer to + [docs](https://models.hathora.dev) for each model to see + what is supported. + base_url: Base API URL for the Hathora TTS service. + """ + + speed: Optional[float] = None + model_config: Optional[list[ConfigOption]] = None + base_url: str = "https://api.models.hathora.dev/inference/v1/tts", + def __init__( self, *, model: str, voice_id: Optional[str] = None, - speed: Optional[float] = None, - model_config: Optional[list[ConfigOption]] = None, - base_url: str = "https://api.models.hathora.dev/inference/v1/tts", api_key: Optional[str] = None, + params: Optional[InputParams] = None, **kwargs, ): """Initialize the Hathora TTS service. @@ -66,26 +80,22 @@ class HathoraTTSService(TTSService): Args: model: Model to use; find available models [here](https://models.hathora.dev). - voice: Voice to use for synthesis (if supported by model). - speed: Speech speed multiplier (if supported by model). - model_config: Some models support additional config, refer to - [docs](https://models.hathora.dev) for each model to see - what is supported. - base_url: Base API URL for the Hathora TTS service. + voice_id: Voice to use for synthesis (if supported by model). api_key: API key for authentication with the Hathora service; provision one [here](https://models.hathora.dev/tokens). + params: Configuration parameters. **kwargs: Additional arguments passed to the parent class. """ super().__init__( **kwargs, - voice_id=voice_id, ) self._model = model - self._speed = speed - self._model_config = model_config - self._base_url = base_url self._api_key = api_key or os.getenv("HATHORA_API_KEY") + self._params = params or HathoraTTSService.InputParams() + self.set_voice(voice_id) + + @traced_tts async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: """Run text-to-speech synthesis on the provided text. @@ -99,17 +109,17 @@ class HathoraTTSService(TTSService): await self.start_processing_metrics() await self.start_ttfb_metrics() - url = f"{self._base_url}" + url = f"{self._params.base_url}" payload = {"model": self._model, "text": text} - if self._voice is not None: - payload["voice"] = self._voice - if self._speed is not None: - payload["speed"] = self._speed - if self._model_config is not None: + if self._voice_id is not None: + payload["voice"] = self._voice_id + if self._params.speed is not None: + payload["speed"] = self._params.speed + if self._params.model_config is not None: payload["model_config"] = [ - {"name": option.name, "value": option.value} for option in self._model_config + {"name": option.name, "value": option.value} for option in self._params.model_config ] yield TTSStartedFrame()