wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

introduce PipelineParams audio input/output sample rates

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2025-02-04 12:22:41 -08:00
parent cc54255c41
commit ab45e481be
61 changed files with 570 additions and 402 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
src/pipecat/audio/vad/silero.py
View File

@@ -5,6 +5,7 @@
#
import time
from typing import Optional
import numpy as np
from loguru import logger
@@ -104,11 +105,8 @@ class SileroOnnxModel:
class SileroVADAnalyzer(VADAnalyzer):
def __init__(self, *, sample_rate: int = 16000, params: VADParams = VADParams()):
super().__init__(sample_rate=sample_rate, num_channels=1, params=params)
if sample_rate != 16000 and sample_rate != 8000:
raise ValueError("Silero VAD sample rate needs to be 16000 or 8000")
def __init__(self, *, sample_rate: Optional[int] = None, params: VADParams = VADParams()):
super().__init__(sample_rate=sample_rate, params=params)
logger.debug("Loading Silero VAD model...")
@@ -138,6 +136,12 @@ class SileroVADAnalyzer(VADAnalyzer):
# VADAnalyzer
#
def set_sample_rate(self, sample_rate: int):
if sample_rate != 16000 and sample_rate != 8000:
raise ValueError("Silero VAD sample rate needs to be 16000 or 8000")
super().set_sample_rate(sample_rate)
def num_frames_required(self) -> int:
return 512 if self.sample_rate == 16000 else 256

19
src/pipecat/audio/vad/vad_analyzer.py
View File

@@ -6,6 +6,7 @@
from abc import abstractmethod
from enum import Enum
from typing import Optional
from loguru import logger
from pydantic import BaseModel
@@ -33,11 +34,11 @@ class VADParams(BaseModel):
class VADAnalyzer:
def __init__(self, *, sample_rate: int, num_channels: int, params: VADParams):
self._sample_rate = sample_rate
self._num_channels = num_channels
self.set_params(params)
def __init__(self, *, sample_rate: Optional[int] = None, params: VADParams):
self._init_sample_rate = sample_rate
self._sample_rate = 0
self._params = params
self._num_channels = 1
self._vad_buffer = b""
@@ -65,13 +66,17 @@ class VADAnalyzer:
def voice_confidence(self, buffer) -> float:
pass
def set_sample_rate(self, sample_rate: int):
self._sample_rate = self._init_sample_rate or sample_rate
self.set_params(self._params)
def set_params(self, params: VADParams):
logger.info(f"Setting VAD params to: {params}")
self._params = params
self._vad_frames = self.num_frames_required()
self._vad_frames_num_bytes = self._vad_frames * self._num_channels * 2
vad_frames_per_sec = self._vad_frames / self._sample_rate
vad_frames_per_sec = self._vad_frames / self.sample_rate
self._vad_start_frames = round(self._params.start_secs / vad_frames_per_sec)
self._vad_stop_frames = round(self._params.stop_secs / vad_frames_per_sec)
@@ -80,7 +85,7 @@ class VADAnalyzer:
self._vad_state: VADState = VADState.QUIET
def _get_smoothed_volume(self, audio: bytes) -> float:
volume = calculate_audio_volume(audio, self._sample_rate)
volume = calculate_audio_volume(audio, self.sample_rate)
return exp_smoothing(volume, self._prev_volume, self._smoothing_factor)
def analyze_audio(self, buffer) -> VADState:

2
src/pipecat/frames/frames.py
View File

@@ -428,6 +428,8 @@ class StartFrame(SystemFrame):
clock: BaseClock
task_manager: TaskManager
audio_in_sample_rate: int = 16000
audio_out_sample_rate: int = 24000
allow_interruptions: bool = False
enable_metrics: bool = False
enable_usage_metrics: bool = False

9
src/pipecat/pipeline/task.py
View File

@@ -40,6 +40,8 @@ HEARTBEAT_MONITOR_SECONDS = HEARTBEAT_SECONDS * 5
class PipelineParams(BaseModel):
model_config = ConfigDict(arbitrary_types_allowed=True)
audio_in_sample_rate: int = 16000
audio_out_sample_rate: int = 24000
allow_interruptions: bool = False
enable_heartbeats: bool = False
enable_metrics: bool = False
@@ -136,6 +138,11 @@ class PipelineTask(BaseTask):
"""Returns the name of this task."""
return self._name
@property
def params(self) -> PipelineParams:
"""Returns the pipeline parameters of this task."""
return self._params
def set_event_loop(self, loop: asyncio.AbstractEventLoop):
self._task_manager.set_event_loop(loop)
@@ -275,6 +282,8 @@ class PipelineTask(BaseTask):
enable_usage_metrics=self._params.enable_usage_metrics,
report_only_initial_ttfb=self._params.report_only_initial_ttfb,
observer=self._observer,
audio_in_sample_rate=self._params.audio_in_sample_rate,
audio_out_sample_rate=self._params.audio_out_sample_rate,
)
await self._source.queue_frame(start_frame, FrameDirection.DOWNSTREAM)

19
src/pipecat/processors/audio/audio_buffer_processor.py
View File

