diff --git a/CHANGELOG.md b/CHANGELOG.md index 5542674ec..a8731571f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,7 +5,7 @@ All notable changes to **Pipecat** will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [0.0.57] - 2025-02-14 ### Added @@ -56,6 +56,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- We don't consider a colon `:` and end of sentence any more. + - Updated `DailyTransport` to respect the `audio_in_stream_on_start` field, ensuring it only starts receiving the audio input if it is enabled. @@ -105,6 +107,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fixed a `FalImageGenService` issue that was causing the event loop to be + blocked while loading the downloadded image. + - Fixed a `CartesiaTTSService` service issue that would cause audio overlapping in some cases. @@ -134,9 +139,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed an issue[#1192] in 11labs where we are trying to reconnect/disconnect the websocket connection even when the connection is already closed. -- Fixed an issue where `has_regular_messages` condition was always been true in - `GoogleLLMContext` due to `Part` having `function_call` & `function_response` with - `None` values. +- Fixed an issue where `has_regular_messages` condition was always true in + `GoogleLLMContext` due to `Part` having `function_call` & `function_response` + with `None` values. + +### Other + +- Added new `instant-voice` example. This example showcases how to enable + instant voice communication as soon as a user connects. + +- Added new `local-input-select-stt` example. This examples allows you to play + with local audio inputs by slecting them through a nice text interface. ## [0.0.56] - 2025-02-06 diff --git a/examples/foundational/01a-local-audio.py b/examples/foundational/01a-local-audio.py index 633697684..20aaedfae 100644 --- a/examples/foundational/01a-local-audio.py +++ b/examples/foundational/01a-local-audio.py @@ -16,7 +16,7 @@ from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineTask from pipecat.services.cartesia import CartesiaTTSService -from pipecat.transports.local.audio import LocalAudioTransport, LocalTransportParams +from pipecat.transports.local.audio import LocalAudioTransport, LocalAudioTransportParams load_dotenv(override=True) @@ -25,7 +25,7 @@ logger.add(sys.stderr, level="DEBUG") async def main(): - transport = LocalAudioTransport(LocalTransportParams(audio_out_enabled=True)) + transport = LocalAudioTransport(LocalAudioTransportParams(audio_out_enabled=True)) tts = CartesiaTTSService( api_key=os.getenv("CARTESIA_API_KEY"), diff --git a/examples/foundational/06a-image-sync.py b/examples/foundational/06a-image-sync.py index 841c335f6..03a7fd3b8 100644 --- a/examples/foundational/06a-image-sync.py +++ b/examples/foundational/06a-image-sync.py @@ -27,7 +27,7 @@ from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.processors.frame_processor import FrameDirection, FrameProcessor -from pipecat.services.cartesia import CartesiaHttpTTSService +from pipecat.services.cartesia import CartesiaTTSService from pipecat.services.openai import OpenAILLMService from pipecat.transports.services.daily import DailyParams, DailyTransport @@ -91,7 +91,7 @@ async def main(): ), ) - tts = CartesiaHttpTTSService( + tts = CartesiaTTSService( api_key=os.getenv("CARTESIA_API_KEY"), voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady ) diff --git a/examples/foundational/13a-whisper-local.py b/examples/foundational/13a-whisper-local.py index 2d3cd5a57..c9bf16782 100644 --- a/examples/foundational/13a-whisper-local.py +++ b/examples/foundational/13a-whisper-local.py @@ -16,7 +16,7 @@ from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineTask from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.services.whisper import WhisperSTTService -from pipecat.transports.local.audio import LocalAudioTransport, LocalTransportParams +from pipecat.transports.local.audio import LocalAudioTransport, LocalAudioTransportParams load_dotenv(override=True) @@ -33,7 +33,7 @@ class TranscriptionLogger(FrameProcessor): async def main(): - transport = LocalAudioTransport(LocalTransportParams(audio_in_enabled=True)) + transport = LocalAudioTransport(LocalAudioTransportParams(audio_in_enabled=True)) stt = WhisperSTTService() diff --git a/examples/local-input-select-stt/bot.py b/examples/local-input-select-stt/bot.py index 1cedbf96c..f472152b9 100644 --- a/examples/local-input-select-stt/bot.py +++ b/examples/local-input-select-stt/bot.py @@ -18,7 +18,7 @@ from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineTask from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.services.whisper import Model, WhisperSTTService -from pipecat.transports.local.audio import LocalAudioTransport, LocalTransportParams +from pipecat.transports.local.audio import LocalAudioTransport, LocalAudioTransportParams load_dotenv(override=True) @@ -36,7 +36,7 @@ class TranscriptionLogger(FrameProcessor): async def main(input_device: int, output_device: int): transport = LocalAudioTransport( - LocalTransportParams( + LocalAudioTransportParams( audio_in_enabled=True, audio_out_enabled=False, input_device_index=input_device, diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 950c155e6..2de3ebae4 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -10,7 +10,6 @@ from abc import abstractmethod from typing import List from pipecat.frames.frames import ( - BotInterruptionFrame, CancelFrame, EmulateUserStartedSpeakingFrame, EmulateUserStoppedSpeakingFrame, @@ -281,6 +280,7 @@ class LLMUserContextAggregator(LLMContextResponseAggregator): await self._cancel_aggregation_task() async def _handle_user_started_speaking(self, _: UserStartedSpeakingFrame): + self._last_user_speaking_time = time.time() self._user_speaking = True async def _handle_user_stopped_speaking(self, _: UserStoppedSpeakingFrame): @@ -358,6 +358,8 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): super().__init__(context=context, role="assistant", **kwargs) self._expect_stripped_words = expect_stripped_words + self._started = False + self.reset() async def process_frame(self, frame: Frame, direction: FrameDirection): @@ -420,7 +422,7 @@ class LLMUserResponseAggregator(LLMUserContextAggregator): class LLMAssistantResponseAggregator(LLMAssistantContextAggregator): - def __init__(self, messages: List[dict], **kwargs): + def __init__(self, messages: List[dict] = [], **kwargs): super().__init__(context=OpenAILLMContext(messages), **kwargs) async def push_aggregation(self): diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 3fb515a87..f27c6d6c2 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -73,10 +73,11 @@ class FrameProcessor: self._metrics.set_processor_name(self.name) # Processors have an input queue. The input queue will be processed - # immediately (default) or it will block if `pause_processing_frames()` is - # called. To resume processing frames we need to call - # `resume_processing_frames()`. + # immediately (default) or it will block if `pause_processing_frames()` + # is called. To resume processing frames we need to call + # `resume_processing_frames()` which will wake up the event. self.__should_block_frames = False + self.__input_event = asyncio.Event() self.__input_frame_task: Optional[asyncio.Task] = None # Every processor in Pipecat should only output frames from a single @@ -335,8 +336,8 @@ class FrameProcessor: def __create_input_task(self): if not self.__input_frame_task: self.__should_block_frames = False + self.__input_event.clear() self.__input_queue = asyncio.Queue() - self.__input_event = asyncio.Event() self.__input_frame_task = self.create_task(self.__input_frame_task_handler()) async def __cancel_input_task(self): diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 0fce2482a..c5adc89e0 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -15,6 +15,7 @@ from loguru import logger from pipecat.audio.utils import calculate_audio_volume, exp_smoothing from pipecat.frames.frames import ( AudioRawFrame, + BotStoppedSpeakingFrame, CancelFrame, EndFrame, ErrorFrame, @@ -75,13 +76,13 @@ class AIService(FrameProcessor): ) for key, value in settings.items(): - print("Update request for:", key, value) + logger.debug("Update request for:", key, value) if key in self._settings: logger.info(f"Updating LLM setting {key} to: [{value}]") self._settings[key] = value elif key in SessionProperties.model_fields: - print("Attempting to update", key, value) + logger.debug("Attempting to update", key, value) try: from pipecat.services.openai_realtime_beta.events import ( @@ -212,6 +213,8 @@ class TTSService(AIService): push_silence_after_stop: bool = False, # if push_silence_after_stop is True, send this amount of audio silence silence_time_s: float = 2.0, + # if True, we will pause processing frames while we are receiving audio + pause_frame_processing: bool = False, # TTS output sample rate sample_rate: Optional[int] = None, text_filter: Optional[BaseTextFilter] = None, @@ -224,6 +227,7 @@ 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._pause_frame_processing: bool = pause_frame_processing self._init_sample_rate = sample_rate self._sample_rate = 0 self._voice_id: str = "" @@ -234,6 +238,7 @@ class TTSService(AIService): self._stop_frame_queue: asyncio.Queue = asyncio.Queue() self._current_sentence: str = "" + self._processing_text: bool = False @property def sample_rate(self) -> int: @@ -299,6 +304,7 @@ class TTSService(AIService): async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) + if ( isinstance(frame, TextFrame) and not isinstance(frame, InterimTranscriptionFrame) @@ -307,9 +313,16 @@ class TTSService(AIService): await self._process_text_frame(frame) elif isinstance(frame, StartInterruptionFrame): await self._handle_interruption(frame, direction) + await self.push_frame(frame, direction) elif isinstance(frame, (LLMFullResponseEndFrame, EndFrame)): + # We pause processing incoming frames if the LLM response included + # text (it might be that it's only a function calling response). We + # pause to avoid audio overlapping. + await self._maybe_pause_frame_processing() + sentence = self._current_sentence self._current_sentence = "" + self._processing_text = False await self._push_tts_frames(sentence) if isinstance(frame, LLMFullResponseEndFrame): if self._push_text_frames: @@ -318,9 +331,16 @@ class TTSService(AIService): await self.push_frame(frame, direction) elif isinstance(frame, TTSSpeakFrame): await self._push_tts_frames(frame.text) + # We pause processing incoming frames because we are sending data to + # the TTS. We pause to avoid audio overlapping. + await self._maybe_pause_frame_processing() await self.flush_audio() + self._processing_text = False elif isinstance(frame, TTSUpdateSettingsFrame): await self._update_settings(frame.settings) + elif isinstance(frame, BotStoppedSpeakingFrame): + await self._maybe_resume_frame_processing() + await self.push_frame(frame, direction) else: await self.push_frame(frame, direction) @@ -347,9 +367,17 @@ class TTSService(AIService): async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection): self._current_sentence = "" + self._processing_text = False if self._text_filter: self._text_filter.handle_interruption() - await self.push_frame(frame, direction) + + async def _maybe_pause_frame_processing(self): + if self._processing_text and self._pause_frame_processing: + await self.pause_processing_frames() + + async def _maybe_resume_frame_processing(self): + if self._pause_frame_processing: + await self.resume_processing_frames() async def _process_text_frame(self, frame: TextFrame): text: Optional[str] = None @@ -371,6 +399,11 @@ class TTSService(AIService): if not text.strip(): return + # This is just a flag that indicates if we sent something to the TTS + # service. It will be cleared if we sent text because of a TTSSpeakFrame + # or when we received an LLMFullResponseEndFrame + self._processing_text = True + await self.start_processing_metrics() if self._text_filter: self._text_filter.reset_interruption() diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index 49bcac2f1..67fc67ebf 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -109,6 +109,7 @@ class CartesiaTTSService(AudioContextWordTTSService, WebsocketService): self, aggregate_sentences=True, push_text_frames=False, + pause_frame_processing=True, sample_rate=sample_rate, **kwargs, ) @@ -274,19 +275,6 @@ class CartesiaTTSService(AudioContextWordTTSService, WebsocketService): else: logger.error(f"{self} error, unknown message type: {msg}") - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - # If we received a TTSSpeakFrame and the LLM response included text (it - # might be that it's only a function calling response) we pause - # processing more frames until we receive a BotStoppedSpeakingFrame. - if isinstance(frame, TTSSpeakFrame): - await self.pause_processing_frames() - elif isinstance(frame, LLMFullResponseEndFrame) and self._context_id: - await self.pause_processing_frames() - elif isinstance(frame, BotStoppedSpeakingFrame): - await self.resume_processing_frames() - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: [{text}]") diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index b2047d34e..3c8cbb3ed 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -192,6 +192,7 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService): push_text_frames=False, push_stop_frames=True, stop_frame_timeout_s=2.0, + pause_frame_processing=True, sample_rate=sample_rate, **kwargs, ) @@ -289,19 +290,6 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService): if isinstance(frame, TTSStoppedFrame): await self.add_word_timestamps([("LLMFullResponseEndFrame", 0), ("Reset", 0)]) - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - # If we received a TTSSpeakFrame and the LLM response included text (it - # might be that it's only a function calling response) we pause - # processing more frames until we receive a BotStoppedSpeakingFrame. - if isinstance(frame, TTSSpeakFrame): - await self.pause_processing_frames() - elif isinstance(frame, LLMFullResponseEndFrame) and self._started: - await self.pause_processing_frames() - elif isinstance(frame, BotStoppedSpeakingFrame): - await self.resume_processing_frames() - async def _connect(self): await self._connect_websocket() diff --git a/src/pipecat/services/fal.py b/src/pipecat/services/fal.py index d229861f1..7173861ab 100644 --- a/src/pipecat/services/fal.py +++ b/src/pipecat/services/fal.py @@ -4,6 +4,7 @@ # SPDX-License-Identifier: BSD 2-Clause License # +import asyncio import io import os from typing import AsyncGenerator, Dict, Optional, Union @@ -53,6 +54,11 @@ class FalImageGenService(ImageGenService): os.environ["FAL_KEY"] = key async def run_image_gen(self, prompt: str) -> AsyncGenerator[Frame, None]: + def load_image_bytes(encoded_image: bytes): + buffer = io.BytesIO(encoded_image) + image = Image.open(buffer) + return (image.tobytes(), image.size, image.format) + logger.debug(f"Generating image from prompt: {prompt}") response = await fal_client.run_async( @@ -73,10 +79,8 @@ class FalImageGenService(ImageGenService): logger.debug(f"Downloading image {image_url} ...") async with self._aiohttp_session.get(image_url) as response: logger.debug(f"Downloaded image {image_url}") - image_stream = io.BytesIO(await response.content.read()) - image = Image.open(image_stream) + encoded_image = await response.content.read() + (image_bytes, size, format) = await asyncio.to_thread(load_image_bytes, encoded_image) - frame = URLImageRawFrame( - url=image_url, image=image.tobytes(), size=image.size, format=image.format - ) + frame = URLImageRawFrame(url=image_url, image=image_bytes, size=size, format=format) yield frame diff --git a/src/pipecat/services/fish.py b/src/pipecat/services/fish.py index d61514eb2..94475aca7 100644 --- a/src/pipecat/services/fish.py +++ b/src/pipecat/services/fish.py @@ -60,7 +60,7 @@ class FishAudioTTSService(TTSService, WebsocketService): params: InputParams = InputParams(), **kwargs, ): - super().__init__(sample_rate=sample_rate, **kwargs) + super().__init__(pause_frame_processing=True, sample_rate=sample_rate, **kwargs) self._api_key = api_key self._base_url = "wss://api.fish.audio/v1/tts/live" @@ -166,16 +166,6 @@ class FishAudioTTSService(TTSService, WebsocketService): except Exception as e: logger.error(f"Error processing message: {e}") - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - if isinstance(frame, TTSSpeakFrame): - await self.pause_processing_frames() - elif isinstance(frame, LLMFullResponseEndFrame) and self._request_id: - await self.pause_processing_frames() - elif isinstance(frame, BotStoppedSpeakingFrame): - await self.resume_processing_frames() - async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection): await super()._handle_interruption(frame, direction) await self.stop_all_metrics() diff --git a/src/pipecat/services/lmnt.py b/src/pipecat/services/lmnt.py index 645494cc1..0272690e8 100644 --- a/src/pipecat/services/lmnt.py +++ b/src/pipecat/services/lmnt.py @@ -73,6 +73,7 @@ class LmntTTSService(TTSService, WebsocketService): TTSService.__init__( self, push_stop_frames=True, + pause_frame_processing=True, sample_rate=sample_rate, **kwargs, ) diff --git a/src/pipecat/services/playht.py b/src/pipecat/services/playht.py index bd9463f91..297e802d3 100644 --- a/src/pipecat/services/playht.py +++ b/src/pipecat/services/playht.py @@ -120,6 +120,7 @@ class PlayHTTTSService(TTSService, WebsocketService): ): TTSService.__init__( self, + pause_frame_processing=True, sample_rate=sample_rate, **kwargs, ) @@ -269,19 +270,6 @@ class PlayHTTTSService(TTSService, WebsocketService): except json.JSONDecodeError: logger.error(f"Invalid JSON message: {message}") - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - # If we received a TTSSpeakFrame and the LLM response included text (it - # might be that it's only a function calling response) we pause - # processing more frames until we receive a BotStoppedSpeakingFrame. - if isinstance(frame, TTSSpeakFrame): - await self.pause_processing_frames() - elif isinstance(frame, LLMFullResponseEndFrame) and self._request_id: - await self.pause_processing_frames() - elif isinstance(frame, BotStoppedSpeakingFrame): - await self.resume_processing_frames() - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: [{text}]") diff --git a/src/pipecat/services/rime.py b/src/pipecat/services/rime.py index 1634578cd..f87405a63 100644 --- a/src/pipecat/services/rime.py +++ b/src/pipecat/services/rime.py @@ -101,6 +101,7 @@ class RimeTTSService(AudioContextWordTTSService, WebsocketService): push_text_frames=False, push_stop_frames=True, stop_frame_timeout_s=2.0, + pause_frame_processing=True, sample_rate=sample_rate, **kwargs, ) @@ -126,7 +127,6 @@ class RimeTTSService(AudioContextWordTTSService, WebsocketService): # State tracking self._context_id = None # Tracks current turn self._receive_task = None - self._started = False self._cumulative_time = 0 # Accumulates time across messages def can_generate_metrics(self) -> bool: @@ -200,7 +200,6 @@ class RimeTTSService(AudioContextWordTTSService, WebsocketService): await self._websocket.send(json.dumps(self._build_eos_msg())) await self._websocket.close() self._websocket = None - self._started = False self._context_id = None except Exception as e: logger.error(f"{self} error closing websocket: {e}") @@ -217,7 +216,6 @@ class RimeTTSService(AudioContextWordTTSService, WebsocketService): await self.stop_all_metrics() if self._context_id: await self._get_websocket().send(json.dumps(self._build_clear_msg())) - self._started = False self._context_id = None def _calculate_word_times(self, words: list, starts: list, ends: list) -> list: @@ -300,21 +298,9 @@ class RimeTTSService(AudioContextWordTTSService, WebsocketService): """Push frame and handle end-of-turn conditions.""" await super().push_frame(frame, direction) if isinstance(frame, (TTSStoppedFrame, StartInterruptionFrame)): - self._started = False if isinstance(frame, TTSStoppedFrame): await self.add_word_timestamps([("LLMFullResponseEndFrame", 0), ("Reset", 0)]) - async def process_frame(self, frame: Frame, direction: FrameDirection): - """Process frames and manage turn state.""" - await super().process_frame(frame, direction) - - if isinstance(frame, TTSSpeakFrame): - await self.pause_processing_frames() - elif isinstance(frame, LLMFullResponseEndFrame) and self._started: - await self.pause_processing_frames() - elif isinstance(frame, BotStoppedSpeakingFrame): - await self.resume_processing_frames() - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: """Generate speech from text. @@ -330,10 +316,9 @@ class RimeTTSService(AudioContextWordTTSService, WebsocketService): await self._connect() try: - if not self._started: + if not self._context_id: await self.start_ttfb_metrics() yield TTSStartedFrame() - self._started = True self._cumulative_time = 0 self._context_id = str(uuid.uuid4()) await self.create_audio_context(self._context_id) diff --git a/src/pipecat/transports/local/audio.py b/src/pipecat/transports/local/audio.py index 30bdaaf79..e7b803f2b 100644 --- a/src/pipecat/transports/local/audio.py +++ b/src/pipecat/transports/local/audio.py @@ -26,17 +26,18 @@ except ModuleNotFoundError as e: raise Exception(f"Missing module: {e}") -class LocalTransportParams(TransportParams): - input_device_index: int = 0 - output_device_index: int = 0 +class LocalAudioTransportParams(TransportParams): + input_device_index: Optional[int] = None + output_device_index: Optional[int] = None class LocalAudioInputTransport(BaseInputTransport): - _params: LocalTransportParams + _params: LocalAudioTransportParams - def __init__(self, py_audio: pyaudio.PyAudio, params: LocalTransportParams): + def __init__(self, py_audio: pyaudio.PyAudio, params: LocalAudioTransportParams): super().__init__(params) self._py_audio = py_audio + self._in_stream = None self._sample_rate = 0 @@ -77,11 +78,12 @@ class LocalAudioInputTransport(BaseInputTransport): class LocalAudioOutputTransport(BaseOutputTransport): - _params: LocalTransportParams + _params: LocalAudioTransportParams - def __init__(self, py_audio: pyaudio.PyAudio, params: TransportParams): + def __init__(self, py_audio: pyaudio.PyAudio, params: LocalAudioTransportParams): super().__init__(params) self._py_audio = py_audio + self._out_stream = None self._sample_rate = 0 @@ -117,7 +119,7 @@ class LocalAudioOutputTransport(BaseOutputTransport): class LocalAudioTransport(BaseTransport): - def __init__(self, params: LocalTransportParams): + def __init__(self, params: LocalAudioTransportParams): super().__init__() self._params = params self._pyaudio = pyaudio.PyAudio() diff --git a/src/pipecat/transports/local/tk.py b/src/pipecat/transports/local/tk.py index 40147d39f..73777c573 100644 --- a/src/pipecat/transports/local/tk.py +++ b/src/pipecat/transports/local/tk.py @@ -34,8 +34,15 @@ except ModuleNotFoundError as e: raise Exception(f"Missing module: {e}") +class TkTransportParams(TransportParams): + audio_input_device_index: Optional[int] = None + audio_output_device_index: Optional[int] = None + + class TkInputTransport(BaseInputTransport): - def __init__(self, py_audio: pyaudio.PyAudio, params: TransportParams): + _params: TkTransportParams + + def __init__(self, py_audio: pyaudio.PyAudio, params: TkTransportParams): super().__init__(params) self._py_audio = py_audio self._in_stream = None @@ -54,6 +61,7 @@ class TkInputTransport(BaseInputTransport): frames_per_buffer=num_frames, stream_callback=self._audio_in_callback, input=True, + input_device_index=self._params.audio_input_device_index, ) self._in_stream.start_stream() @@ -76,6 +84,8 @@ class TkInputTransport(BaseInputTransport): class TkOutputTransport(BaseOutputTransport): + _params: TkTransportParams + def __init__(self, tk_root: tk.Tk, py_audio: pyaudio.PyAudio, params: TransportParams): super().__init__(params) self._py_audio = py_audio @@ -103,6 +113,7 @@ class TkOutputTransport(BaseOutputTransport): channels=self._params.audio_out_channels, rate=self._sample_rate, output=True, + output_device_index=self._params.audio_output_device_index, ) self._out_stream.start_stream() diff --git a/src/pipecat/utils/string.py b/src/pipecat/utils/string.py index c269c5e57..916dfad41 100644 --- a/src/pipecat/utils/string.py +++ b/src/pipecat/utils/string.py @@ -13,8 +13,8 @@ ENDOFSENTENCE_PATTERN_STR = r""" (?