From 05ae3a17033480d844ed15e9511a98bf8072a5a6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 28 May 2025 11:21:26 -0700 Subject: [PATCH] DailyTransport: expose start_transcription/stop_transcription --- CHANGELOG.md | 8 +++ src/pipecat/audio/vad/vad_analyzer.py | 2 +- src/pipecat/transports/services/daily.py | 90 +++++++++++++++--------- 3 files changed, 67 insertions(+), 33 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 184d96239..5d44d108b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,14 @@ 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] + +### Added + +- Added new functions `DailyTransport.start_transcription()` and + `DailyTransport.stop_transcription()` to be able to start and stop Daily + transcription dynamically (maybe with different settings). + ## [0.0.68] - 2025-05-28 ### Added diff --git a/src/pipecat/audio/vad/vad_analyzer.py b/src/pipecat/audio/vad/vad_analyzer.py index 915649031..073e61712 100644 --- a/src/pipecat/audio/vad/vad_analyzer.py +++ b/src/pipecat/audio/vad/vad_analyzer.py @@ -71,7 +71,7 @@ class VADAnalyzer(ABC): self.set_params(self._params) def set_params(self, params: VADParams): - logger.info(f"Setting VAD params to: {params}") + logger.debug(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 diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 03d9e14ca..0011dd2fd 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -476,7 +476,7 @@ class DailyTransportClient(EventHandler): logger.info(f"Joined {self._room_url}") if self._params.transcription_enabled: - await self._start_transcription() + await self.start_transcription(self._params.transcription_settings) await self._callbacks.on_joined(data) @@ -491,23 +491,6 @@ class DailyTransportClient(EventHandler): self._joining = False await self._callbacks.on_error(error_msg) - async def _start_transcription(self): - if not self._token: - logger.warning("Transcription can't be started without a room token") - return - - logger.info(f"Enabling transcription with settings {self._params.transcription_settings}") - - future = self._get_event_loop().create_future() - self._client.start_transcription( - settings=self._params.transcription_settings.model_dump(exclude_none=True), - completion=completion_callback(future), - ) - error = await future - if error: - logger.error(f"Unable to start transcription: {error}") - return - async def _join(self): future = self._get_event_loop().create_future() @@ -577,7 +560,7 @@ class DailyTransportClient(EventHandler): logger.info(f"Leaving {self._room_url}") if self._params.transcription_enabled: - await self._stop_transcription() + await self.stop_transcription() # Remove any custom tracks, if any. for track_name, _ in self._custom_audio_tracks.items(): @@ -597,15 +580,6 @@ class DailyTransportClient(EventHandler): logger.error(error_msg) await self._callbacks.on_error(error_msg) - async def _stop_transcription(self): - if not self._token: - return - future = self._get_event_loop().create_future() - self._client.stop_transcription(completion=completion_callback(future)) - error = await future - if error: - logger.error(f"Unable to stop transcription: {error}") - async def _leave(self): future = self._get_event_loop().create_future() self._client.leave(completion=completion_callback(future)) @@ -623,14 +597,22 @@ class DailyTransportClient(EventHandler): return self._client.participant_counts() async def start_dialout(self, settings): + logger.debug(f"Starting dialout: settings={settings}") + future = self._get_event_loop().create_future() self._client.start_dialout(settings, completion=completion_callback(future)) - await future + error = await future + if error: + logger.error(f"Unable to start dialout: {error}") async def stop_dialout(self, participant_id): + logger.debug(f"Stopping dialout: participant_id={participant_id}") + future = self._get_event_loop().create_future() self._client.stop_dialout(participant_id, completion=completion_callback(future)) - await future + error = await future + if error: + logger.error(f"Unable to stop dialout: {error}") async def send_dtmf(self, settings): future = self._get_event_loop().create_future() @@ -648,16 +630,54 @@ class DailyTransportClient(EventHandler): await future async def start_recording(self, streaming_settings, stream_id, force_new): + logger.debug( + f"Starting recording: stream_id={stream_id} force_new={force_new} settings={streaming_settings}" + ) + future = self._get_event_loop().create_future() self._client.start_recording( streaming_settings, stream_id, force_new, completion=completion_callback(future) ) - await future + error = await future + if error: + logger.error(f"Unable to start recording: {error}") async def stop_recording(self, stream_id): + logger.debug(f"Stopping recording: stream_id={stream_id}") + future = self._get_event_loop().create_future() self._client.stop_recording(stream_id, completion=completion_callback(future)) - await future + error = await future + if error: + logger.error(f"Unable to stop recording: {error}") + + async def start_transcription(self, settings): + if not self._token: + logger.warning("Transcription can't be started without a room token") + return + + logger.debug(f"Starting transcription: settings={settings}") + + future = self._get_event_loop().create_future() + self._client.start_transcription( + settings=self._params.transcription_settings.model_dump(exclude_none=True), + completion=completion_callback(future), + ) + error = await future + if error: + logger.error(f"Unable to start transcription: {error}") + + async def stop_transcription(self): + if not self._token: + return + + logger.debug(f"Stopping transcription") + + future = self._get_event_loop().create_future() + self._client.stop_transcription(completion=completion_callback(future)) + error = await future + if error: + logger.error(f"Unable to stop transcription: {error}") async def send_prebuilt_chat_message(self, message: str, user_name: Optional[str] = None): if not self._joined: @@ -1366,6 +1386,12 @@ class DailyTransport(BaseTransport): async def stop_recording(self, stream_id=None): await self._client.stop_recording(stream_id) + async def start_transcription(self, settings=None): + await self._client.start_transcription(settings) + + async def stop_transcription(self): + await self._client.stop_transcription() + async def send_prebuilt_chat_message(self, message: str, user_name: Optional[str] = None): """Sends a chat message to Daily's Prebuilt main room.