From eacd2a4b71071dbd9108afa838ba637b0c39c350 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 21 Jan 2026 14:10:56 -0800 Subject: [PATCH] FrameProcessor: add broadcast_frame_instance() --- src/pipecat/processors/frame_processor.py | 34 +++++ src/pipecat/serializers/protobuf.py | 2 +- tests/test_frame_processor.py | 166 +++++++++++++++++++++- 3 files changed, 200 insertions(+), 2 deletions(-) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 70f44dfca..10aafc5a8 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -12,7 +12,9 @@ management, and frame flow control mechanisms. """ import asyncio +import dataclasses import traceback +from copy import deepcopy from dataclasses import dataclass from enum import Enum from typing import ( @@ -782,6 +784,38 @@ class FrameProcessor(BaseObject): await self.push_frame(frame_cls(**kwargs)) await self.push_frame(frame_cls(**kwargs), FrameDirection.UPSTREAM) + async def broadcast_frame_instance(self, frame: Frame): + """Broadcasts a frame instance upstream and downstream. + + This method extracts the class and init fields from the given frame + instance and creates two new instances to push upstream and downstream. + + Args: + frame: The frame instance to broadcast. + + Note: + Prefer using `broadcast_frame()` when possible, as it is more + efficient. This method should only be used when you are not the + creator of the frame and need to broadcast an existing instance. + """ + frame_cls = type(frame) + init_fields = {f.name: getattr(frame, f.name) for f in dataclasses.fields(frame) if f.init} + extra_fields = { + f.name: getattr(frame, f.name) + for f in dataclasses.fields(frame) + if not f.init and f.name not in ("id", "name") + } + + new_frame = frame_cls(**deepcopy(init_fields)) + for k, v in deepcopy(extra_fields).items(): + setattr(new_frame, k, v) + await self.push_frame(new_frame) + + new_frame = frame_cls(**deepcopy(init_fields)) + for k, v in deepcopy(extra_fields).items(): + setattr(new_frame, k, v) + await self.push_frame(new_frame, FrameDirection.UPSTREAM) + async def __start(self, frame: StartFrame): """Handle the start frame to initialize processor state. diff --git a/src/pipecat/serializers/protobuf.py b/src/pipecat/serializers/protobuf.py index 6d989c7dd..912f20b42 100644 --- a/src/pipecat/serializers/protobuf.py +++ b/src/pipecat/serializers/protobuf.py @@ -126,7 +126,7 @@ class ProtobufFrameSerializer(FrameSerializer): if "pts" in args_dict: del args_dict["pts"] - # Special handling for MessageFrame -> OutputTransportMessageUrgentFrame + # Special handling for MessageFrame -> InputTransportMessageFrame if class_name == MessageFrame: try: msg = json.loads(args_dict["data"]) diff --git a/tests/test_frame_processor.py b/tests/test_frame_processor.py index b3a11ab43..3f8c8ae70 100644 --- a/tests/test_frame_processor.py +++ b/tests/test_frame_processor.py @@ -6,7 +6,8 @@ import asyncio import unittest -from dataclasses import dataclass +from dataclasses import dataclass, field +from typing import List from pipecat.frames.frames import ( DataFrame, @@ -24,6 +25,15 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.tests.utils import SleepFrame, run_test +@dataclass +class BroadcastTestFrame(DataFrame): + """Test frame with init fields for broadcast testing.""" + + text: str = "" + value: int = 0 + items: List[str] = field(default_factory=list) + + class TestFrameProcessor(unittest.IsolatedAsyncioTestCase): async def test_before_after_events(self): identity = IdentityFilter() @@ -186,3 +196,157 @@ class TestFrameProcessor(unittest.IsolatedAsyncioTestCase): frames_to_send=frames_to_send, expected_down_frames=expected_down_frames, ) + + async def test_broadcast_frame(self): + """Test that broadcast_frame creates two separate frames with fresh IDs.""" + downstream_frames: List[Frame] = [] + upstream_frames: List[Frame] = [] + + class BroadcastTestProcessor(FrameProcessor): + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, TextFrame): + await self.broadcast_frame( + BroadcastTestFrame, text="hello", value=42, items=["a", "b"] + ) + else: + await self.push_frame(frame, direction) + + class CaptureProcessor(FrameProcessor): + def __init__(self, capture_list: List[Frame], direction: FrameDirection): + super().__init__() + self._capture_list = capture_list + self._capture_direction = direction + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if direction == self._capture_direction and isinstance(frame, BroadcastTestFrame): + self._capture_list.append(frame) + await self.push_frame(frame, direction) + + up_capture = CaptureProcessor(upstream_frames, FrameDirection.UPSTREAM) + broadcaster = BroadcastTestProcessor() + down_capture = CaptureProcessor(downstream_frames, FrameDirection.DOWNSTREAM) + + pipeline = Pipeline([up_capture, broadcaster, down_capture]) + + frames_to_send = [TextFrame(text="trigger")] + expected_down_frames = [BroadcastTestFrame] + expected_up_frames = [BroadcastTestFrame] + + await run_test( + pipeline, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + expected_up_frames=expected_up_frames, + ) + + # Verify we got one frame in each direction + self.assertEqual(len(downstream_frames), 1) + self.assertEqual(len(upstream_frames), 1) + + down_frame = downstream_frames[0] + up_frame = upstream_frames[0] + + # Verify the frames have different IDs (they are separate instances) + self.assertNotEqual(down_frame.id, up_frame.id) + + # Verify the frames have the correct field values + self.assertEqual(down_frame.text, "hello") + self.assertEqual(down_frame.value, 42) + self.assertEqual(down_frame.items, ["a", "b"]) + self.assertEqual(up_frame.text, "hello") + self.assertEqual(up_frame.value, 42) + self.assertEqual(up_frame.items, ["a", "b"]) + + # Verify the items lists are separate instances (not shared references) + self.assertIsNot(down_frame.items, up_frame.items) + + async def test_broadcast_frame_instance(self): + """Test that broadcast_frame_instance copies all fields except id and name.""" + downstream_frames: List[Frame] = [] + upstream_frames: List[Frame] = [] + original_frame: List[Frame] = [] + + class BroadcastInstanceTestProcessor(FrameProcessor): + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, BroadcastTestFrame): + # Set some non-init fields on the frame + frame.pts = 12345 + frame.metadata = {"key": "value", "nested": {"a": 1}} + original_frame.append(frame) + await self.broadcast_frame_instance(frame) + else: + await self.push_frame(frame, direction) + + class CaptureProcessor(FrameProcessor): + def __init__(self, capture_list: List[Frame], direction: FrameDirection): + super().__init__() + self._capture_list = capture_list + self._capture_direction = direction + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if direction == self._capture_direction and isinstance(frame, BroadcastTestFrame): + self._capture_list.append(frame) + await self.push_frame(frame, direction) + + up_capture = CaptureProcessor(upstream_frames, FrameDirection.UPSTREAM) + broadcaster = BroadcastInstanceTestProcessor() + down_capture = CaptureProcessor(downstream_frames, FrameDirection.DOWNSTREAM) + + pipeline = Pipeline([up_capture, broadcaster, down_capture]) + + # Create a frame with mutable fields to test deep copying + test_frame = BroadcastTestFrame(text="test", value=99, items=["x", "y", "z"]) + + frames_to_send = [test_frame] + expected_down_frames = [BroadcastTestFrame] + expected_up_frames = [BroadcastTestFrame] + + await run_test( + pipeline, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + expected_up_frames=expected_up_frames, + ) + + # Verify we got one frame in each direction + self.assertEqual(len(downstream_frames), 1) + self.assertEqual(len(upstream_frames), 1) + self.assertEqual(len(original_frame), 1) + + orig = original_frame[0] + down_frame = downstream_frames[0] + up_frame = upstream_frames[0] + + # Verify the frames have different IDs and names (fresh values) + self.assertNotEqual(down_frame.id, orig.id) + self.assertNotEqual(up_frame.id, orig.id) + self.assertNotEqual(down_frame.id, up_frame.id) + self.assertNotEqual(down_frame.name, orig.name) + self.assertNotEqual(up_frame.name, orig.name) + + # Verify init fields are copied correctly + self.assertEqual(down_frame.text, "test") + self.assertEqual(down_frame.value, 99) + self.assertEqual(down_frame.items, ["x", "y", "z"]) + self.assertEqual(up_frame.text, "test") + self.assertEqual(up_frame.value, 99) + self.assertEqual(up_frame.items, ["x", "y", "z"]) + + # Verify non-init fields (except id/name) are copied + self.assertEqual(down_frame.pts, 12345) + self.assertEqual(down_frame.metadata, {"key": "value", "nested": {"a": 1}}) + self.assertEqual(up_frame.pts, 12345) + self.assertEqual(up_frame.metadata, {"key": "value", "nested": {"a": 1}}) + + # Verify mutable fields are deep copied (not shared references) + self.assertIsNot(down_frame.items, orig.items) + self.assertIsNot(up_frame.items, orig.items) + self.assertIsNot(down_frame.items, up_frame.items) + self.assertIsNot(down_frame.metadata, orig.metadata) + self.assertIsNot(up_frame.metadata, orig.metadata) + self.assertIsNot(down_frame.metadata, up_frame.metadata) + self.assertIsNot(down_frame.metadata["nested"], up_frame.metadata["nested"])