diff --git a/CHANGELOG.md b/CHANGELOG.md index 82a5feb0b..b3404d1ab 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,7 +21,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed -- Update `livekit` to 1.0.13. +- Updated `livekit` to 1.0.13. - `torch` and `torchaudio` are no longer required for running Smart Turn locally. This avoids gigabytes of dependencies being installed. @@ -44,6 +44,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fixed an issue that would cause `push_interruption_task_frame_and_wait()` to + not wait if a previous interruption had already happened. + - Fixed a couple of bugs in `ServiceSwitcher`: - Using multiple `ServiceSwitcher`s in a pipeline would result in an error. - `ServiceSwitcherFrame`s (such as `ManuallySwitchServiceFrame`s) were having diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 76400fc22..85ce18548 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -220,6 +220,11 @@ class FrameProcessor(BaseObject): self.__process_event: Optional[asyncio.Event] = None self.__process_frame_task: Optional[asyncio.Task] = None + # To interrupt a pipeline, we push an `InterruptionTaskFrame` upstream. + # Then we wait for the corresponding `InterruptionFrame` to travel from + # the start of the pipeline back to the processor that sent the + # `InterruptionTaskFrame`. This wait is handled using the following + # event. self._wait_for_interruption = False self._wait_interruption_event = asyncio.Event() @@ -632,7 +637,9 @@ class FrameProcessor(BaseObject): await self.__internal_push_frame(frame, direction) - if isinstance(frame, InterruptionFrame): + # If we are waiting for an interruption and we get an interruption, then + # we can unblock `push_interruption_task_frame_and_wait()`. + if self._wait_for_interruption and isinstance(frame, InterruptionFrame): self._wait_interruption_event.set() async def push_interruption_task_frame_and_wait(self): diff --git a/tests/test_frame_processor.py b/tests/test_frame_processor.py new file mode 100644 index 000000000..7cbe258f8 --- /dev/null +++ b/tests/test_frame_processor.py @@ -0,0 +1,67 @@ +# +# Copyright (c) 2024-2025 Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import unittest + +from pipecat.frames.frames import ( + EndFrame, + Frame, + InterruptionFrame, + TextFrame, + TransportMessageUrgentFrame, +) +from pipecat.pipeline.pipeline import Pipeline +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.tests.utils import SleepFrame, run_test + + +class TestFrameProcessor(unittest.IsolatedAsyncioTestCase): + async def test_interruption_and_wait(self): + class DelayFrameProcessor(FrameProcessor): + """This processors just gives time to the event loop to change + between tasks. Otherwise things happen to fast.""" + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + await asyncio.sleep(0.1) + await self.push_frame(frame, direction) + + class InterruptFrameProcessor(FrameProcessor): + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, TextFrame): + await self.push_interruption_task_frame_and_wait() + await self.push_frame(TransportMessageUrgentFrame(message=frame.text)) + else: + await self.push_frame(frame, direction) + + pipeline = Pipeline([DelayFrameProcessor(), InterruptFrameProcessor()]) + + frames_to_send = [ + # Just a random interruption to make sure we don't clear anything + # before the actual `InterruptionTaskFrame` interruption. + InterruptionFrame(), + # This will generate an `InterruptionTaskFrame` and will wait for an + # `InterruptionFrame`. + TextFrame(text="Hello from Pipecat!"), + # Just give time for everything to complete. + SleepFrame(sleep=0.5), + EndFrame(), + ] + expected_down_frames = [ + InterruptionFrame, + InterruptionFrame, + TransportMessageUrgentFrame, + EndFrame, + ] + await run_test( + pipeline, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + send_end_frame=False, + )