diff --git a/examples/foundational/25-simli-layer.py b/examples/foundational/25-simli-layer.py index ee772dc08..15b62047f 100644 --- a/examples/foundational/25-simli-layer.py +++ b/examples/foundational/25-simli-layer.py @@ -32,8 +32,7 @@ logger.add(sys.stderr, level="DEBUG") async def main(): async with aiohttp.ClientSession() as session: - (room_url, token) = await configure(session) - + _, token = await configure(session) print("Creating room") aiohttp_session = aiohttp.ClientSession() daily_helper = DailyRESTHelper( @@ -41,15 +40,11 @@ async def main(): daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"), aiohttp_session=aiohttp_session, ) - room = await daily_helper.create_room(DailyRoomParams()) expiry_time: float = 60 * 60 token = await daily_helper.get_token(room.url, expiry_time) - - print("Room created ", room.url) - transport = DailyTransport( room.url, token, @@ -84,7 +79,7 @@ async def main(): ) llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o-mini") - + messages = [ { "role": "system", @@ -98,13 +93,9 @@ async def main(): # "content": "Eres Chatbot, un amigable y útil robot. Tu objetivo es demostrar tus capacidades de una manera breve. Tus respuestas se convertiran a audio así que nunca no debes incluir caracteres especiales. Contesta a lo que el usuario pregunte de una manera creativa, útil y breve. Empieza por presentarte a ti mismo.", }, ] - simliAi = SimliVideoService( + simli_ai = SimliVideoService( SimliConfig(os.getenv("SIMLI_API_KEY"), os.getenv("SIMLI_FACE_ID")) ) - print("starting connection to simi") - await simliAi.startConnection() - print("connection started") - context = OpenAILLMContext(messages) context_aggregator = llm.create_context_aggregator(context) @@ -114,7 +105,7 @@ async def main(): context_aggregator.user(), llm, tts, - simliAi, + simli_ai, transport.output(), context_aggregator.assistant(), ] diff --git a/src/pipecat/services/simli.py b/src/pipecat/services/simli.py index d25264438..5813cb27e 100644 --- a/src/pipecat/services/simli.py +++ b/src/pipecat/services/simli.py @@ -8,58 +8,57 @@ from pipecat.frames.frames import ( EndFrame, CancelFrame, ) -from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, StartFrame import numpy as np from av import AudioFrame from av.audio.resampler import AudioResampler + from simli import SimliClient, SimliConfig +from loguru import logger class SimliVideoService(FrameProcessor): - def __init__( - self, simliConfig: SimliConfig, useTurnServer=False, latencyInterval=60 - ): + def __init__(self, simli_config: SimliConfig, use_turn_server=False, latency_interval=0): super().__init__() - self.simliClient = SimliClient(simliConfig, useTurnServer, latencyInterval) + self._simli_client = SimliClient(simli_config, use_turn_server, latency_interval) - self.pipecatResampler: AudioResampler = None - self.name = "SimliAi" - self.ready = False - self.simliResampler = AudioResampler("s16", 1, 16000) - self.AudioTask: asyncio.Task = None - self.VideoTask: asyncio.Task = None + self._ready = False + self._pipecat_resampler: AudioResampler = None + self._simli_resampler = AudioResampler("s16", 1, 16000) - async def startConnection(self): - await self.simliClient.Initialize() - self.ready = True + self._audio_task: asyncio.Task = None + self._video_task: asyncio.Task = None + + async def _start_connection(self): + await self._simli_client.Initialize() + self._ready = True # Create task to consume and process audio and video - self.AudioTask = asyncio.create_task(self.consume_and_process_audio()) - self.VideoTask = asyncio.create_task(self.consume_and_process_video()) + self._audio_task = asyncio.create_task(self._consume_and_process_audio()) + self._video_task = asyncio.create_task(self._consume_and_process_video()) - async def consume_and_process_audio(self): - async for audio_frame in self.simliClient.getAudioStreamIterator(): + async def _consume_and_process_audio(self): + while self._pipecat_resampler is None: + await asyncio.sleep(0.001) + async for audio_frame in self._simli_client.getAudioStreamIterator(): # Process the audio frame try: - resampledFrames = self.pipecatResampler.resample(audio_frame) - for resampled_frame in resampledFrames: + resampled_frames = self._pipecat_resampler.resample(audio_frame) + for resampled_frame in resampled_frames: await self.push_frame( TTSAudioRawFrame( audio=resampled_frame.to_ndarray().tobytes(), - sample_rate=self.pipecatResampler.rate, + sample_rate=self._pipecat_resampler.rate, num_channels=1, ), ) except Exception as e: - print(e) - import traceback + logger.exception(f"{self} exception: {e}") - traceback.print_exc() - - async def consume_and_process_video(self): - async for video_frame in self.simliClient.getVideoStreamIterator( - targetFormat="rgb24" - ): + async def _consume_and_process_video(self): + while self._pipecat_resampler is None: + await asyncio.sleep(0.001) + async for video_frame in self._simli_client.getVideoStreamIterator(targetFormat="rgb24"): # Process the video frame convertedFrame: OutputImageRawFrame = OutputImageRawFrame( image=video_frame.to_rgb().to_image().tobytes(), @@ -73,44 +72,41 @@ class SimliVideoService(FrameProcessor): async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) - - if isinstance(frame, TTSAudioRawFrame): + if isinstance(frame, StartFrame): + await self._start_connection() + elif isinstance(frame, TTSAudioRawFrame): # Send audio frame to Simli try: - if self.ready: - AudioFrame + if self._ready: oldFrame = AudioFrame.from_ndarray( np.frombuffer(frame.audio, dtype=np.int16)[None, :], layout=frame.num_channels, ) oldFrame.sample_rate = frame.sample_rate - if self.pipecatResampler is None: - self.pipecatResampler = AudioResampler( + if self._pipecat_resampler is None: + self._pipecat_resampler = AudioResampler( "s16", oldFrame.layout, oldFrame.sample_rate ) - resampledFrame = self.simliResampler.resample(oldFrame) + resampledFrame = self._simli_resampler.resample(oldFrame) for frame in resampledFrame: - await self.simliClient.send( - frame.to_ndarray().astype(np.int16).tobytes() - ) + await self._simli_client.send(frame.to_ndarray().astype(np.int16).tobytes()) return else: - print( + logger.warning( "Simli Connection is not Initialized properly, passing audio to next processor" ) await self.push_frame(frame, direction) except Exception as e: - print(e) - import traceback - - traceback.print_exc() + logger.exception(f"{self} exception: {e}") elif isinstance(frame, (EndFrame, CancelFrame)): - await self.simliClient.stop() - self.AudioTask.cancel() - self.VideoTask.cancel() + await self._simli_client.stop() + self._audio_task.cancel() + await self._audio_task + self._video_task.cancel() + await self._video_task elif isinstance(frame, StartInterruptionFrame): - await self.simliClient.clearBuffer() + await self._simli_client.clearBuffer() await self.push_frame(frame, direction)