diff --git a/CHANGELOG.md b/CHANGELOG.md index a1753c421..432eaf9bb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,6 +28,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed an issue where `camera_out_enabled` would cause the highg CPU usage if no image was provided. +### Performance + +- Removed unnecessary audio input tasks. + ## [0.0.24] - 2024-05-29 ### Added diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 71733c59c..c09c5a47c 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -59,10 +59,7 @@ class BaseInputTransport(FrameProcessor): if self._params.audio_in_enabled or self._params.vad_enabled: loop = self.get_event_loop() - self._audio_in_thread = loop.run_in_executor( - self._in_executor, self._audio_in_thread_handler) - self._audio_out_thread = loop.run_in_executor( - self._in_executor, self._audio_out_thread_handler) + self._audio_thread = loop.run_in_executor(self._in_executor, self._audio_thread_handler) async def stop(self): if not self._running: @@ -73,8 +70,7 @@ class BaseInputTransport(FrameProcessor): # Wait for the threads to finish. if self._params.audio_in_enabled or self._params.vad_enabled: - await self._audio_in_thread - await self._audio_out_thread + await self._audio_thread self._push_frame_task.cancel() @@ -167,7 +163,9 @@ class BaseInputTransport(FrameProcessor): vad_state = new_vad_state return vad_state - def _audio_in_thread_handler(self): + def _audio_thread_handler(self): + vad_state: VADState = VADState.QUIET + sample_rate = self._params.audio_in_sample_rate num_channels = self._params.audio_in_channels num_frames = int(sample_rate / 100) # 10ms of audio @@ -179,32 +177,19 @@ class BaseInputTransport(FrameProcessor): audio=audio_frames, sample_rate=sample_rate, num_channels=num_channels) - self._audio_in_queue.put(frame) + + audio_passthrough = True + + # Check VAD and push event if necessary. We just care about + # changes from QUIET to SPEAKING and vice versa. + if self._params.vad_enabled: + vad_state = self._handle_vad(frame.audio, vad_state) + audio_passthrough = self._params.vad_audio_passthrough + + # Push audio downstream if passthrough. + if audio_passthrough: + future = asyncio.run_coroutine_threadsafe( + self._internal_push_frame(frame), self.get_event_loop()) + future.result() except BaseException as e: logger.error(f"Error reading audio frames: {e}") - - def _audio_out_thread_handler(self): - vad_state: VADState = VADState.QUIET - while self._running: - try: - frame = self._audio_in_queue.get(timeout=1) - - audio_passthrough = True - - # Check VAD and push event if necessary. We just care about changes - # from QUIET to SPEAKING and vice versa. - if self._params.vad_enabled: - vad_state = self._handle_vad(frame.audio, vad_state) - audio_passthrough = self._params.vad_audio_passthrough - - # Push audio downstream if passthrough. - if audio_passthrough: - future = asyncio.run_coroutine_threadsafe( - self._internal_push_frame(frame), self.get_event_loop()) - future.result() - - self._audio_in_queue.task_done() - except queue.Empty: - pass - except BaseException as e: - logger.error(f"Error pushing audio frames: {e}")