Merge pull request #2409 from pipecat-ai/mb/refactor-playht-http
Refactor PlayHTHttpTTSService to use aiohttp
This commit is contained in:
Mark Backman
@@ -14,6 +14,7 @@ import io
|
||||
import json
|
||||
import struct
|
||||
import uuid
|
||||
import warnings
|
||||
from typing import AsyncGenerator, Optional
|
||||
|
||||
import aiohttp
|
||||
@@ -37,14 +38,11 @@ from pipecat.transcriptions.language import Language
|
||||
from pipecat.utils.tracing.service_decorators import traced_tts
|
||||
|
||||
try:
|
||||
from pyht.async_client import AsyncClient
|
||||
from pyht.client import Format, TTSOptions
|
||||
from pyht.client import Language as PlayHTLanguage
|
||||
from websockets.asyncio.client import connect as websocket_connect
|
||||
from websockets.protocol import State
|
||||
except ModuleNotFoundError as e:
|
||||
logger.error(f"Exception: {e}")
|
||||
logger.error("In order to use PlayHT, you need to `pip install pipecat-ai[playht]`.")
|
||||
logger.error("In order to use PlayHTTTSService, you need to `pip install pipecat-ai[playht]`.")
|
||||
raise Exception(f"Missing module: {e}")
|
||||
|
||||
|
||||
@@ -429,7 +427,8 @@ class PlayHTHttpTTSService(TTSService):
|
||||
user_id: str,
|
||||
voice_url: str,
|
||||
voice_engine: str = "Play3.0-mini",
|
||||
protocol: str = "http", # Options: http, ws
|
||||
protocol: Optional[str] = None,
|
||||
output_format: str = "wav",
|
||||
sample_rate: Optional[int] = None,
|
||||
params: Optional[InputParams] = None,
|
||||
**kwargs,
|
||||
@@ -441,40 +440,46 @@ class PlayHTHttpTTSService(TTSService):
|
||||
user_id: PlayHT user ID for authentication.
|
||||
voice_url: URL of the voice to use for synthesis.
|
||||
voice_engine: Voice engine to use. Defaults to "Play3.0-mini".
|
||||
protocol: Protocol to use ("http" or "ws"). Defaults to "http".
|
||||
protocol: Protocol to use ("http" or "ws").
|
||||
|
||||
.. deprecated:: 0.0.80
|
||||
This parameter no longer has any effect and will be removed in a future version.
|
||||
Use PlayHTTTSService for WebSocket or PlayHTHttpTTSService for HTTP.
|
||||
|
||||
output_format: Audio output format. Defaults to "wav".
|
||||
sample_rate: Audio sample rate. If None, uses default.
|
||||
params: Additional input parameters for voice customization.
|
||||
**kwargs: Additional arguments passed to parent TTSService.
|
||||
"""
|
||||
super().__init__(sample_rate=sample_rate, **kwargs)
|
||||
|
||||
# Warn about deprecated protocol parameter if explicitly provided
|
||||
if protocol:
|
||||
warnings.warn(
|
||||
"The 'protocol' parameter is deprecated and will be removed in a future version.",
|
||||
DeprecationWarning,
|
||||
stacklevel=2,
|
||||
)
|
||||
|
||||
params = params or PlayHTHttpTTSService.InputParams()
|
||||
|
||||
self._user_id = user_id
|
||||
self._api_key = api_key
|
||||
|
||||
self._client = AsyncClient(
|
||||
user_id=self._user_id,
|
||||
api_key=self._api_key,
|
||||
)
|
||||
|
||||
# Check if voice_engine contains protocol information (backward compatibility)
|
||||
if "-http" in voice_engine:
|
||||
# Extract the base engine name
|
||||
voice_engine = voice_engine.replace("-http", "")
|
||||
protocol = "http"
|
||||
elif "-ws" in voice_engine:
|
||||
# Extract the base engine name
|
||||
voice_engine = voice_engine.replace("-ws", "")
|
||||
protocol = "ws"
|
||||
|
||||
self._settings = {
|
||||
"language": self.language_to_service_language(params.language)
|
||||
if params.language
|
||||
else "english",
|
||||
"format": Format.FORMAT_WAV,
|
||||
"output_format": output_format,
|
||||
"voice_engine": voice_engine,
|
||||
"protocol": protocol,
|
||||
"speed": params.speed,
|
||||
"seed": params.seed,
|
||||
}
|
||||
@@ -490,26 +495,6 @@ class PlayHTHttpTTSService(TTSService):
|
||||
await super().start(frame)
|
||||
self._settings["sample_rate"] = self.sample_rate
|
||||
|
||||
def _create_options(self) -> TTSOptions:
|
||||
"""Create TTSOptions object from current settings."""
|
||||
language_str = self._settings["language"]
|
||||
playht_language = None
|
||||
if language_str:
|
||||
# Convert string to PlayHT Language enum
|
||||
for lang in PlayHTLanguage:
|
||||
if lang.value == language_str:
|
||||
playht_language = lang
|
||||
break
|
||||
|
||||
return TTSOptions(
|
||||
voice=self._voice_id,
|
||||
language=playht_language,
|
||||
sample_rate=self.sample_rate,
|
||||
format=self._settings["format"],
|
||||
speed=self._settings["speed"],
|
||||
seed=self._settings["seed"],
|
||||
)
|
||||
|
||||
def can_generate_metrics(self) -> bool:
|
||||
"""Check if this service can generate processing metrics.
|
||||
|
||||
@@ -542,41 +527,78 @@ class PlayHTHttpTTSService(TTSService):
|
||||
logger.debug(f"{self}: Generating TTS [{text}]")
|
||||
|
||||
try:
|
||||
options = self._create_options()
|
||||
|
||||
await self.start_ttfb_metrics()
|
||||
|
||||
playht_gen = self._client.tts(
|
||||
text,
|
||||
voice_engine=self._settings["voice_engine"],
|
||||
protocol=self._settings["protocol"],
|
||||
options=options,
|
||||
)
|
||||
# Prepare the request payload
|
||||
payload = {
|
||||
"text": text,
|
||||
"voice": self._voice_id,
|
||||
"voice_engine": self._settings["voice_engine"],
|
||||
"output_format": self._settings["output_format"],
|
||||
"sample_rate": self.sample_rate,
|
||||
"language": self._settings["language"],
|
||||
}
|
||||
|
||||
# Add optional parameters if they exist
|
||||
if self._settings["speed"] is not None:
|
||||
payload["speed"] = self._settings["speed"]
|
||||
if self._settings["seed"] is not None:
|
||||
payload["seed"] = self._settings["seed"]
|
||||
|
||||
headers = {
|
||||
"Authorization": f"Bearer {self._api_key}",
|
||||
"X-User-Id": self._user_id,
|
||||
"Content-Type": "application/json",
|
||||
"Accept": "*/*",
|
||||
}
|
||||
|
||||
await self.start_tts_usage_metrics(text)
|
||||
|
||||
yield TTSStartedFrame()
|
||||
|
||||
b = bytearray()
|
||||
in_header = True
|
||||
async for chunk in playht_gen:
|
||||
# skip the RIFF header.
|
||||
if in_header:
|
||||
b.extend(chunk)
|
||||
if len(b) <= 36:
|
||||
continue
|
||||
else:
|
||||
fh = io.BytesIO(b)
|
||||
fh.seek(36)
|
||||
(data, size) = struct.unpack("<4sI", fh.read(8))
|
||||
while data != b"data":
|
||||
fh.read(size)
|
||||
(data, size) = struct.unpack("<4sI", fh.read(8))
|
||||
in_header = False
|
||||
elif len(chunk) > 0:
|
||||
await self.stop_ttfb_metrics()
|
||||
frame = TTSAudioRawFrame(chunk, self.sample_rate, 1)
|
||||
yield frame
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(
|
||||
"https://api.play.ht/api/v2/tts/stream",
|
||||
headers=headers,
|
||||
json=payload,
|
||||
) as response:
|
||||
if response.status not in (200, 201):
|
||||
error_text = await response.text()
|
||||
raise Exception(f"PlayHT API error {response.status}: {error_text}")
|
||||
|
||||
in_header = True
|
||||
buffer = b""
|
||||
|
||||
CHUNK_SIZE = self.chunk_size
|
||||
|
||||
async for chunk in response.content.iter_chunked(CHUNK_SIZE):
|
||||
if len(chunk) == 0:
|
||||
continue
|
||||
|
||||
# Skip the RIFF header
|
||||
if in_header:
|
||||
buffer += chunk
|
||||
if len(buffer) <= 36:
|
||||
continue
|
||||
else:
|
||||
fh = io.BytesIO(buffer)
|
||||
fh.seek(36)
|
||||
(data, size) = struct.unpack("<4sI", fh.read(8))
|
||||
while data != b"data":
|
||||
fh.read(size)
|
||||
(data, size) = struct.unpack("<4sI", fh.read(8))
|
||||
# Extract audio data after header
|
||||
audio_data = buffer[fh.tell() :]
|
||||
if len(audio_data) > 0:
|
||||
await self.stop_ttfb_metrics()
|
||||
frame = TTSAudioRawFrame(audio_data, self.sample_rate, 1)
|
||||
yield frame
|
||||
in_header = False
|
||||
elif len(chunk) > 0:
|
||||
await self.stop_ttfb_metrics()
|
||||
frame = TTSAudioRawFrame(chunk, self.sample_rate, 1)
|
||||
yield frame
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"{self} error generating TTS: {e}")
|
||||
finally:
|
||||
|
||||
Reference in New Issue
Block a user