diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index a9b58d986..ffef7c99c 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -1603,46 +1603,17 @@ class MixerEnableFrame(MixerControlFrame): @dataclass -class ManuallySwitchServiceBaseFrame: - """Base class for frames that request a manual service switch from a ServiceSwitcher. +class ServiceSwitcherFrame(ControlFrame): + """A base class for frames that affect ServiceSwitcher behavior.""" - Must be inherited alongside either ServiceSwitcherFrame or ServiceSwitcherControlFrame. + pass + + +@dataclass +class ManuallySwitchServiceFrame(ServiceSwitcherFrame): + """A frame to request a manual switch in the active service in a ServiceSwitcher. + + Handled by ServiceSwitcherStrategyManual to switch the active service. """ service: "FrameProcessor" - - -@dataclass -class ServiceSwitcherFrame(SystemFrame): - """A base class for frames that affect ServiceSwitcher behavior immediately.""" - - pass - - -@dataclass -class ManuallySwitchServiceFrame(ServiceSwitcherFrame, ManuallySwitchServiceBaseFrame): - """A frame to request an immediate manual switch in the active service in a ServiceSwitcher. - - Handled by ServiceSwitcherStrategyManual to switch the active service. - """ - - pass - - -@dataclass -class ServiceSwitcherControlFrame(ControlFrame): - """A base class for frames that affect ServiceSwitcher behavior, in order.""" - - pass - - -@dataclass -class ManuallySwitchServiceControlFrame( - ServiceSwitcherControlFrame, ManuallySwitchServiceBaseFrame -): - """A frame to request a manual switch in the active service in a ServiceSwitcher, in order. - - Handled by ServiceSwitcherStrategyManual to switch the active service. - """ - - pass diff --git a/src/pipecat/pipeline/service_switcher.py b/src/pipecat/pipeline/service_switcher.py index 41efcb974..fe99751e4 100644 --- a/src/pipecat/pipeline/service_switcher.py +++ b/src/pipecat/pipeline/service_switcher.py @@ -12,10 +12,8 @@ from typing import Any, Generic, List, Optional, Type, TypeVar from pipecat.frames.frames import ( ControlFrame, Frame, - ManuallySwitchServiceBaseFrame, - ServiceSwitcherControlFrame, + ManuallySwitchServiceFrame, ServiceSwitcherFrame, - SystemFrame, ) from pipecat.pipeline.parallel_pipeline import ParallelPipeline from pipecat.processors.filters.function_filter import FunctionFilter @@ -30,9 +28,7 @@ class ServiceSwitcherStrategy: self.services = services self.active_service: Optional[FrameProcessor] = None - def handle_frame( - self, frame: ServiceSwitcherFrame | ServiceSwitcherControlFrame, direction: FrameDirection - ): + def handle_frame(self, frame: ServiceSwitcherFrame, direction: FrameDirection): """Handle a frame that controls service switching. This method can be overridden by subclasses to implement specific logic @@ -57,16 +53,14 @@ class ServiceSwitcherStrategyManual(ServiceSwitcherStrategy): super().__init__(services) self.active_service = services[0] if services else None - def handle_frame( - self, frame: ServiceSwitcherFrame | ServiceSwitcherControlFrame, direction: FrameDirection - ): + def handle_frame(self, frame: ServiceSwitcherFrame, direction: FrameDirection): """Handle a frame that controls service switching. Args: frame: The frame to handle. direction: The direction of the frame (upstream or downstream). """ - if isinstance(frame, ManuallySwitchServiceBaseFrame): + if isinstance(frame, ManuallySwitchServiceFrame): self._set_active(frame.service) else: raise ValueError(f"Unsupported frame type: {type(frame)}") @@ -116,7 +110,7 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]): async def process_frame(self, frame, direction): """Process a frame through the filter, handling special internal filter-updating frames.""" - if isinstance(frame, ServiceSwitcher.ServiceSwitcherFilterBaseFrame): + if isinstance(frame, ServiceSwitcher.ServiceSwitcherFilterFrame): self.active_service = frame.active_service # Two ServiceSwitcherFilters "sandwich" a service. Push the # frame only to update the other side of the sandwich, but @@ -128,22 +122,10 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]): await super().process_frame(frame, direction) @dataclass - class ServiceSwitcherFilterBaseFrame: - """Base class for internal frames used by ServiceSwitcher to filter frames based on active service.""" - - active_service: FrameProcessor - - @dataclass - class ServiceSwitcherFilterFrame(SystemFrame, ServiceSwitcherFilterBaseFrame): + class ServiceSwitcherFilterFrame(ControlFrame): """An internal frame used by ServiceSwitcher to filter frames based on active service.""" - pass - - @dataclass - class ServiceSwitcherFilterControlFrame(ControlFrame, ServiceSwitcherFilterBaseFrame): - """An internal control frame used by ServiceSwitcher to filter frames based on active service.""" - - pass + active_service: FrameProcessor @staticmethod def _make_pipeline_definitions( @@ -181,19 +163,11 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]): """ await super().process_frame(frame, direction) - if isinstance(frame, (ServiceSwitcherFrame, ServiceSwitcherControlFrame)): + if isinstance(frame, ServiceSwitcherFrame): self.strategy.handle_frame(frame, direction) - if isinstance(frame, ServiceSwitcherFrame): - # Apply immediate update (system frame) - service_switcher_filter_frame = ServiceSwitcher.ServiceSwitcherFilterFrame( - active_service=self.strategy.active_service - ) - else: - # Apply in-order update (control frame) - service_switcher_filter_frame = ServiceSwitcher.ServiceSwitcherFilterControlFrame( - active_service=self.strategy.active_service - ) - # Queue frame that updates filters with new active service - # (Hack: we need access ParallelPipeline internals here to queue the frame in each of the pipelines) + service_switcher_filter_frame = ServiceSwitcher.ServiceSwitcherFilterFrame( + active_service=self.strategy.active_service + ) + # Hack: we need access ParallelPipeline internals here to queue the frame in each of the pipelines for p in self._pipelines: await p.queue_frame(service_switcher_filter_frame, direction) diff --git a/tests/test_service_switcher.py b/tests/test_service_switcher.py index b8a508c82..26f6614a4 100644 --- a/tests/test_service_switcher.py +++ b/tests/test_service_switcher.py @@ -6,12 +6,10 @@ """Unit tests for ServiceSwitcher and related components.""" -import asyncio import unittest from pipecat.frames.frames import ( Frame, - ManuallySwitchServiceControlFrame, ManuallySwitchServiceFrame, TextFrame, ) @@ -101,30 +99,6 @@ class TestServiceSwitcherStrategyManual(unittest.IsolatedAsyncioTestCase): self.assertNotEqual(strategy.active_service, self.service2) self.assertEqual(strategy.active_service, self.service3) - def test_handle_manually_switch_service_control_frame(self): - """Test manual service switching with ManuallySwitchServiceControlFrame.""" - strategy = ServiceSwitcherStrategyManual(self.services) - - # Initially service1 should be active - self.assertEqual(strategy.active_service, self.service1) - self.assertNotEqual(strategy.active_service, self.service2) - - # Switch to service2 - switch_frame = ManuallySwitchServiceControlFrame(service=self.service2) - strategy.handle_frame(switch_frame, FrameDirection.DOWNSTREAM) - - self.assertNotEqual(strategy.active_service, self.service1) - self.assertEqual(strategy.active_service, self.service2) - self.assertNotEqual(strategy.active_service, self.service3) - - # Switch to service3 - switch_frame = ManuallySwitchServiceControlFrame(service=self.service3) - strategy.handle_frame(switch_frame, FrameDirection.DOWNSTREAM) - - self.assertNotEqual(strategy.active_service, self.service1) - self.assertNotEqual(strategy.active_service, self.service2) - self.assertEqual(strategy.active_service, self.service3) - def test_handle_frame_invalid_service(self): """Test that switching to an invalid service raises an error.""" strategy = ServiceSwitcherStrategyManual(self.services) @@ -207,8 +181,8 @@ class TestServiceSwitcher(unittest.IsolatedAsyncioTestCase): for i, frame in enumerate(text_frames): self.assertEqual(frame.text, f"Hello {i + 1}") - async def test_service_switching_in_order(self): - """Test that after service switching using ManuallySwitchServiceControlFrame, the new active service receives frames while others don't.""" + async def test_service_switching(self): + """Test that after service switching using ManuallySwitchServiceFrame, the new active service receives frames while others don't.""" switcher = ServiceSwitcher(self.services, ServiceSwitcherStrategyManual) # Reset counters @@ -219,11 +193,11 @@ class TestServiceSwitcher(unittest.IsolatedAsyncioTestCase): await run_test( switcher, frames_to_send=[ - TextFrame("Frame for service1"), - ManuallySwitchServiceControlFrame(service=self.service2), - TextFrame("Frame for service2"), + TextFrame("Hello 1"), + ManuallySwitchServiceFrame(service=self.service2), + TextFrame("Hello 2"), ], - expected_down_frames=[TextFrame, ManuallySwitchServiceControlFrame, TextFrame], + expected_down_frames=[TextFrame, ManuallySwitchServiceFrame, TextFrame], ) # Verify service2 received the frame @@ -240,45 +214,9 @@ class TestServiceSwitcher(unittest.IsolatedAsyncioTestCase): self.assertEqual(len(service1_text_frames), 1) self.assertEqual(len(service2_text_frames), 1) self.assertEqual(len(service3_text_frames), 0) - self.assertEqual(service1_text_frames[0].text, "Frame for service1") - self.assertEqual(service2_text_frames[0].text, "Frame for service2") - async def test_service_switching_immediate(self): - """Test that after service switching using ManuallySwitchServiceFrame, the new active service receives frames while others don't.""" - switcher = ServiceSwitcher(self.services, ServiceSwitcherStrategyManual) - - # Reset counters - for service in self.services: - service.reset_counters() - - # Send a test frame, a switch frame, and another test frame - await run_test( - switcher, - frames_to_send=[ - # Out of order on purpose - ManuallySwitchServiceFrame should jump the queue - TextFrame("Frame 1 for service2"), - ManuallySwitchServiceFrame(service=self.service2), - TextFrame("Frame 2 for service2"), - ], - expected_down_frames=[ManuallySwitchServiceFrame, TextFrame, TextFrame], - ) - - # Verify service2 received the frame - service1_text_frames = [ - f for f in self.service1.processed_frames if isinstance(f, TextFrame) - ] - service2_text_frames = [ - f for f in self.service2.processed_frames if isinstance(f, TextFrame) - ] - service3_text_frames = [ - f for f in self.service3.processed_frames if isinstance(f, TextFrame) - ] - - self.assertEqual(len(service1_text_frames), 0) - self.assertEqual(len(service2_text_frames), 2) - self.assertEqual(len(service3_text_frames), 0) - self.assertEqual(service2_text_frames[0].text, "Frame 1 for service2") - self.assertEqual(service2_text_frames[1].text, "Frame 2 for service2") + self.assertEqual(service1_text_frames[0].text, "Hello 1") + self.assertEqual(service2_text_frames[0].text, "Hello 2") if __name__ == "__main__":