Merge pull request #2525 from pipecat-ai/aleix/daily-fix-send-dtmf

DailyTransport: fix sending DTMF tones

CHANGELOG.md

@@ -96,6 +96,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed a `DailyTransport` issue that prevented DTMF tones from being sent.
+
 - Fixed a missing import in `SentryMetrics`.
 
 - Fixed `AWSPollyTTSService` to support AWS credential provider chain (IAM

src/pipecat/pipeline/task.py

@@ -58,7 +58,7 @@ HEARTBEAT_MONITOR_SECS = HEARTBEAT_SECS * 10
 
 IDLE_TIMEOUT_SECS = 300
 
-CANCEL_TIMEOUT_SECS = 10.0
+CANCEL_TIMEOUT_SECS = 20.0
 
 
 class PipelineParams(BaseModel):

src/pipecat/transports/services/daily.py

@@ -350,6 +350,7 @@ class DailyTransportClient(EventHandler):
         self._video_renderers = {}
         self._transcription_ids = []
         self._transcription_status = None
+        self._dial_out_session_id: str = ""
 
         self._joining = False
         self._joined = False
@@ -639,6 +640,9 @@ class DailyTransportClient(EventHandler):
 
     async def _join(self):
         """Execute the actual room join operation."""
+        if not self._client:
+            return
+
         future = self._get_event_loop().create_future()
 
         camera_enabled = self._params.video_out_enabled and self._params.camera_out_enabled
@@ -730,6 +734,9 @@ class DailyTransportClient(EventHandler):
 
     async def _leave(self):
         """Execute the actual room leave operation."""
+        if not self._client:
+            return
+
         future = self._get_event_loop().create_future()
         self._client.leave(completion=completion_callback(future))
         return await asyncio.wait_for(future, timeout=10)
@@ -790,6 +797,14 @@ class DailyTransportClient(EventHandler):
         Args:
             settings: DTMF settings including tones and target session.
         """
+        session_id = settings.get("sessionId") or self._dial_out_session_id
+        if not session_id:
+            logger.error("Unable to send DTMF: 'sessionId' is not set")
+            return
+
+        # Update 'sessionId' field.
+        settings["sessionId"] = session_id
+
         future = self._get_event_loop().create_future()
         self._client.send_dtmf(settings, completion=completion_callback(future))
         await future
@@ -1166,6 +1181,7 @@ class DailyTransportClient(EventHandler):
         Args:
             data: Dial-out connection data.
         """
+        self._dial_out_session_id = data["sessionId"] if "sessionId" in data else ""
         self._call_event_callback(self._callbacks.on_dialout_connected, data)
 
     def on_dialout_stopped(self, data: Any):
@@ -1174,6 +1190,9 @@ class DailyTransportClient(EventHandler):
         Args:
             data: Dial-out stop data.
         """
+        # Cleanup only if our session stopped.
+        if data["sessionId"] == self._dial_out_session_id:
+            self._dial_out_session_id = ""
         self._call_event_callback(self._callbacks.on_dialout_stopped, data)
 
     def on_dialout_error(self, data: Any):
@@ -1182,6 +1201,9 @@ class DailyTransportClient(EventHandler):
         Args:
             data: Dial-out error data.
         """
+        # Cleanup only if our session errored out.
+        if data["sessionId"] == self._dial_out_session_id:
+            self._dial_out_session_id = ""
         self._call_event_callback(self._callbacks.on_dialout_error, data)
 
     def on_dialout_warning(self, data: Any):