@@ -5,6 +5,7 @@
#
import time
from typing import Optional
from pipecat.audio.utils import create_default_resampler, interleave_stereo_audio, mix_audio
from pipecat.frames.frames import (
@@ -14,6 +15,7 @@ from pipecat.frames.frames import (
Frame,
InputAudioRawFrame,
OutputAudioRawFrame,
StartFrame,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
@@ -33,10 +35,16 @@ class AudioBufferProcessor(FrameProcessor):
"""
def __init__(
self, *, sample_rate: int = 24000, num_channels: int = 1, buffer_size: int = 0, **kwargs
self,
*,
sample_rate: Optional[int] = None,
num_channels: int = 1,
buffer_size: int = 0,
**kwargs,
):
super().__init__(**kwargs)
self._sample_rate = sample_rate
self._init_sample_rate = sample_rate
self._sample_rate = 0
self._num_channels = num_channels
self._buffer_size = buffer_size
@@ -86,6 +94,10 @@ class AudioBufferProcessor(FrameProcessor):
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
# Update output sample rate if necessary.
if isinstance(frame, StartFrame):
self._update_sample_rate(frame)
if self._recording and isinstance(frame, InputAudioRawFrame):
# Add silence if we need to.
silence = self._compute_silence(self._last_user_frame_at)
@@ -113,6 +125,9 @@ class AudioBufferProcessor(FrameProcessor):
await self.push_frame(frame, direction)
def _update_sample_rate(self, frame: StartFrame):
self._sample_rate = self._init_sample_rate or frame.audio_out_sample_rate
async def _call_on_audio_data_handler(self):
if not self.has_audio() or not self._recording:
return

8
src/pipecat/processors/audio/vad/silero.py
View File

@@ -4,6 +4,8 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
from typing import Optional
from loguru import logger
from pipecat.audio.vad.silero import SileroVADAnalyzer
@@ -11,6 +13,7 @@ from pipecat.audio.vad.vad_analyzer import VADParams, VADState
from pipecat.frames.frames import (
AudioRawFrame,
Frame,
StartFrame,
StartInterruptionFrame,
StopInterruptionFrame,
UserStartedSpeakingFrame,
@@ -23,7 +26,7 @@ class SileroVAD(FrameProcessor):
def __init__(
self,
*,
sample_rate: int = 16000,
sample_rate: Optional[int] = None,
vad_params: VADParams = VADParams(),
audio_passthrough: bool = False,
):
@@ -41,6 +44,9 @@ class SileroVAD(FrameProcessor):
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
if isinstance(frame, StartFrame):
self._vad_analyzer.set_sample_rate(frame.audio_in_sample_rate)
if isinstance(frame, AudioRawFrame):
await self._analyze_audio(frame)
if self._audio_passthrough:

9
src/pipecat/processors/gstreamer/pipeline_source.py
View File

@@ -5,6 +5,7 @@
#
import asyncio
from typing import Optional
from loguru import logger
from pydantic import BaseModel
@@ -38,7 +39,7 @@ class GStreamerPipelineSource(FrameProcessor):
class OutputParams(BaseModel):
video_width: int = 1280
video_height: int = 720
audio_sample_rate: int = 24000
audio_sample_rate: Optional[int] = None
audio_channels: int = 1
clock_sync: bool = True
@@ -46,6 +47,7 @@ class GStreamerPipelineSource(FrameProcessor):
super().__init__(**kwargs)
self._out_params = out_params
self._sample_rate = 0
Gst.init()
@@ -90,6 +92,7 @@ class GStreamerPipelineSource(FrameProcessor):
await self.push_frame(frame, direction)
async def _start(self, frame: StartFrame):
self._sample_rate = self._out_params.audio_sample_rate or frame.audio_out_sample_rate
self._player.set_state(Gst.State.PLAYING)
async def _stop(self, frame: EndFrame):
@@ -122,7 +125,7 @@ class GStreamerPipelineSource(FrameProcessor):
audioresample = Gst.ElementFactory.make("audioresample", None)
audiocapsfilter = Gst.ElementFactory.make("capsfilter", None)
audiocaps = Gst.Caps.from_string(
f"audio/x-raw,format=S16LE,rate={self._out_params.audio_sample_rate},channels={self._out_params.audio_channels},layout=interleaved"
f"audio/x-raw,format=S16LE,rate={self._sample_rate},channels={self._out_params.audio_channels},layout=interleaved"
)
audiocapsfilter.set_property("caps", audiocaps)
appsink_audio = Gst.ElementFactory.make("appsink", None)
@@ -188,7 +191,7 @@ class GStreamerPipelineSource(FrameProcessor):
(_, info) = buffer.map(Gst.MapFlags.READ)
frame = OutputAudioRawFrame(
audio=info.data,
sample_rate=self._out_params.audio_sample_rate,
sample_rate=self._sample_rate,
num_channels=self._out_params.audio_channels,
)
asyncio.run_coroutine_threadsafe(self.push_frame(frame), self.get_event_loop())

5
src/pipecat/serializers/base_serializer.py
View File

@@ -7,7 +7,7 @@
from abc import ABC, abstractmethod
from enum import Enum
from pipecat.frames.frames import Frame
from pipecat.frames.frames import Frame, StartFrame
class FrameSerializerType(Enum):
@@ -21,6 +21,9 @@ class FrameSerializer(ABC):
def type(self) -> FrameSerializerType:
pass
async def setup(self, frame: StartFrame):
pass
@abstractmethod
async def serialize(self, frame: Frame) -> str | bytes | None:
pass

24
src/pipecat/serializers/telnyx.py
View File

@@ -6,6 +6,7 @@
import base64
import json
from typing import Optional
from pydantic import BaseModel
@@ -22,6 +23,7 @@ from pipecat.frames.frames import (
InputAudioRawFrame,
InputDTMFFrame,
KeypadEntry,
StartFrame,
StartInterruptionFrame,
)
from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializerType
@@ -29,8 +31,8 @@ from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializer
class TelnyxFrameSerializer(FrameSerializer):
class InputParams(BaseModel):
telnyx_sample_rate: int = 8000
sample_rate: int = 16000
telnyx_sample_rate: Optional[int] = None
sample_rate: Optional[int] = None
inbound_encoding: str = "PCMU"
outbound_encoding: str = "PCMU"
@@ -52,17 +54,21 @@ class TelnyxFrameSerializer(FrameSerializer):
def type(self) -> FrameSerializerType:
return FrameSerializerType.TEXT
async def setup(self, frame: StartFrame):
self._telnyx_sample_rate = self._params.telnyx_sample_rate or frame.audio_in_sample_rate
self._sample_rate = self._params.sample_rate or frame.audio_out_sample_rate
async def serialize(self, frame: Frame) -> str | bytes | None:
if isinstance(frame, AudioRawFrame):
data = frame.audio
if self._params.inbound_encoding == "PCMU":
serialized_data = await pcm_to_ulaw(
data, frame.sample_rate, self._params.telnyx_sample_rate, self._resampler
data, frame.sample_rate, self._telnyx_sample_rate, self._resampler
)
elif self._params.inbound_encoding == "PCMA":
serialized_data = await pcm_to_alaw(
data, frame.sample_rate, self._params.telnyx_sample_rate, self._resampler
data, frame.sample_rate, self._telnyx_sample_rate, self._resampler
)
else:
raise ValueError(f"Unsupported encoding: {self._params.inbound_encoding}")
@@ -89,22 +95,22 @@ class TelnyxFrameSerializer(FrameSerializer):
if self._params.outbound_encoding == "PCMU":
deserialized_data = await ulaw_to_pcm(
payload,
self._params.telnyx_sample_rate,
self._params.sample_rate,
self._telnyx_sample_rate,
self._sample_rate,
self._resampler,
)
elif self._params.outbound_encoding == "PCMA":
deserialized_data = await alaw_to_pcm(
payload,
self._params.telnyx_sample_rate,
self._params.sample_rate,
self._telnyx_sample_rate,
self._sample_rate,
self._resampler,
)
else:
raise ValueError(f"Unsupported encoding: {self._params.outbound_encoding}")
audio_frame = InputAudioRawFrame(
audio=deserialized_data, num_channels=1, sample_rate=self._params.sample_rate
audio=deserialized_data, num_channels=1, sample_rate=self._sample_rate
)
return audio_frame
elif message["event"] == "dtmf":

19
src/pipecat/serializers/twilio.py
View File

@@ -6,6 +6,7 @@
import base64
import json
from typing import Optional
from pydantic import BaseModel
@@ -16,6 +17,7 @@ from pipecat.frames.frames import (
InputAudioRawFrame,
InputDTMFFrame,
KeypadEntry,
StartFrame,
StartInterruptionFrame,
TransportMessageFrame,
TransportMessageUrgentFrame,
@@ -25,19 +27,26 @@ from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializer
class TwilioFrameSerializer(FrameSerializer):
class InputParams(BaseModel):
twilio_sample_rate: int = 8000
sample_rate: int = 16000
twilio_sample_rate: Optional[int] = None
sample_rate: Optional[int] = None
def __init__(self, stream_sid: str, params: InputParams = InputParams()):
self._stream_sid = stream_sid
self._params = params
self._twilio_sample_rate = 0
self._sample_rate = 0
self._resampler = create_default_resampler()
@property
def type(self) -> FrameSerializerType:
return FrameSerializerType.TEXT
async def setup(self, frame: StartFrame):
self._twilio_sample_rate = self._params.twilio_sample_rate or frame.audio_in_sample_rate
self._sample_rate = self._params.sample_rate or frame.audio_out_sample_rate
async def serialize(self, frame: Frame) -> str | bytes | None:
if isinstance(frame, StartInterruptionFrame):
answer = {"event": "clear", "streamSid": self._stream_sid}
@@ -46,7 +55,7 @@ class TwilioFrameSerializer(FrameSerializer):
data = frame.audio
serialized_data = await pcm_to_ulaw(
data, frame.sample_rate, self._params.twilio_sample_rate, self._resampler
data, frame.sample_rate, self._twilio_sample_rate, self._resampler
)
payload = base64.b64encode(serialized_data).decode("utf-8")
answer = {
@@ -67,10 +76,10 @@ class TwilioFrameSerializer(FrameSerializer):
payload = base64.b64decode(payload_base64)
deserialized_data = await ulaw_to_pcm(
payload, self._params.twilio_sample_rate, self._params.sample_rate, self._resampler
payload, self._twilio_sample_rate, self._sample_rate, self._resampler
)
audio_frame = InputAudioRawFrame(
audio=deserialized_data, num_channels=1, sample_rate=self._params.sample_rate
audio=deserialized_data, num_channels=1, sample_rate=self._sample_rate
)
return audio_frame
elif message["event"] == "dtmf":

57
src/pipecat/services/ai_services.py
View File

@@ -213,7 +213,7 @@ class TTSService(AIService):
# if push_silence_after_stop is True, send this amount of audio silence
silence_time_s: float = 2.0,
# TTS output sample rate
sample_rate: int = 24000,
sample_rate: Optional[int] = None,
text_filter: Optional[BaseTextFilter] = None,
**kwargs,
):
@@ -224,7 +224,8 @@ class TTSService(AIService):
self._stop_frame_timeout_s: float = stop_frame_timeout_s
self._push_silence_after_stop: bool = push_silence_after_stop
self._silence_time_s: float = silence_time_s
self._sample_rate: int = sample_rate
self._init_sample_rate = sample_rate
self._sample_rate = 0
self._voice_id: str = ""
self._settings: Dict[str, Any] = {}
self._text_filter: Optional[BaseTextFilter] = text_filter
@@ -248,16 +249,20 @@ class TTSService(AIService):
async def flush_audio(self):
pass
def language_to_service_language(self, language: Language) -> str | None:
return Language(language)
# Converts the text to audio.
@abstractmethod
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
pass
def language_to_service_language(self, language: Language) -> str | None:
return Language(language)
async def update_setting(self, key: str, value: Any):
pass
async def start(self, frame: StartFrame):
await super().start(frame)
self._sample_rate = self._init_sample_rate or frame.audio_out_sample_rate
if self._push_stop_frames:
self._stop_frame_task = self.create_task(self._stop_frame_handler())
@@ -467,9 +472,17 @@ class WordTTSService(TTSService):
class STTService(AIService):
"""STTService is a base class for speech-to-text services."""
def __init__(self, audio_passthrough=False, **kwargs):
def __init__(
self,
audio_passthrough=False,
# STT input sample rate
sample_rate: Optional[int] = None,
**kwargs,
):
super().__init__(**kwargs)
self._audio_passthrough = audio_passthrough
self._init_sample_rate = sample_rate
self._sample_rate = 0
self._settings: Dict[str, Any] = {}
self._muted: bool = False
@@ -478,6 +491,10 @@ class STTService(AIService):
"""Returns whether the STT service is currently muted."""
return self._muted
@property
def sample_rate(self) -> int:
return self._sample_rate
@abstractmethod
async def set_model(self, model: str):
self.set_model_name(model)
@@ -491,6 +508,10 @@ class STTService(AIService):
"""Returns transcript as a string"""
pass
async def start(self, frame: StartFrame):
await super().start(frame)
self._sample_rate = self._init_sample_rate or frame.audio_in_sample_rate
async def _update_settings(self, settings: Mapping[str, Any]):
logger.info(f"Updating STT settings: {self._settings}")
for key, value in settings.items():
@@ -540,17 +561,15 @@ class SegmentedSTTService(STTService):
min_volume: float = 0.6,
max_silence_secs: float = 0.3,
max_buffer_secs: float = 1.5,
sample_rate: int = 24000,
num_channels: int = 1,
sample_rate: Optional[int] = None,
**kwargs,
):
super().__init__(**kwargs)
super().__init__(sample_rate=sample_rate, **kwargs)
self._min_volume = min_volume
self._max_silence_secs = max_silence_secs
self._max_buffer_secs = max_buffer_secs
self._sample_rate = sample_rate
self._num_channels = num_channels
(self._content, self._wave) = self._new_wave()
self._content = None
self._wave = None
self._silence_num_frames = 0
# Volume exponential smoothing
self._smoothing_factor = 0.2
@@ -569,8 +588,8 @@ class SegmentedSTTService(STTService):
# If buffer is not empty and we have enough data or there's been a long
# silence, transcribe the audio gathered so far.
silence_secs = self._silence_num_frames / self._sample_rate
buffer_secs = self._wave.getnframes() / self._sample_rate
silence_secs = self._silence_num_frames / self.sample_rate
buffer_secs = self._wave.getnframes() / self.sample_rate
if self._content.tell() > 0 and (
buffer_secs > self._max_buffer_secs or silence_secs > self._max_silence_secs
):
@@ -580,18 +599,24 @@ class SegmentedSTTService(STTService):
await self.process_generator(self.run_stt(self._content.read()))
(self._content, self._wave) = self._new_wave()
async def start(self, frame: StartFrame):
await super().start(frame)
(self._content, self._wave) = self._new_wave()
async def stop(self, frame: EndFrame):
await super().stop(frame)
self._wave.close()
async def cancel(self, frame: CancelFrame):
await super().cancel(frame)
self._wave.close()
def _new_wave(self):
content = io.BytesIO()
ww = wave.open(content, "wb")
ww.setsampwidth(2)
ww.setnchannels(self._num_channels)
ww.setframerate(self._sample_rate)
ww.setnchannels(1)
ww.setframerate(self.sample_rate)
return (content, ww)
def _get_smoothed_volume(self, frame: AudioRawFrame) -> float:

17
src/pipecat/services/assemblyai.py
View File

@@ -5,7 +5,7 @@
#
import asyncio
from typing import AsyncGenerator
from typing import AsyncGenerator, Optional
from loguru import logger
@@ -38,20 +38,17 @@ class AssemblyAISTTService(STTService):
self,
*,
api_key: str,
sample_rate: int = 16000,
sample_rate: Optional[int] = None,
encoding: AudioEncoding = AudioEncoding("pcm_s16le"),
language=Language.EN, # Only English is supported for Realtime
**kwargs,
):
super().__init__(**kwargs)
super().__init__(sample_rate=sample_rate, **kwargs)
aai.settings.api_key = api_key
self._transcriber: aai.RealtimeTranscriber | None = None
# Store reference to the main event loop for use in callback functions
self._loop = asyncio.get_event_loop()
self._settings = {
"sample_rate": sample_rate,
"encoding": encoding,
"language": language,
}
@@ -121,7 +118,7 @@ class AssemblyAISTTService(STTService):
# Schedule the coroutine to run in the main event loop
# This is necessary because this callback runs in a different thread
asyncio.run_coroutine_threadsafe(self.push_frame(frame), self._loop)
asyncio.run_coroutine_threadsafe(self.push_frame(frame), self.get_event_loop())
def on_error(error: aai.RealtimeError):
"""Callback for handling errors from AssemblyAI.
@@ -131,14 +128,16 @@ class AssemblyAISTTService(STTService):
"""
logger.error(f"{self}: An error occurred: {error}")
# Schedule the coroutine to run in the main event loop
asyncio.run_coroutine_threadsafe(self.push_frame(ErrorFrame(str(error))), self._loop)
asyncio.run_coroutine_threadsafe(
self.push_frame(ErrorFrame(str(error))), self.get_event_loop()
)
def on_close():
"""Callback for when the connection to AssemblyAI is closed."""
logger.info(f"{self}: Disconnected from AssemblyAI")
self._transcriber = aai.RealtimeTranscriber(
sample_rate=self._settings["sample_rate"],
sample_rate=self.sample_rate,
encoding=self._settings["encoding"],
on_data=on_data,
on_error=on_error,

9
src/pipecat/services/aws.py
View File

@@ -124,7 +124,7 @@ class PollyTTSService(TTSService):
aws_session_token: Optional[str] = None,
region: Optional[str] = None,
voice_id: str = "Joanna",
sample_rate: int = 24000,
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
**kwargs,
):
@@ -138,7 +138,6 @@ class PollyTTSService(TTSService):
region_name=region,
)
self._settings = {
"sample_rate": sample_rate,
"engine": params.engine,
"language": self.language_to_service_language(params.language)
if params.language
@@ -226,9 +225,7 @@ class PollyTTSService(TTSService):
yield None
return
audio_data = await self._resampler.resample(
audio_data, 16000, self._settings["sample_rate"]
)
audio_data = await self._resampler.resample(audio_data, 16000, self.sample_rate)
await self.start_tts_usage_metrics(text)
@@ -239,7 +236,7 @@ class PollyTTSService(TTSService):
chunk = audio_data[i : i + chunk_size]
if len(chunk) > 0:
await self.stop_ttfb_metrics()
frame = TTSAudioRawFrame(chunk, self._settings["sample_rate"], 1)
frame = TTSAudioRawFrame(chunk, self.sample_rate, 1)
yield frame
yield TTSStoppedFrame()

41
src/pipecat/services/azure.py
View File

@@ -450,14 +450,13 @@ class AzureBaseTTSService(TTSService):
api_key: str,
region: str,
voice="en-US-SaraNeural",
sample_rate: int = 24000,
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
**kwargs,
):
super().__init__(sample_rate=sample_rate, **kwargs)
self._settings = {
"sample_rate": sample_rate,
"emphasis": params.emphasis,
"language": self.language_to_service_language(params.language)
if params.language
@@ -537,7 +536,7 @@ class AzureTTSService(AzureBaseTTSService):
speech_recognition_language=self._settings["language"],
)
speech_config.set_speech_synthesis_output_format(
sample_rate_to_output_format(self._settings["sample_rate"])
sample_rate_to_output_format(self.sample_rate)
)
speech_config.set_service_property(
"synthesizer.synthesis.connection.synthesisConnectionImpl",
@@ -591,7 +590,7 @@ class AzureTTSService(AzureBaseTTSService):
yield TTSAudioRawFrame(
audio=chunk,
sample_rate=self._settings["sample_rate"],
sample_rate=self.sample_rate,
num_channels=1,
)
@@ -612,7 +611,7 @@ class AzureHttpTTSService(AzureBaseTTSService):
speech_recognition_language=self._settings["language"],
)
speech_config.set_speech_synthesis_output_format(
sample_rate_to_output_format(self._settings["sample_rate"])
sample_rate_to_output_format(self.sample_rate)
)
self._speech_synthesizer = SpeechSynthesizer(speech_config=speech_config, audio_config=None)
@@ -633,7 +632,7 @@ class AzureHttpTTSService(AzureBaseTTSService):
# Azure always sends a 44-byte header. Strip it off.
yield TTSAudioRawFrame(
audio=result.audio_data[44:],
sample_rate=self._settings["sample_rate"],
sample_rate=self.sample_rate,
num_channels=1,
)
yield TTSStoppedFrame()
@@ -650,24 +649,14 @@ class AzureSTTService(STTService):
*,
api_key: str,
region: str,
language=Language.EN_US,
sample_rate=24000,
channels=1,
language: Language = Language.EN_US,
sample_rate: Optional[int] = None,
**kwargs,
):
super().__init__(**kwargs)
super().__init__(sample_rate=sample_rate, **kwargs)
speech_config = SpeechConfig(subscription=api_key, region=region)
speech_config.speech_recognition_language = language
stream_format = AudioStreamFormat(samples_per_second=sample_rate, channels=channels)
self._audio_stream = PushAudioInputStream(stream_format)
audio_config = AudioConfig(stream=self._audio_stream)
self._speech_recognizer = SpeechRecognizer(
speech_config=speech_config, audio_config=audio_config
)
self._speech_recognizer.recognized.connect(self._on_handle_recognized)
self._speech_config = SpeechConfig(subscription=api_key, region=region)
self._speech_config.speech_recognition_language = language
async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
await self.start_processing_metrics()
@@ -677,6 +666,16 @@ class AzureSTTService(STTService):
async def start(self, frame: StartFrame):
await super().start(frame)
stream_format = AudioStreamFormat(samples_per_second=self.sample_rate, channels=1)
self._audio_stream = PushAudioInputStream(stream_format)
audio_config = AudioConfig(stream=self._audio_stream)
self._speech_recognizer = SpeechRecognizer(
speech_config=self._speech_config, audio_config=audio_config
)
self._speech_recognizer.recognized.connect(self._on_handle_recognized)
self._speech_recognizer.start_continuous_recognition_async()
async def stop(self, frame: EndFrame):

19
src/pipecat/services/cartesia.py
View File

@@ -89,7 +89,7 @@ class CartesiaTTSService(WordTTSService, WebsocketService):
cartesia_version: str = "2024-06-10",
url: str = "wss://api.cartesia.ai/tts/websocket",
model: str = "sonic",
sample_rate: int = 24000,
sample_rate: Optional[int] = None,
encoding: str = "pcm_s16le",
container: str = "raw",
params: InputParams = InputParams(),
@@ -121,7 +121,7 @@ class CartesiaTTSService(WordTTSService, WebsocketService):
"output_format": {
"container": container,
"encoding": encoding,
"sample_rate": sample_rate,
"sample_rate": 0,
},
"language": self.language_to_service_language(params.language)
if params.language
@@ -174,6 +174,7 @@ class CartesiaTTSService(WordTTSService, WebsocketService):
async def start(self, frame: StartFrame):
await super().start(frame)
self._settings["output_format"]["sample_rate"] = self.sample_rate
await self._connect()
async def stop(self, frame: EndFrame):
@@ -262,7 +263,7 @@ class CartesiaTTSService(WordTTSService, WebsocketService):
self.start_word_timestamps()
frame = TTSAudioRawFrame(
audio=base64.b64decode(msg["data"]),
sample_rate=self._settings["output_format"]["sample_rate"],
sample_rate=self.sample_rate,
num_channels=1,
)
await self.push_frame(frame)
@@ -328,7 +329,7 @@ class CartesiaHttpTTSService(TTSService):
voice_id: str,
model: str = "sonic",
base_url: str = "https://api.cartesia.ai",
sample_rate: int = 24000,
sample_rate: Optional[int] = None,
encoding: str = "pcm_s16le",
container: str = "raw",
params: InputParams = InputParams(),
@@ -341,7 +342,7 @@ class CartesiaHttpTTSService(TTSService):
"output_format": {
"container": container,
"encoding": encoding,
"sample_rate": sample_rate,
"sample_rate": 0,
},
"language": self.language_to_service_language(params.language)
if params.language
@@ -360,6 +361,10 @@ class CartesiaHttpTTSService(TTSService):
def language_to_service_language(self, language: Language) -> str | None:
return language_to_cartesia_language(language)
async def start(self, frame: StartFrame):
await super().start(frame)
self._settings["output_format"]["sample_rate"] = self.sample_rate
async def stop(self, frame: EndFrame):
await super().stop(frame)
await self._client.close()
@@ -394,9 +399,7 @@ class CartesiaHttpTTSService(TTSService):
)
frame = TTSAudioRawFrame(
audio=output["audio"],
sample_rate=self._settings["output_format"]["sample_rate"],
num_channels=1,
audio=output["audio"], sample_rate=self.sample_rate, num_channels=1
)
yield frame
except Exception as e:

19
src/pipecat/services/deepgram.py
View File

@@ -5,7 +5,7 @@
#
import asyncio
from typing import AsyncGenerator
from typing import AsyncGenerator, Optional
from loguru import logger
@@ -53,14 +53,13 @@ class DeepgramTTSService(TTSService):
*,
api_key: str,
voice: str = "aura-helios-en",
sample_rate: int = 24000,
sample_rate: Optional[int] = None,
encoding: str = "linear16",
**kwargs,
):
super().__init__(sample_rate=sample_rate, **kwargs)
self._settings = {
"sample_rate": sample_rate,
"encoding": encoding,
}
self.set_voice(voice)
@@ -75,7 +74,7 @@ class DeepgramTTSService(TTSService):
options = SpeakOptions(
model=self._voice_id,
encoding=self._settings["encoding"],
sample_rate=self._settings["sample_rate"],
sample_rate=self.sample_rate,
container="none",
)
@@ -103,9 +102,7 @@ class DeepgramTTSService(TTSService):
chunk = audio_buffer.read(chunk_size)
if not chunk:
break
frame = TTSAudioRawFrame(
audio=chunk, sample_rate=self._settings["sample_rate"], num_channels=1
)
frame = TTSAudioRawFrame(audio=chunk, sample_rate=self.sample_rate, num_channels=1)
yield frame
yield TTSStoppedFrame()
@@ -121,15 +118,16 @@ class DeepgramSTTService(STTService):
*,
api_key: str,
url: str = "",
live_options: LiveOptions = None,
sample_rate: Optional[int] = None,
live_options: Optional[LiveOptions] = None,
**kwargs,
):
super().__init__(**kwargs)
super().__init__(sample_rate=sample_rate, **kwargs)
default_options = LiveOptions(
encoding="linear16",
language=Language.EN,
model="nova-2-general",
sample_rate=16000,
channels=1,
interim_results=True,
smart_format=True,
@@ -187,6 +185,7 @@ class DeepgramSTTService(STTService):
async def start(self, frame: StartFrame):
await super().start(frame)
self._settings["sample_rate"] = self.sample_rate
await self._connect()
async def stop(self, frame: EndFrame):

52
src/pipecat/services/elevenlabs.py
View File

@@ -104,17 +104,17 @@ def language_to_elevenlabs_language(language: Language) -> str | None:
return result
def sample_rate_from_output_format(output_format: str) -> int:
match output_format:
case "pcm_16000":
return 16000
case "pcm_22050":
return 22050
case "pcm_24000":
return 24000
case "pcm_44100":
return 44100
return 16000
def output_format_from_sample_rate(sample_rate: int) -> str:
match sample_rate:
case 16000:
return "pcm_16000"
case 22050:
return "pcm_22050"
case 24000:
return "pcm_24000"
case 44100:
return "pcm_44100"
return "pcm_16000"
def calculate_word_times(
@@ -165,7 +165,7 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService):
voice_id: str,
model: str = "eleven_flash_v2_5",
url: str = "wss://api.elevenlabs.io",
output_format: ElevenLabsOutputFormat = "pcm_24000",
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
**kwargs,
):
@@ -189,7 +189,7 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService):
push_text_frames=False,
push_stop_frames=True,
stop_frame_timeout_s=2.0,
sample_rate=sample_rate_from_output_format(output_format),
sample_rate=sample_rate,
**kwargs,
)
WebsocketService.__init__(self)
@@ -197,11 +197,9 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService):
self._api_key = api_key
self._url = url
self._settings = {
"sample_rate": sample_rate_from_output_format(output_format),
"language": self.language_to_service_language(params.language)
if params.language
else None,
"output_format": output_format,
"optimize_streaming_latency": params.optimize_streaming_latency,
"stability": params.stability,
"similarity_boost": params.similarity_boost,
@@ -211,6 +209,7 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService):
}
self.set_model_name(model)
self.set_voice(voice_id)
self._output_format = "" # initialized in start()
self._voice_settings = self._set_voice_settings()
# Indicates if we have sent TTSStartedFrame. It will reset to False when
@@ -254,7 +253,7 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService):
await self._disconnect()
await self._connect()
async def _update_settings(self, settings: Dict[str, Any]):
async def _update_settings(self, settings: Mapping[str, Any]):
prev_voice = self._voice_id
await super()._update_settings(settings)
if not prev_voice == self._voice_id:
@@ -264,6 +263,7 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService):
async def start(self, frame: StartFrame):
await super().start(frame)
self._output_format = output_format_from_sample_rate(self.sample_rate)
await self._connect()
async def stop(self, frame: EndFrame):
@@ -322,7 +322,7 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService):
voice_id = self._voice_id
model = self.model_name
output_format = self._settings["output_format"]
output_format = self._output_format
url = f"{self._url}/v1/text-to-speech/{voice_id}/stream-input?model_id={model}&output_format={output_format}&auto_mode={self._settings['auto_mode']}"
if self._settings["optimize_streaming_latency"]:
@@ -375,7 +375,7 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService):
self.start_word_timestamps()
audio = base64.b64decode(msg["audio"])
frame = TTSAudioRawFrame(audio, self._settings["sample_rate"], 1)
frame = TTSAudioRawFrame(audio, self.sample_rate, 1)
await self.push_frame(frame)
if msg.get("alignment"):
word_times = calculate_word_times(msg["alignment"], self._cumulative_time)
@@ -428,7 +428,7 @@ class ElevenLabsHttpTTSService(TTSService):
aiohttp_session: aiohttp ClientSession
model: Model ID (default: "eleven_flash_v2_5" for low latency)
base_url: API base URL
output_format: Audio output format (PCM)
sample_rate: Output sample rate
params: Additional parameters for voice configuration
"""
@@ -448,24 +448,21 @@ class ElevenLabsHttpTTSService(TTSService):
aiohttp_session: aiohttp.ClientSession,
model: str = "eleven_flash_v2_5",
base_url: str = "https://api.elevenlabs.io",
output_format: ElevenLabsOutputFormat = "pcm_24000",
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
**kwargs,
):
super().__init__(sample_rate=sample_rate_from_output_format(output_format), **kwargs)
super().__init__(sample_rate=sample_rate, **kwargs)
self._api_key = api_key
self._base_url = base_url
self._output_format = output_format
self._params = params
self._session = aiohttp_session
self._settings = {
"sample_rate": sample_rate_from_output_format(output_format),
"language": self.language_to_service_language(params.language)
if params.language
else None,
"output_format": output_format,
"optimize_streaming_latency": params.optimize_streaming_latency,
"stability": params.stability,
"similarity_boost": params.similarity_boost,
@@ -474,6 +471,7 @@ class ElevenLabsHttpTTSService(TTSService):
}
self.set_model_name(model)
self.set_voice(voice_id)
self._output_format = "" # initialized in start()
self._voice_settings = self._set_voice_settings()
def can_generate_metrics(self) -> bool:
@@ -508,6 +506,10 @@ class ElevenLabsHttpTTSService(TTSService):
return voice_settings or None
async def start(self, frame: StartFrame):
await super().start(frame)
self._output_format = output_format_from_sample_rate(self.sample_rate)
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using ElevenLabs streaming API.
@@ -570,7 +572,7 @@ class ElevenLabsHttpTTSService(TTSService):
async for chunk in response.content:
if chunk:
await self.stop_ttfb_metrics()
yield TTSAudioRawFrame(chunk, self._settings["sample_rate"], 1)
yield TTSAudioRawFrame(chunk, self.sample_rate, 1)
yield TTSStoppedFrame()

9
src/pipecat/services/fish.py
View File

@@ -56,7 +56,7 @@ class FishAudioTTSService(TTSService, WebsocketService):
api_key: str,
model: str, # This is the reference_id
output_format: FishAudioOutputFormat = "pcm",
sample_rate: int = 24000,
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
**kwargs,
):
@@ -70,7 +70,7 @@ class FishAudioTTSService(TTSService, WebsocketService):
self._started = False
self._settings = {
"sample_rate": sample_rate,
"sample_rate": 0,
"latency": params.latency,
"format": output_format,
"prosody": {
@@ -92,6 +92,7 @@ class FishAudioTTSService(TTSService, WebsocketService):
async def start(self, frame: StartFrame):
await super().start(frame)
self._settings["sample_rate"] = self.sample_rate
await self._connect()
async def stop(self, frame: EndFrame):
@@ -157,9 +158,7 @@ class FishAudioTTSService(TTSService, WebsocketService):
audio_data = msg.get("audio")
# Only process larger chunks to remove msgpack overhead
if audio_data and len(audio_data) > 1024:
frame = TTSAudioRawFrame(
audio_data, self._settings["sample_rate"], 1
)
frame = TTSAudioRawFrame(audio_data, self.sample_rate, 1)
await self.push_frame(frame)
await self.stop_ttfb_metrics()
continue

2
src/pipecat/services/gemini_multimodal_live/events.py
View File

@@ -48,7 +48,7 @@ class AudioInputMessage(BaseModel):
realtimeInput: RealtimeInput
@classmethod
def from_raw_audio(cls, raw_audio: bytes, sample_rate=16000) -> "AudioInputMessage":
def from_raw_audio(cls, raw_audio: bytes, sample_rate: int) -> "AudioInputMessage":
data = base64.b64encode(raw_audio).decode("utf-8")
return cls(
realtimeInput=RealtimeInput(

8
src/pipecat/services/gemini_multimodal_live/gemini.py
View File

@@ -203,6 +203,8 @@ class GeminiMultimodalLiveLLMService(LLMService):
self._bot_audio_buffer = bytearray()
self._bot_text_buffer = ""
self._sample_rate = 24000
self._settings = {
"frequency_penalty": params.frequency_penalty,
"max_tokens": params.max_tokens,
@@ -521,7 +523,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
if self._audio_input_paused:
return
# Send all audio to Gemini
evt = events.AudioInputMessage.from_raw_audio(frame.audio)
evt = events.AudioInputMessage.from_raw_audio(frame.audio, frame.sample_rate)
await self.send_client_event(evt)
# Manage a buffer of audio to use for transcription
audio = frame.audio
@@ -650,7 +652,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
inline_data = part.inlineData
if not inline_data:
return
if inline_data.mimeType != "audio/pcm;rate=24000":
if inline_data.mimeType != f"audio/pcm;rate={self._sample_rate}":
logger.warning(f"Unrecognized server_content format {inline_data.mimeType}")
return
@@ -665,7 +667,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
self._bot_audio_buffer.extend(audio)
frame = TTSAudioRawFrame(
audio=audio,
sample_rate=24000,
sample_rate=self._sample_rate,
num_channels=1,
)
await self.push_frame(frame)

7
src/pipecat/services/gladia.py
View File

@@ -131,7 +131,6 @@ def language_to_gladia_language(language: Language) -> str | None:
class GladiaSTTService(STTService):
class InputParams(BaseModel):
sample_rate: Optional[int] = 16000
language: Optional[Language] = Language.EN
endpointing: Optional[float] = 0.2
maximum_duration_without_endpointing: Optional[int] = 10
@@ -144,17 +143,18 @@ class GladiaSTTService(STTService):
api_key: str,
url: str = "https://api.gladia.io/v2/live",
confidence: float = 0.5,
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
**kwargs,
):
super().__init__(**kwargs)
super().__init__(sample_rate=sample_rate, **kwargs)
self._api_key = api_key
self._url = url
self._settings = {
"encoding": "wav/pcm",
"bit_depth": 16,
"sample_rate": params.sample_rate,
"sample_rate": 0,
"channels": 1,
"language_config": {
"languages": [self.language_to_service_language(params.language)]
@@ -178,6 +178,7 @@ class GladiaSTTService(STTService):
async def start(self, frame: StartFrame):
await super().start(frame)
self._settings["sample_rate"] = self.sample_rate
response = await self._setup_gladia()
self._websocket = await websockets.connect(response["url"])
self._receive_task = self.create_task(self._receive_task_handler())

7
src/pipecat/services/google/google.py
View File

@@ -883,14 +883,13 @@ class GoogleTTSService(TTSService):
credentials: Optional[str] = None,
credentials_path: Optional[str] = None,
voice_id: str = "en-US-Neural2-A",
sample_rate: int = 24000,
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
**kwargs,
):
super().__init__(sample_rate=sample_rate, **kwargs)
self._settings = {
"sample_rate": sample_rate,
"pitch": params.pitch,
"rate": params.rate,
"volume": params.volume,
@@ -996,7 +995,7 @@ class GoogleTTSService(TTSService):
)
audio_config = texttospeech_v1.AudioConfig(
audio_encoding=texttospeech_v1.AudioEncoding.LINEAR16,
sample_rate_hertz=self._settings["sample_rate"],
sample_rate_hertz=self.sample_rate,
)
request = texttospeech_v1.SynthesizeSpeechRequest(
@@ -1019,7 +1018,7 @@ class GoogleTTSService(TTSService):
if not chunk:
break
await self.stop_ttfb_metrics()
frame = TTSAudioRawFrame(chunk, self._settings["sample_rate"], 1)
frame = TTSAudioRawFrame(chunk, self.sample_rate, 1)
yield frame
await asyncio.sleep(0) # Allow other tasks to run

9
src/pipecat/services/lmnt.py
View File

@@ -5,7 +5,7 @@
#
import json
from typing import AsyncGenerator
from typing import AsyncGenerator, Optional
from loguru import logger
@@ -66,7 +66,7 @@ class LmntTTSService(TTSService, WebsocketService):
*,
api_key: str,
voice_id: str,
sample_rate: int = 24000,
sample_rate: Optional[int] = None,
language: Language = Language.EN,
**kwargs,
):
@@ -81,7 +81,6 @@ class LmntTTSService(TTSService, WebsocketService):
self._api_key = api_key
self._voice_id = voice_id
self._settings = {
"sample_rate": sample_rate,
"language": self.language_to_service_language(language),
"format": "raw", # Use raw format for direct PCM data
}
@@ -132,7 +131,7 @@ class LmntTTSService(TTSService, WebsocketService):
"X-API-Key": self._api_key,
"voice": self._voice_id,
"format": self._settings["format"],
"sample_rate": self._settings["sample_rate"],
"sample_rate": self.sample_rate,
"language": self._settings["language"],
}
@@ -175,7 +174,7 @@ class LmntTTSService(TTSService, WebsocketService):
await self.stop_ttfb_metrics()
frame = TTSAudioRawFrame(
audio=message,
sample_rate=self._settings["sample_rate"],
sample_rate=self.sample_rate,
num_channels=1,
)
await self.push_frame(frame)

9
src/pipecat/services/openai.py
View File

@@ -415,17 +415,14 @@ class OpenAITTSService(TTSService):
def __init__(
self,
*,
api_key: str | None = None,
api_key: Optional[str] = None,
voice: str = "alloy",
model: Literal["tts-1", "tts-1-hd"] = "tts-1",
sample_rate: int = 24000,
sample_rate: Optional[int] = None,
**kwargs,
):
super().__init__(sample_rate=sample_rate, **kwargs)
self._settings = {
"sample_rate": sample_rate,
}
self.set_model_name(model)
self.set_voice(voice)
@@ -465,7 +462,7 @@ class OpenAITTSService(TTSService):
async for chunk in r.iter_bytes(8192):
if len(chunk) > 0:
await self.stop_ttfb_metrics()
frame = TTSAudioRawFrame(chunk, self._settings["sample_rate"], 1)
frame = TTSAudioRawFrame(chunk, self.sample_rate, 1)
yield frame
yield TTSStoppedFrame()
except BadRequestError as e:

14
src/pipecat/services/playht.py
View File

@@ -113,7 +113,7 @@ class PlayHTTTSService(TTSService, WebsocketService):
user_id: str,
voice_url: str,
voice_engine: str = "Play3.0-mini",
sample_rate: int = 24000,
sample_rate: Optional[int] = None,
output_format: str = "wav",
params: InputParams = InputParams(),
**kwargs,
@@ -132,7 +132,6 @@ class PlayHTTTSService(TTSService, WebsocketService):
self._request_id = None
self._settings = {
"sample_rate": sample_rate,
"language": self.language_to_service_language(params.language)
if params.language
else "english",
@@ -250,7 +249,7 @@ class PlayHTTTSService(TTSService, WebsocketService):
if message.startswith(b"RIFF"):
continue
await self.stop_ttfb_metrics()
frame = TTSAudioRawFrame(message, self._settings["sample_rate"], 1)
frame = TTSAudioRawFrame(message, self.sample_rate, 1)
await self.push_frame(frame)
else:
logger.debug(f"Received text message: {message}")
@@ -301,7 +300,7 @@ class PlayHTTTSService(TTSService, WebsocketService):
"voice": self._voice_id,
"voice_engine": self._settings["voice_engine"],
"output_format": self._settings["output_format"],
"sample_rate": self._settings["sample_rate"],
"sample_rate": self.sample_rate,
"language": self._settings["language"],
"speed": self._settings["speed"],
"seed": self._settings["seed"],
@@ -339,7 +338,7 @@ class PlayHTHttpTTSService(TTSService):
user_id: str,
voice_url: str,
voice_engine: str = "Play3.0-mini-http", # Options: Play3.0-mini-http, Play3.0-mini-ws
sample_rate: int = 24000,
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
**kwargs,
):
@@ -353,7 +352,6 @@ class PlayHTHttpTTSService(TTSService):
api_key=self._api_key,
)
self._settings = {
"sample_rate": sample_rate,
"language": self.language_to_service_language(params.language)
if params.language
else "english",
@@ -377,7 +375,7 @@ class PlayHTHttpTTSService(TTSService):
self._options = TTSOptions(
voice=self._voice_id,
language=playht_language,
sample_rate=self._settings["sample_rate"],
sample_rate=self.sample_rate,
format=self._settings["format"],
speed=self._settings["speed"],
seed=self._settings["seed"],
@@ -422,7 +420,7 @@ class PlayHTHttpTTSService(TTSService):
else:
if len(chunk):
await self.stop_ttfb_metrics()
frame = TTSAudioRawFrame(chunk, self._settings["sample_rate"], 1)
frame = TTSAudioRawFrame(chunk, self.sample_rate, 1)
yield frame
yield TTSStoppedFrame()
except Exception as e:

6
src/pipecat/services/rime.py
View File

@@ -34,7 +34,7 @@ class RimeHttpTTSService(TTSService):
api_key: str,
voice_id: str = "eva",
model: str = "mist",
sample_rate: int = 24000,
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
**kwargs,
):
@@ -43,7 +43,6 @@ class RimeHttpTTSService(TTSService):
self._api_key = api_key
self._base_url = "https://users.rime.ai/v1/rime-tts"
self._settings = {
"samplingRate": sample_rate,
"speedAlpha": params.speed_alpha,
"reduceLatency": params.reduce_latency,
"pauseBetweenBrackets": params.pause_between_brackets,
@@ -71,6 +70,7 @@ class RimeHttpTTSService(TTSService):
payload["text"] = text
payload["speaker"] = self._voice_id
payload["modelId"] = self._model_name
payload["samplingRate"] = self.sample_rate
try:
await self.start_ttfb_metrics()
@@ -96,7 +96,7 @@ class RimeHttpTTSService(TTSService):
first_chunk = False
if chunk:
frame = TTSAudioRawFrame(chunk, self._settings["samplingRate"], 1)
frame = TTSAudioRawFrame(chunk, self.sample_rate, 1)
yield frame
yield TTSStoppedFrame()

31
src/pipecat/services/riva.py
View File

@@ -49,7 +49,7 @@ class FastPitchTTSService(TTSService):
api_key: str,
server: str = "grpc.nvcf.nvidia.com:443",
voice_id: str = "English-US.Female-1",
sample_rate: int = 24000,
sample_rate: Optional[int] = None,
function_id: str = "0149dedb-2be8-4195-b9a0-e57e0e14f972",
params: InputParams = InputParams(),
**kwargs,
@@ -57,7 +57,6 @@ class FastPitchTTSService(TTSService):
super().__init__(sample_rate=sample_rate, **kwargs)
self._api_key = api_key
self._voice_id = voice_id
self._sample_rate = sample_rate
self._language_code = params.language
self._quality = params.quality
@@ -87,7 +86,7 @@ class FastPitchTTSService(TTSService):
text,
self._voice_id,
self._language_code,
sample_rate_hz=self._sample_rate,
sample_rate_hz=self.sample_rate,
audio_prompt_file=None,
quality=self._quality,
custom_dictionary={},
@@ -114,7 +113,7 @@ class FastPitchTTSService(TTSService):
await self.stop_ttfb_metrics()
frame = TTSAudioRawFrame(
audio=resp.audio,
sample_rate=self._sample_rate,
sample_rate=self.sample_rate,
num_channels=1,
)
yield frame
@@ -136,10 +135,11 @@ class ParakeetSTTService(STTService):
api_key: str,
server: str = "grpc.nvcf.nvidia.com:443",
function_id: str = "1598d209-5e27-4d3c-8079-4751568b1081",
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
**kwargs,
):
super().__init__(**kwargs)
super().__init__(sample_rate=sample_rate, **kwargs)
self._api_key = api_key
self._profanity_filter = False
self._automatic_punctuation = False
@@ -154,7 +154,6 @@ class ParakeetSTTService(STTService):
self._stop_history_eou = -1
self._stop_threshold_eou = -1.0
self._custom_configuration = ""
self._sample_rate: int = 16000
self.set_model_name("parakeet-ctc-1.1b-asr")
@@ -166,6 +165,14 @@ class ParakeetSTTService(STTService):
self._asr_service = riva.client.ASRService(auth)
self._queue = asyncio.Queue()
def can_generate_metrics(self) -> bool:
return False
async def start(self, frame: StartFrame):
await super().start(frame)
config = riva.client.StreamingRecognitionConfig(
config=riva.client.RecognitionConfig(
encoding=riva.client.AudioEncoding.LINEAR_PCM,
@@ -175,14 +182,16 @@ class ParakeetSTTService(STTService):
profanity_filter=self._profanity_filter,
enable_automatic_punctuation=self._automatic_punctuation,
verbatim_transcripts=not self._no_verbatim_transcripts,
sample_rate_hertz=self._sample_rate,
sample_rate_hertz=self.sample_rate,
audio_channel_count=1,
),
interim_results=True,
)
riva.client.add_word_boosting_to_config(
config, self._boosted_lm_words, self._boosted_lm_score
)
riva.client.add_endpoint_parameters_to_config(
config,
self._start_history,
@@ -193,15 +202,9 @@ class ParakeetSTTService(STTService):
self._stop_threshold_eou,
)
riva.client.add_custom_configuration_to_config(config, self._custom_configuration)
self._config = config
self._queue = asyncio.Queue()
def can_generate_metrics(self) -> bool:
return False
async def start(self, frame: StartFrame):
await super().start(frame)
self._thread_task = self.create_task(self._thread_task_handler())
self._response_task = self.create_task(self._response_task_handler())
self._response_queue = asyncio.Queue()

12
src/pipecat/services/xtts.py
View File

@@ -4,7 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
from typing import Any, AsyncGenerator, Dict
from typing import Any, AsyncGenerator, Dict, Optional
import aiohttp
from loguru import logger
@@ -76,7 +76,7 @@ class XTTSService(TTSService):
base_url: str,
aiohttp_session: aiohttp.ClientSession,
language: Language = Language.EN,
sample_rate: int = 24000,
sample_rate: Optional[int] = None,
**kwargs,
):
super().__init__(sample_rate=sample_rate, **kwargs)
@@ -164,18 +164,18 @@ class XTTSService(TTSService):
# XTTS uses 24000 so we need to resample to our desired rate.
resampled_audio = await self._resampler.resample(
bytes(process_data), 24000, self._sample_rate
bytes(process_data), 24000, self.sample_rate
)
# Create the frame with the resampled audio
frame = TTSAudioRawFrame(resampled_audio, self._sample_rate, 1)
frame = TTSAudioRawFrame(resampled_audio, self.sample_rate, 1)
yield frame
# Process any remaining data in the buffer.
if len(buffer) > 0:
resampled_audio = await self._resampler.resample(
bytes(buffer), 24000, self._sample_rate
bytes(buffer), 24000, self.sample_rate
)
frame = TTSAudioRawFrame(resampled_audio, self._sample_rate, 1)
frame = TTSAudioRawFrame(resampled_audio, self.sample_rate, 1)
yield frame
yield TTSStoppedFrame()

31
src/pipecat/transports/base_input.py
View File

@@ -35,6 +35,9 @@ class BaseInputTransport(FrameProcessor):
self._params = params
# Input sample rate. It will be initialized on StartFrame.
self._sample_rate = 0
# We read audio from a single queue one at a time and we then run VAD in
# a thread. Therefore, only one thread should be necessary.
self._executor = ThreadPoolExecutor(max_workers=1)
@@ -43,10 +46,23 @@ class BaseInputTransport(FrameProcessor):
# if passthrough is enabled.
self._audio_task = None
@property
def sample_rate(self) -> int:
return self._sample_rate
@property
def vad_analyzer(self) -> VADAnalyzer | None:
return self._params.vad_analyzer
async def start(self, frame: StartFrame):
self._sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate
# Configure VAD analyzer.
if self._params.vad_enabled and self._params.vad_analyzer:
self._params.vad_analyzer.set_sample_rate(self._sample_rate)
# Start audio filter.
if self._params.audio_in_filter:
await self._params.audio_in_filter.start(self._params.audio_in_sample_rate)
await self._params.audio_in_filter.start(self._sample_rate)
# Create audio input queue and task if needed.
if self._params.audio_in_enabled or self._params.vad_enabled:
self._audio_in_queue = asyncio.Queue()
@@ -67,9 +83,6 @@ class BaseInputTransport(FrameProcessor):
await self.cancel_task(self._audio_task)
self._audio_task = None
def vad_analyzer(self) -> VADAnalyzer | None:
return self._params.vad_analyzer
async def push_audio_frame(self, frame: InputAudioRawFrame):
if self._params.audio_in_enabled or self._params.vad_enabled:
await self._audio_in_queue.put(frame)
@@ -104,9 +117,8 @@ class BaseInputTransport(FrameProcessor):
await self.push_frame(frame, direction)
await self.stop(frame)
elif isinstance(frame, VADParamsUpdateFrame):
vad_analyzer = self.vad_analyzer()
if vad_analyzer:
vad_analyzer.set_params(frame.params)
if self.vad_analyzer:
self.vad_analyzer.set_params(frame.params)
elif isinstance(frame, FilterUpdateSettingsFrame) and self._params.audio_in_filter:
await self._params.audio_in_filter.process_frame(frame)
# Other frames
@@ -140,11 +152,10 @@ class BaseInputTransport(FrameProcessor):
async def _vad_analyze(self, audio_frame: InputAudioRawFrame) -> VADState:
state = VADState.QUIET
vad_analyzer = self.vad_analyzer()
if vad_analyzer:
if self.vad_analyzer:
logger.trace(f"{self}: analyzing VAD on {audio_frame}")
state = await self.get_event_loop().run_in_executor(
self._executor, vad_analyzer.analyze_audio, audio_frame.audio
self._executor, self.vad_analyzer.analyze_audio, audio_frame.audio
)
logger.trace(f"{self}: done analyzing VAD on {audio_frame}")
return state

26
src/pipecat/transports/base_output.py
View File

@@ -57,12 +57,11 @@ class BaseOutputTransport(FrameProcessor):
# framerate.
self._camera_images = None
# We will write 20ms audio at a time. If we receive long audio frames we
# will chunk them. This will help with interruption handling.
audio_bytes_10ms = (
int(self._params.audio_out_sample_rate / 100) * self._params.audio_out_channels * 2
)
self._audio_chunk_size = audio_bytes_10ms * 2
# Output sample rate. It will be initialized on StartFrame.
self._sample_rate = 0
# Chunk size that will be written. It will be computed on StartFrame
self._audio_chunk_size = 0
self._audio_buffer = bytearray()
self._stopped_event = asyncio.Event()
@@ -70,10 +69,21 @@ class BaseOutputTransport(FrameProcessor):
# Indicates if the bot is currently speaking.
self._bot_speaking = False
@property
def sample_rate(self) -> int:
return self._sample_rate
async def start(self, frame: StartFrame):
self._sample_rate = self._params.audio_out_sample_rate or frame.audio_out_sample_rate
# We will write 20ms audio at a time. If we receive long audio frames we
# will chunk them. This will help with interruption handling.
audio_bytes_10ms = int(self._sample_rate / 100) * self._params.audio_out_channels * 2
self._audio_chunk_size = audio_bytes_10ms * 2
# Start audio mixer.
if self._params.audio_out_mixer:
await self._params.audio_out_mixer.start(self._params.audio_out_sample_rate)
await self._params.audio_out_mixer.start(self._sample_rate)
self._create_camera_task()
self._create_sink_tasks()
@@ -298,7 +308,7 @@ class BaseOutputTransport(FrameProcessor):
# Generate an audio frame with only the mixer's part.
frame = OutputAudioRawFrame(
audio=await self._params.audio_out_mixer.mix(silence),
sample_rate=self._params.audio_out_sample_rate,
sample_rate=self._sample_rate,
num_channels=self._params.audio_out_channels,
)
yield frame

4
src/pipecat/transports/base_transport.py
View File

@@ -31,12 +31,12 @@ class TransportParams(BaseModel):
camera_out_color_format: str = "RGB"
audio_out_enabled: bool = False
audio_out_is_live: bool = False
audio_out_sample_rate: int = 24000
audio_out_sample_rate: Optional[int] = None
audio_out_channels: int = 1
audio_out_bitrate: int = 96000
audio_out_mixer: Optional[BaseAudioMixer] = None
audio_in_enabled: bool = False
audio_in_sample_rate: int = 16000
audio_in_sample_rate: Optional[int] = None
audio_in_channels: int = 1
audio_in_filter: Optional[BaseAudioFilter] = None
vad_enabled: bool = False

70
src/pipecat/transports/local/audio.py
View File

@@ -28,35 +28,40 @@ except ModuleNotFoundError as e:
class LocalAudioInputTransport(BaseInputTransport):
def __init__(self, py_audio: pyaudio.PyAudio, params: TransportParams):
super().__init__(params)
self._py_audio = py_audio
self._in_stream = None
self._sample_rate = 0
sample_rate = self._params.audio_in_sample_rate
num_frames = int(sample_rate / 100) * 2 # 20ms of audio
async def start(self, frame: StartFrame):
await super().start(frame)
self._in_stream = py_audio.open(
format=py_audio.get_format_from_width(2),
channels=params.audio_in_channels,
rate=params.audio_in_sample_rate,
self._sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate
num_frames = int(self._sample_rate / 100) * 2 # 20ms of audio
self._in_stream = self._py_audio.open(
format=self._py_audio.get_format_from_width(2),
channels=self._params.audio_in_channels,
rate=self._sample_rate,
frames_per_buffer=num_frames,
stream_callback=self._audio_in_callback,
input=True,
)
async def start(self, frame: StartFrame):
await super().start(frame)
self._in_stream.start_stream()
async def cleanup(self):
await super().cleanup()
self._in_stream.stop_stream()
# This is not very pretty (taken from PyAudio docs).
while self._in_stream.is_active():
await asyncio.sleep(0.1)
self._in_stream.close()
if self._in_stream:
self._in_stream.stop_stream()
# This is not very pretty (taken from PyAudio docs).
while self._in_stream.is_active():
await asyncio.sleep(0.1)
self._in_stream.close()
self._in_stream = None
def _audio_in_callback(self, in_data, frame_count, time_info, status):
frame = InputAudioRawFrame(
audio=in_data,
sample_rate=self._params.audio_in_sample_rate,
sample_rate=self._sample_rate,
num_channels=self._params.audio_in_channels,
)
@@ -68,32 +73,41 @@ class LocalAudioInputTransport(BaseInputTransport):
class LocalAudioOutputTransport(BaseOutputTransport):
def __init__(self, py_audio: pyaudio.PyAudio, params: TransportParams):
super().__init__(params)
self._py_audio = py_audio
self._out_stream = None
self._sample_rate = 0
# We only write audio frames from a single task, so only one thread
# should be necessary.
self._executor = ThreadPoolExecutor(max_workers=1)
self._out_stream = py_audio.open(
format=py_audio.get_format_from_width(2),
channels=params.audio_out_channels,
rate=params.audio_out_sample_rate,
output=True,
)
async def start(self, frame: StartFrame):
await super().start(frame)
self._sample_rate = self._params.audio_out_sample_rate or frame.audio_out_sample_rate
self._out_stream = self._py_audio.open(
format=self._py_audio.get_format_from_width(2),
channels=self._params.audio_out_channels,
rate=self._sample_rate,
output=True,
)
self._out_stream.start_stream()
async def cleanup(self):
await super().cleanup()
self._out_stream.stop_stream()
# This is not very pretty (taken from PyAudio docs).
while self._out_stream.is_active():
await asyncio.sleep(0.1)
self._out_stream.close()
if self._out_stream:
self._out_stream.stop_stream()
# This is not very pretty (taken from PyAudio docs).
while self._out_stream.is_active():
await asyncio.sleep(0.1)
self._out_stream.close()
async def write_raw_audio_frames(self, frames: bytes):
await self.get_event_loop().run_in_executor(self._executor, self._out_stream.write, frames)
if self._out_stream:
await self.get_event_loop().run_in_executor(
self._executor, self._out_stream.write, frames
)
class LocalAudioTransport(BaseTransport):

69
src/pipecat/transports/local/tk.py
View File

@@ -36,35 +36,39 @@ except ModuleNotFoundError as e:
class TkInputTransport(BaseInputTransport):
def __init__(self, py_audio: pyaudio.PyAudio, params: TransportParams):
super().__init__(params)
self._py_audio = py_audio
self._in_stream = None
self._sample_rate = 0
sample_rate = self._params.audio_in_sample_rate
num_frames = int(sample_rate / 100) * 2 # 20ms of audio
async def start(self, frame: StartFrame):
await super().start(frame)
self._in_stream = py_audio.open(
format=py_audio.get_format_from_width(2),
channels=params.audio_in_channels,
rate=params.audio_in_sample_rate,
self._sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate
num_frames = int(self._sample_rate / 100) * 2 # 20ms of audio
self._in_stream = self._py_audio.open(
format=self._py_audio.get_format_from_width(2),
channels=self._params.audio_in_channels,
rate=self._sample_rate,
frames_per_buffer=num_frames,
stream_callback=self._audio_in_callback,
input=True,
)
async def start(self, frame: StartFrame):
await super().start(frame)
self._in_stream.start_stream()
async def cleanup(self):
await super().cleanup()
self._in_stream.stop_stream()
# This is not very pretty (taken from PyAudio docs).
while self._in_stream.is_active():
await asyncio.sleep(0.1)
self._in_stream.close()
if self._in_stream:
self._in_stream.stop_stream()
# This is not very pretty (taken from PyAudio docs).
while self._in_stream.is_active():
await asyncio.sleep(0.1)
self._in_stream.close()
def _audio_in_callback(self, in_data, frame_count, time_info, status):
frame = InputAudioRawFrame(
audio=in_data,
sample_rate=self._params.audio_in_sample_rate,
sample_rate=self._sample_rate,
num_channels=self._params.audio_in_channels,
)
@@ -76,18 +80,14 @@ class TkInputTransport(BaseInputTransport):
class TkOutputTransport(BaseOutputTransport):
def __init__(self, tk_root: tk.Tk, py_audio: pyaudio.PyAudio, params: TransportParams):
super().__init__(params)
self._py_audio = py_audio
self._out_stream = None
self._sample_rate = 0
# We only write audio frames from a single task, so only one thread
# should be necessary.
self._executor = ThreadPoolExecutor(max_workers=1)
self._out_stream = py_audio.open(
format=py_audio.get_format_from_width(2),
channels=params.audio_out_channels,
rate=params.audio_out_sample_rate,
output=True,
)
# Start with a neutral gray background.
array = np.ones((1024, 1024, 3)) * 128
data = f"P5 {1024} {1024} 255 ".encode() + array.astype(np.uint8).tobytes()
@@ -97,18 +97,31 @@ class TkOutputTransport(BaseOutputTransport):
async def start(self, frame: StartFrame):
await super().start(frame)
self._sample_rate = self._params.audio_out_sample_rate or frame.audio_out_sample_rate
self._out_stream = self._py_audio.open(
format=self._py_audio.get_format_from_width(2),
channels=self._params.audio_out_channels,
rate=self._sample_rate,
output=True,
)
self._out_stream.start_stream()
async def cleanup(self):
await super().cleanup()
self._out_stream.stop_stream()
# This is not very pretty (taken from PyAudio docs).
while self._out_stream.is_active():
await asyncio.sleep(0.1)
self._out_stream.close()
if self._out_stream:
self._out_stream.stop_stream()
# This is not very pretty (taken from PyAudio docs).
while self._out_stream.is_active():
await asyncio.sleep(0.1)
self._out_stream.close()
async def write_raw_audio_frames(self, frames: bytes):
await self.get_event_loop().run_in_executor(self._executor, self._out_stream.write, frames)
if self._out_stream:
await self.get_event_loop().run_in_executor(
self._executor, self._out_stream.write, frames
)
async def write_frame_to_camera(self, frame: OutputImageRawFrame):
self.get_event_loop().call_soon(self._write_frame_to_tk, frame)

15
src/pipecat/transports/network/fastapi_websocket.py
View File

@@ -69,6 +69,7 @@ class FastAPIWebsocketInputTransport(BaseInputTransport):
async def start(self, frame: StartFrame):
await super().start(frame)
await self._params.serializer.setup(frame)
if self._params.session_timeout:
self._monitor_websocket_task = self.create_task(self._monitor_websocket())
await self._callbacks.on_client_connected(self._websocket)
@@ -118,9 +119,19 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
self._websocket = websocket
self._params = params
self._send_interval = (self._audio_chunk_size / self._params.audio_out_sample_rate) / 2
# write_raw_audio_frames() is called quickly, as soon as we get audio
# (e.g. from the TTS), and since this is just a network connection we
# would be sending it to quickly. Instead, we want to block to emulate
# an audio device, this is what the send interval is. It will be
# computed on StartFrame.
self._send_interval = 0
self._next_send_time = 0
async def start(self, frame: StartFrame):
await super().start(frame)
await self._params.serializer.setup(frame)
self._send_interval = (self._audio_chunk_size / self.sample_rate) / 2
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
@@ -136,7 +147,7 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
frame = OutputAudioRawFrame(
audio=frames,
sample_rate=self._params.audio_out_sample_rate,
sample_rate=self.sample_rate,
num_channels=self._params.audio_out_channels,
)

12
src/pipecat/transports/network/websocket_client.py
View File

@@ -126,6 +126,7 @@ class WebsocketClientInputTransport(BaseInputTransport):
async def start(self, frame: StartFrame):
await super().start(frame)
await self._params.serializer.setup(frame)
await self._session.setup(frame)
await self._session.connect()
@@ -154,11 +155,18 @@ class WebsocketClientOutputTransport(BaseOutputTransport):
self._session = session
self._params = params
self._send_interval = (self._audio_chunk_size / self._params.audio_out_sample_rate) / 2
# write_raw_audio_frames() is called quickly, as soon as we get audio
# (e.g. from the TTS), and since this is just a network connection we
# would be sending it to quickly. Instead, we want to block to emulate
# an audio device, this is what the send interval is. It will be
# computed on StartFrame.
self._send_interval = 0
self._next_send_time = 0
async def start(self, frame: StartFrame):
await super().start(frame)
self._send_interval = (self._audio_chunk_size / self.sample_rate) / 2
await self._params.serializer.setup(frame)
await self._session.setup(frame)
await self._session.connect()
@@ -176,7 +184,7 @@ class WebsocketClientOutputTransport(BaseOutputTransport):
async def write_raw_audio_frames(self, frames: bytes):
frame = OutputAudioRawFrame(
audio=frames,
sample_rate=self._params.audio_out_sample_rate,
sample_rate=self.sample_rate,
num_channels=self._params.audio_out_channels,
)

51
src/pipecat/transports/network/websocket_server.py
View File

@@ -24,7 +24,6 @@ from pipecat.frames.frames import (
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.serializers.base_serializer import FrameSerializer
from pipecat.serializers.protobuf import ProtobufFrameSerializer
from pipecat.transports.base_input import BaseInputTransport
from pipecat.transports.base_output import BaseOutputTransport
from pipecat.transports.base_transport import BaseTransport, TransportParams
@@ -39,7 +38,7 @@ except ModuleNotFoundError as e:
class WebsocketServerParams(TransportParams):
add_wav_header: bool = False
serializer: FrameSerializer = ProtobufFrameSerializer()
serializer: FrameSerializer
session_timeout: int | None = None
@@ -67,20 +66,32 @@ class WebsocketServerInputTransport(BaseInputTransport):
self._websocket: websockets.WebSocketServerProtocol | None = None
self._server_task = None
# This task will monitor the websocket connection periodically.
self._monitor_task = None
self._stop_server_event = asyncio.Event()
async def start(self, frame: StartFrame):
await super().start(frame)
await self._params.serializer.setup(frame)
self._server_task = self.create_task(self._server_task_handler())
async def stop(self, frame: EndFrame):
await super().stop(frame)
self._stop_server_event.set()
await self.wait_for_task(self._server_task)
if self._monitor_task:
await self.cancel_task(self._monitor_task)
if self._server_task:
await self.wait_for_task(self._server_task)
async def cancel(self, frame: CancelFrame):
await super().cancel(frame)
await self.cancel_task(self._server_task)
if self._monitor_task:
await self.cancel_task(self._monitor_task)
if self._server_task:
await self.cancel_task(self._server_task)
async def _server_task_handler(self):
logger.info(f"Starting websocket server on {self._host}:{self._port}")
@@ -100,7 +111,9 @@ class WebsocketServerInputTransport(BaseInputTransport):
# Create a task to monitor the websocket connection
if self._params.session_timeout:
self.create_task(self._monitor_websocket(websocket))
self._monitor_task = self.create_task(
self._monitor_websocket(websocket, self._params.session_timeout)
)
# Handle incoming messages
try:
@@ -125,10 +138,13 @@ class WebsocketServerInputTransport(BaseInputTransport):
logger.info(f"Client {websocket.remote_address} disconnected")
async def _monitor_websocket(self, websocket: websockets.WebSocketServerProtocol):
"""Wait for self._params.session_timeout seconds, if the websocket is still open, trigger timeout event."""
async def _monitor_websocket(
self, websocket: websockets.WebSocketServerProtocol, session_timeout: int
):
"""Wait for session_timeout seconds, if the websocket is still open,
trigger timeout event."""
try:
await asyncio.sleep(self._params.session_timeout)
await asyncio.sleep(session_timeout)
if not websocket.closed:
await self._callbacks.on_session_timeout(websocket)
except asyncio.CancelledError:
@@ -144,7 +160,12 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
self._websocket: websockets.WebSocketServerProtocol | None = None
self._send_interval = (self._audio_chunk_size / self._params.audio_out_sample_rate) / 2
# write_raw_audio_frames() is called quickly, as soon as we get audio
# (e.g. from the TTS), and since this is just a network connection we
# would be sending it to quickly. Instead, we want to block to emulate
# an audio device, this is what the send interval is. It will be
# computed on StartFrame.
self._send_interval = 0
self._next_send_time = 0
async def set_client_connection(self, websocket: websockets.WebSocketServerProtocol | None):
@@ -153,6 +174,11 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
logger.warning("Only one client allowed, using new connection")
self._websocket = websocket
async def start(self, frame: StartFrame):
await super().start(frame)
await self._params.serializer.setup(frame)
self._send_interval = (self._audio_chunk_size / self.sample_rate) / 2
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
@@ -168,7 +194,7 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
frame = OutputAudioRawFrame(
audio=frames,
sample_rate=self._params.audio_out_sample_rate,
sample_rate=self.sample_rate,
num_channels=self._params.audio_out_channels,
)
@@ -213,14 +239,13 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
class WebsocketServerTransport(BaseTransport):
def __init__(
self,
params: WebsocketServerParams,
host: str = "localhost",
port: int = 8765,
params: WebsocketServerParams = WebsocketServerParams(),
input_name: str | None = None,
output_name: str | None = None,
loop: asyncio.AbstractEventLoop | None = None,
):
super().__init__(input_name=input_name, output_name=output_name, loop=loop)
super().__init__(input_name=input_name, output_name=output_name)
self._host = host
self._port = port
self._params = params

80
src/pipecat/transports/services/daily.py
View File

@@ -71,11 +71,11 @@ class DailyTransportMessageUrgentFrame(TransportMessageUrgentFrame):
class WebRTCVADAnalyzer(VADAnalyzer):
def __init__(self, *, sample_rate=16000, num_channels=1, params: VADParams = VADParams()):
super().__init__(sample_rate=sample_rate, num_channels=num_channels, params=params)
def __init__(self, *, sample_rate: Optional[int] = None, params: VADParams = VADParams()):
super().__init__(sample_rate=sample_rate, params=params)
self._webrtc_vad = Daily.create_native_vad(
reset_period_ms=VAD_RESET_PERIOD_MS, sample_rate=sample_rate, channels=num_channels
reset_period_ms=VAD_RESET_PERIOD_MS, sample_rate=self.sample_rate, channels=1
)
logger.debug("Loaded native WebRTC VAD")
@@ -222,33 +222,13 @@ class DailyTransportClient(EventHandler):
self._callback_queue = asyncio.Queue()
self._callback_task = None
# Input and ouput sample rates. They will be initialize on setup().
self._in_sample_rate = 0
self._out_sample_rate = 0
self._camera: VirtualCameraDevice | None = None
if self._params.camera_out_enabled:
self._camera = Daily.create_camera_device(
self._camera_name(),
width=self._params.camera_out_width,
height=self._params.camera_out_height,
color_format=self._params.camera_out_color_format,
)
self._mic: VirtualMicrophoneDevice | None = None
if self._params.audio_out_enabled:
self._mic = Daily.create_microphone_device(
self._mic_name(),
sample_rate=self._params.audio_out_sample_rate,
channels=self._params.audio_out_channels,
non_blocking=True,
)
self._speaker: VirtualSpeakerDevice | None = None
if self._params.audio_in_enabled or self._params.vad_enabled:
self._speaker = Daily.create_speaker_device(
self._speaker_name(),
sample_rate=self._params.audio_in_sample_rate,
channels=self._params.audio_in_channels,
non_blocking=True,
)
Daily.select_speaker_device(self._speaker_name())
def _camera_name(self):
return f"camera-{self}"
@@ -281,7 +261,7 @@ class DailyTransportClient(EventHandler):
if not self._speaker:
return None
sample_rate = self._params.audio_in_sample_rate
sample_rate = self._in_sample_rate
num_channels = self._params.audio_in_channels
num_frames = int(sample_rate / 100) * 2 # 20ms of audio
@@ -315,6 +295,34 @@ class DailyTransportClient(EventHandler):
self._camera.write_frame(frame.image)
async def setup(self, frame: StartFrame):
self._in_sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate
self._out_sample_rate = self._params.audio_out_sample_rate or frame.audio_out_sample_rate
if self._params.camera_out_enabled and not self._camera:
self._camera = Daily.create_camera_device(
self._camera_name(),
width=self._params.camera_out_width,
height=self._params.camera_out_height,
color_format=self._params.camera_out_color_format,
)
if self._params.audio_out_enabled and not self._mic:
self._mic = Daily.create_microphone_device(
self._mic_name(),
sample_rate=self._out_sample_rate,
channels=self._params.audio_out_channels,
non_blocking=True,
)
if (self._params.audio_in_enabled or self._params.vad_enabled) and not self._speaker:
self._speaker = Daily.create_speaker_device(
self._speaker_name(),
sample_rate=self._in_sample_rate,
channels=self._params.audio_in_channels,
non_blocking=True,
)
Daily.select_speaker_device(self._speaker_name())
if not self._task_manager:
self._task_manager = frame.task_manager
self._callback_task = self._task_manager.create_task(
@@ -707,6 +715,7 @@ class DailyInputTransport(BaseInputTransport):
super().__init__(params, **kwargs)
self._client = client
self._params = params
self._video_renderers = {}
@@ -715,11 +724,10 @@ class DailyInputTransport(BaseInputTransport):
self._audio_in_task = None
self._vad_analyzer: VADAnalyzer | None = params.vad_analyzer
if params.vad_enabled and not params.vad_analyzer:
self._vad_analyzer = WebRTCVADAnalyzer(
sample_rate=self._params.audio_in_sample_rate,
num_channels=self._params.audio_in_channels,
)
@property
def vad_analyzer(self) -> VADAnalyzer | None:
return self._vad_analyzer
async def start(self, frame: StartFrame):
# Parent start.
@@ -728,6 +736,9 @@ class DailyInputTransport(BaseInputTransport):
await self._client.setup(frame)
# Join the room.
await self._client.join()
# Inialize WebRTC VAD if needed.
if self._params.vad_enabled and not self._params.vad_analyzer:
self._vad_analyzer = WebRTCVADAnalyzer(sample_rate=self.sample_rate)
# Create audio task. It reads audio frames from Daily and push them
# internally for VAD processing.
if self._params.audio_in_enabled or self._params.vad_enabled:
@@ -757,9 +768,6 @@ class DailyInputTransport(BaseInputTransport):
await super().cleanup()
await self._client.cleanup()
def vad_analyzer(self) -> VADAnalyzer | None:
return self._vad_analyzer
#
# FrameProcessor
#

16
src/pipecat/transports/services/livekit.py
View File

@@ -101,6 +101,7 @@ class LiveKitTransportClient:
return self._room
async def setup(self, frame: StartFrame):
self._out_sample_rate = self._params.audio_out_sample_rate or frame.audio_out_sample_rate
if not self._task_manager:
self._task_manager = frame.task_manager
self._room = rtc.Room(loop=self._task_manager.get_event_loop())
@@ -138,7 +139,7 @@ class LiveKitTransportClient:
# Set up audio source and track
self._audio_source = rtc.AudioSource(
self._params.audio_out_sample_rate, self._params.audio_out_channels
self._out_sample_rate, self._params.audio_out_channels
)
self._audio_track = rtc.LocalAudioTrack.create_audio_track(
"pipecat-audio", self._audio_source
@@ -351,6 +352,10 @@ class LiveKitInputTransport(BaseInputTransport):
self._vad_analyzer: VADAnalyzer | None = params.vad_analyzer
self._resampler = create_default_resampler()
@property
def vad_analyzer(self) -> VADAnalyzer | None:
return self._vad_analyzer
async def start(self, frame: StartFrame):
await super().start(frame)
await self._client.setup(frame)
@@ -372,9 +377,6 @@ class LiveKitInputTransport(BaseInputTransport):
if self._audio_in_task and (self._params.audio_in_enabled or self._params.vad_enabled):
await self.cancel_task(self._audio_in_task)
def vad_analyzer(self) -> VADAnalyzer | None:
return self._vad_analyzer
async def push_app_message(self, message: Any, sender: str):
frame = LiveKitTransportMessageUrgentFrame(message=message, participant_id=sender)
await self.push_frame(frame)
@@ -401,12 +403,12 @@ class LiveKitInputTransport(BaseInputTransport):
audio_frame = audio_frame_event.frame
audio_data = await self._resampler.resample(
audio_frame.data.tobytes(), audio_frame.sample_rate, self._params.audio_in_sample_rate
audio_frame.data.tobytes(), audio_frame.sample_rate, self.sample_rate
)
return AudioRawFrame(
audio=audio_data,
sample_rate=self._params.audio_in_sample_rate,
sample_rate=self.sample_rate,
num_channels=audio_frame.num_channels,
)
@@ -448,7 +450,7 @@ class LiveKitOutputTransport(BaseOutputTransport):
return rtc.AudioFrame(
data=pipecat_audio,
sample_rate=self._params.audio_out_sample_rate,
sample_rate=self.sample_rate,
num_channels=self._params.audio_out_channels,
samples_per_channel=samples_per_channel,
)