diff --git a/src/pipecat/pipeline/pipeline.py b/src/pipecat/pipeline/pipeline.py index c13cc61a1..165717ad3 100644 --- a/src/pipecat/pipeline/pipeline.py +++ b/src/pipecat/pipeline/pipeline.py @@ -91,5 +91,7 @@ class Pipeline(BasePipeline): def _link_processors(self): prev = self._processors[0] for curr in self._processors[1:]: + prev.set_parent(self) prev.link(curr) prev = curr + prev.set_parent(self) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 18c03f169..405936e06 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -72,6 +72,7 @@ class FrameProcessor: **kwargs): self.id: int = obj_id() self.name = name or f"{self.__class__.__name__}#{obj_count(self)}" + self._parent: "FrameProcessor" | None = None self._prev: "FrameProcessor" | None = None self._next: "FrameProcessor" | None = None self._loop: asyncio.AbstractEventLoop = loop or asyncio.get_running_loop() @@ -126,7 +127,7 @@ class FrameProcessor: async def cleanup(self): pass - def link(self, processor: 'FrameProcessor'): + def link(self, processor: "FrameProcessor"): self._next = processor processor._prev = self logger.debug(f"Linking {self} -> {self._next}") @@ -134,6 +135,12 @@ class FrameProcessor: def get_event_loop(self) -> asyncio.AbstractEventLoop: return self._loop + def set_parent(self, parent: "FrameProcessor"): + self._parent = parent + + def get_parent(self) -> "FrameProcessor": + return self._parent + async def process_frame(self, frame: Frame, direction: FrameDirection): if isinstance(frame, StartFrame): self._allow_interruptions = frame.allow_interruptions