wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix a bug preventing usage of multiple ServiceSwitchers in a pipeline

Browse Source
This commit is contained in:
Paul Kompfner
2025-09-17 16:09:18 -04:00
parent 2cd2567a37
commit 8bc3c89140
2 changed files with 6 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
src/pipecat/pipeline/service_switcher.py
View File

@@ -61,20 +61,21 @@ class ServiceSwitcherStrategyManual(ServiceSwitcherStrategy):
direction: The direction of the frame (upstream or downstream).
"""
if isinstance(frame, ManuallySwitchServiceFrame):
self._set_active(frame.service)
self._set_active_if_available(frame.service)
else:
raise ValueError(f"Unsupported frame type: {type(frame)}")
def _set_active(self, service: FrameProcessor):
"""Set the active service to the given one.
def _set_active_if_available(self, service: FrameProcessor):
"""Set the active service to the given one, if it is in the list of available services.
If it's not in the list, the request is ignored, as it may have been
intended for another ServiceSwitcher in the pipeline.
Args:
service: The service to set as active.
"""
if service in self.services:
self.active_service = service
else:
raise ValueError(f"Service {service} is not in the list of available services.")
StrategyType = TypeVar("StrategyType", bound=ServiceSwitcherStrategy)

13
tests/test_service_switcher.py
View File

@@ -100,19 +100,6 @@ class TestServiceSwitcherStrategyManual(unittest.IsolatedAsyncioTestCase):
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)
invalid_service = MockFrameProcessor("invalid")
switch_frame = ManuallySwitchServiceFrame(service=invalid_service)
with self.assertRaises(ValueError) as context:
strategy.handle_frame(switch_frame, FrameDirection.DOWNSTREAM)
self.assertIn("Service", str(context.exception))
self.assertIn("is not in the list of available services", str(context.exception))
def test_handle_frame_unsupported_frame_type(self):
"""Test that unsupported frame types raise an error."""
strategy = ServiceSwitcherStrategyManual(self.services)