diff --git a/.claude/skills/changelog/SKILL.md b/.claude/skills/changelog/SKILL.md index 89e13a40e..1ef8f324e 100644 --- a/.claude/skills/changelog/SKILL.md +++ b/.claude/skills/changelog/SKILL.md @@ -26,7 +26,7 @@ Create changelog files for the important commits in this PR. The PR number is pr - `{PR_NUMBER}.performance.md` - for performance improvements - `{PR_NUMBER}.other.md` - for other changes -4. Each changelog file should at least contain a main single line starting with `- ` followed by a clear description of the change. +4. Each changelog file should at least contain a main single line starting with `- ` followed by a clear description of the change. No line wrapping. 5. If the change is complicated, changelog files can have indented lines after the main line with additional details or code samples. diff --git a/changelog/3719.added.2.md b/changelog/3719.added.2.md new file mode 100644 index 000000000..77d8956d7 --- /dev/null +++ b/changelog/3719.added.2.md @@ -0,0 +1 @@ +- Added `write_transport_frame()` hook to `BaseOutputTransport` allowing transport subclasses to handle custom frame types that flow through the audio queue. diff --git a/changelog/3719.added.md b/changelog/3719.added.md new file mode 100644 index 000000000..bc1c2d6b1 --- /dev/null +++ b/changelog/3719.added.md @@ -0,0 +1 @@ +- Added `DailySIPTransferFrame` and `DailySIPReferFrame` to the Daily transport. These frames queue SIP transfer and SIP REFER operations with audio, so the operation executes only after the bot finishes its current utterance. diff --git a/changelog/3719.changed.md b/changelog/3719.changed.md new file mode 100644 index 000000000..f42d0303b --- /dev/null +++ b/changelog/3719.changed.md @@ -0,0 +1 @@ +- `DailyUpdateRemoteParticipantsFrame` is no longer deprecated and is now queued with audio like other transport frames. diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index c4f59a61d..6fe31aab5 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -237,6 +237,18 @@ class BaseOutputTransport(FrameProcessor): else: await self._write_dtmf_audio(frame) + async def write_transport_frame(self, frame: Frame): + """Handle a queued frame after preceding audio has been sent. + + Override in transport subclasses to handle custom frame types that + flow through the audio queue. Called by the media sender after the + frame has waited for any preceding audio to finish. + + Args: + frame: The frame to handle. + """ + pass + def _supports_native_dtmf(self) -> bool: """Override in transport implementations that support native DTMF. @@ -681,6 +693,8 @@ class BaseOutputTransport(FrameProcessor): await self._transport.send_message(frame) elif isinstance(frame, OutputDTMFFrame): await self._transport.write_dtmf(frame) + else: + await self._transport.write_transport_frame(frame) def _next_frame(self) -> AsyncGenerator[Frame, None]: """Generate the next frame for audio processing. diff --git a/src/pipecat/transports/daily/transport.py b/src/pipecat/transports/daily/transport.py index e3b246253..5df7bcffd 100644 --- a/src/pipecat/transports/daily/transport.py +++ b/src/pipecat/transports/daily/transport.py @@ -15,7 +15,7 @@ import asyncio import time from concurrent.futures import CancelledError as FuturesCancelledError from concurrent.futures import ThreadPoolExecutor -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import Any, Awaitable, Callable, Dict, Mapping, Optional, Tuple import aiohttp @@ -25,7 +25,7 @@ from pydantic import BaseModel from pipecat.audio.vad.vad_analyzer import VADAnalyzer, VADParams from pipecat.frames.frames import ( CancelFrame, - ControlFrame, + DataFrame, EndFrame, Frame, InputAudioRawFrame, @@ -183,34 +183,44 @@ class DailyInputTransportMessageUrgentFrame(DailyInputTransportMessageFrame): @dataclass -class DailyUpdateRemoteParticipantsFrame(ControlFrame): - """Frame to update remote participants in Daily calls. +class DailySIPTransferFrame(DataFrame): + """SIP call transfer frame for transport queuing. - .. deprecated:: 0.0.87 - `DailyUpdateRemoteParticipantsFrame` is deprecated and will be removed in a future version. - Create your own custom frame and use a custom processor to handle it or use, for example, - `on_after_push_frame` event instead in the output transport. + A SIP call transfer that will be queued. The transfer will happen after any + preceding audio finishes playing, allowing the bot to complete its current + utterance before the transfer occurs. + + Parameters: + settings: SIP call transfer settings. + """ + + settings: Mapping[str, Any] = field(default_factory=dict) + + +@dataclass +class DailySIPReferFrame(DataFrame): + """SIP REFER frame for transport queuing. + + A SIP REFER that will be queued. The REFER will happen after any preceding + audio finishes playing, allowing the bot to complete its current utterance + before the REFER occurs. + + Parameters: + settings: SIP REFER settings. + """ + + settings: Mapping[str, Any] = field(default_factory=dict) + + +@dataclass +class DailyUpdateRemoteParticipantsFrame(DataFrame): + """Frame to update remote participants in Daily calls. Parameters: remote_participants: See https://reference-python.daily.co/api_reference.html#daily.CallClient.update_remote_participants. """ - remote_participants: Mapping[str, Any] = None - - def __post_init__(self): - super().__post_init__() - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "DailyUpdateRemoteParticipantsFrame is deprecated and will be removed in a future version." - "Instead, create your own custom frame and handle it in the " - '`@transport.output().event_handler("on_after_push_frame")` event handler or a ' - "custom processor.", - DeprecationWarning, - stacklevel=2, - ) + remote_participants: Mapping[str, Any] = field(default_factory=dict) class WebRTCVADAnalyzer(VADAnalyzer): @@ -1946,18 +1956,6 @@ class DailyOutputTransport(BaseOutputTransport): # Leave the room. await self._client.leave() - async def process_frame(self, frame: Frame, direction: FrameDirection): - """Process outgoing frames, including transport messages. - - Args: - frame: The frame to process. - direction: The direction of frame flow in the pipeline. - """ - await super().process_frame(frame, direction) - - if isinstance(frame, DailyUpdateRemoteParticipantsFrame): - await self._client.update_remote_participants(frame.remote_participants) - async def send_message( self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame ): @@ -1968,7 +1966,7 @@ class DailyOutputTransport(BaseOutputTransport): """ error = await self._client.send_message(frame) if error: - logger.error(f"Unable to send message: {error}") + await self.push_error(f"Unable to send message: {error}") async def register_video_destination(self, destination: str): """Register a video output destination. @@ -2011,6 +2009,25 @@ class DailyOutputTransport(BaseOutputTransport): """ return await self._client.write_video_frame(frame) + async def write_transport_frame(self, frame: Frame): + """Handle queued SIP frames after preceding audio has been sent. + + Args: + frame: The frame to handle. + """ + if isinstance(frame, DailySIPTransferFrame): + error = await self._client.sip_call_transfer(frame.settings) + if error: + await self.push_error(f"Unable to transfer SIP call: {error}") + elif isinstance(frame, DailySIPReferFrame): + error = await self._client.sip_refer(frame.settings) + if error: + await self.push_error(f"Unable to perform SIP REFER: {error}") + elif isinstance(frame, DailyUpdateRemoteParticipantsFrame): + error = await self._client.update_remote_participants(frame.remote_participants) + if error: + await self.push_error(f"Unable to update remote participants: {error}") + def _supports_native_dtmf(self) -> bool: """Daily supports native DTMF via telephone events.