diff --git a/examples/foundational/27-simli-layer.py b/examples/foundational/27-simli-layer.py index 40f6b0c3d..a8fdf7c3b 100644 --- a/examples/foundational/27-simli-layer.py +++ b/examples/foundational/27-simli-layer.py @@ -61,7 +61,7 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si ) simli_ai = SimliVideoService( - SimliConfig(os.getenv("SIMLI_API_KEY"), os.getenv("SIMLI_FACE_ID")) + SimliConfig(os.getenv("SIMLI_API_KEY"), os.getenv("SIMLI_FACE_ID")), ) llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o-mini") diff --git a/src/pipecat/services/simli/video.py b/src/pipecat/services/simli/video.py index 6ba7da4b8..41d1c47b9 100644 --- a/src/pipecat/services/simli/video.py +++ b/src/pipecat/services/simli/video.py @@ -18,6 +18,8 @@ from pipecat.frames.frames import ( OutputImageRawFrame, StartInterruptionFrame, TTSAudioRawFrame, + TTSStoppedFrame, + UserStartedSpeakingFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, StartFrame from pipecat.utils.asyncio.watchdog_async_iterator import WatchdogAsyncIterator @@ -45,24 +47,39 @@ class SimliVideoService(FrameProcessor): simli_config: SimliConfig, use_turn_server: bool = False, latency_interval: int = 0, + simli_url: str = "https://api.simli.ai", + isTrinityAvatar: bool = False, ): """Initialize the Simli video service. Args: simli_config: Configuration object for Simli client settings. use_turn_server: Whether to use TURN server for connection. Defaults to False. - latency_interval: Latency interval setting for video processing. Defaults to 0. + latency_interval: Latency interval setting for sending health checks to check the latency to Simli Servers. Defaults to 0. + simli_url: URL of the simli servers. Can be changed for custom deployments of enterprise users. + """ super().__init__() - self._simli_client = SimliClient(simli_config, use_turn_server, latency_interval) + self._initialized = False + simli_config.maxIdleTime += 5 + simli_config.maxSessionLength += 5 + self._simli_client = SimliClient( + simli_config, + use_turn_server, + latency_interval, + simliURL=simli_url, + ) - self._pipecat_resampler_event = asyncio.Event() self._pipecat_resampler: AudioResampler = None + self._pipecat_resampler_event = asyncio.Event() self._simli_resampler = AudioResampler("s16", "mono", 16000) - self._initialized = False self._audio_task: asyncio.Task = None self._video_task: asyncio.Task = None + self._started = False + self._is_gs_avatar = isTrinityAvatar + self._previously_interrupted = True + self._audio_buffer = bytearray() async def _start_connection(self): """Start the connection to Simli service and begin processing tasks.""" @@ -71,11 +88,9 @@ class SimliVideoService(FrameProcessor): self._initialized = True # Create task to consume and process audio and video - if not self._audio_task: - self._audio_task = self.create_task(self._consume_and_process_audio()) - - if not self._video_task: - self._video_task = self.create_task(self._consume_and_process_video()) + await self._simli_client.sendSilence() + self._audio_task = self.create_task(self._consume_and_process_audio()) + self._video_task = self.create_task(self._consume_and_process_video()) async def _consume_and_process_audio(self): """Consume audio frames from Simli and push them downstream.""" @@ -118,8 +133,8 @@ class SimliVideoService(FrameProcessor): """ await super().process_frame(frame, direction) if isinstance(frame, StartFrame): - await self.push_frame(frame, direction) await self._start_connection() + await self.push_frame(frame, direction) elif isinstance(frame, TTSAudioRawFrame): # Send audio frame to Simli try: @@ -137,19 +152,47 @@ class SimliVideoService(FrameProcessor): resampled_frames = self._simli_resampler.resample(old_frame) for resampled_frame in resampled_frames: - await self._simli_client.send( - resampled_frame.to_ndarray().astype(np.int16).tobytes() - ) + audioBytes = resampled_frame.to_ndarray().astype(np.int16).tobytes() + if self._previously_interrupted and self._is_gs_avatar: + self._audio_buffer.extend(audioBytes) + if len(self._audio_buffer) >= 128000: + try: + for flushFrame in self._simli_resampler.resample(None): + self._audio_buffer.extend( + flushFrame.to_ndarray().astype(np.int16).tobytes() + ) + finally: + await self._simli_client.playImmediate(self._audio_buffer) + self._previously_interrupted = False + self._audio_buffer = bytearray() + else: + await self._simli_client.send(audioBytes) + return except Exception as e: logger.exception(f"{self} exception: {e}") + elif isinstance(frame, TTSStoppedFrame): + try: + if self._previously_interrupted and len(self._audio_buffer) > 0: + await self._simli_client.playImmediate(self._audio_buffer) + self._previously_interrupted = False + self._audio_buffer = bytearray() + + except Exception as e: + logger.exception(f"{self} exception: {e}") + return + if isinstance(frame, StartFrame): + if not self._preinitialize: + await self._start_connection() + self._started = True + await self._simli_client.send((0).to_bytes(1, "big") * 6000) elif isinstance(frame, (EndFrame, CancelFrame)): await self._stop() - await self.push_frame(frame, direction) - elif isinstance(frame, StartInterruptionFrame): - await self._simli_client.clearBuffer() - await self.push_frame(frame, direction) - else: - await self.push_frame(frame, direction) + elif isinstance(frame, (StartInterruptionFrame, UserStartedSpeakingFrame)): + if not self._previously_interrupted: + await self._simli_client.clearBuffer() + self._previously_interrupted = True and self._is_gs_avatar + + await self.push_frame(frame, direction) async def _stop(self): """Stop the Simli client and cancel processing tasks."""