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

Merge pull request #2034 from pipecat-ai/khk/pause-frames

Browse Source 
small fix for processor pause/resume frames
This commit is contained in:
Aleix Conchillo Flaqué
2025-06-23 17:08:41 -07:00
committed by GitHub
parent f52ac6e99c 6b24f89fa7
commit 40496e7b0f
2 changed files with 22 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
src/pipecat/frames/frames.py
View File

@@ -7,6 +7,7 @@
from dataclasses import dataclass, field from dataclasses import dataclass, field
from enum import Enum from enum import Enum
from typing import ( from typing import (
TYPE_CHECKING,
Any, Any,
Awaitable, Awaitable,
Callable, Callable,
@@ -26,6 +27,9 @@ from pipecat.transcriptions.language import Language
from pipecat.utils.time import nanoseconds_to_str from pipecat.utils.time import nanoseconds_to_str
from pipecat.utils.utils import obj_count, obj_id from pipecat.utils.utils import obj_count, obj_id
if TYPE_CHECKING:
from pipecat.processors.frame_processor import FrameProcessor
class KeypadEntry(str, Enum): class KeypadEntry(str, Enum):
"""DTMF entries.""" """DTMF entries."""
@@ -529,25 +533,25 @@ class StopTaskFrame(SystemFrame):
@dataclass @dataclass
class FrameProcessorPauseUrgentFrame(SystemFrame): class FrameProcessorPauseUrgentFrame(SystemFrame):
"""This processor is used to pause frame processing for the given processor """This frame is used to pause frame processing for the given processor as
as fast as possible. Pausing frame processing will keep frames in the fast as possible. Pausing frame processing will keep frames in the internal
internal queue which will then be processed when frame processing is resumed queue which will then be processed when frame processing is resumed with
with `FrameProcessorResumeFrame`. `FrameProcessorResumeFrame`.
""" """
processor: str processor: "FrameProcessor"
@dataclass @dataclass
class FrameProcessorResumeUrgentFrame(SystemFrame): class FrameProcessorResumeUrgentFrame(SystemFrame):
"""This processor is used to resume frame processing for the given processor """This frame is used to resume frame processing for the given processor
if it was previously paused as fast as possible. After resuming frame if it was previously paused as fast as possible. After resuming frame
processing all queued frames will be processed in the order received. processing all queued frames will be processed in the order received.
""" """
processor: str processor: "FrameProcessor"
@dataclass @dataclass
@@ -879,23 +883,25 @@ class StopFrame(ControlFrame):
@dataclass @dataclass
class FrameProcessorPauseFrame(ControlFrame): class FrameProcessorPauseFrame(ControlFrame):
"""This processor is used to pause frame processing for the given """This frame is used to pause frame processing for the given
processor. Pausing frame processing will keep frames in the internal queue processor. Pausing frame processing will keep frames in the internal queue
which will then be processed when frame processing is resumed with which will then be processed when frame processing is resumed with
`FrameProcessorResumeFrame`.""" `FrameProcessorResumeFrame`.
processor: str """
processor: "FrameProcessor"
@dataclass @dataclass
class FrameProcessorResumeFrame(ControlFrame): class FrameProcessorResumeFrame(ControlFrame):
"""This processor is used to resume frame processing for the given processor """This frame is used to resume frame processing for the given processor if
if it was previously paused. After resuming frame processing all queued it was previously paused. After resuming frame processing all queued frames
frames will be processed in the order received. will be processed in the order received.
""" """
processor: str processor: "FrameProcessor"
@dataclass @dataclass

4
src/pipecat/processors/frame_processor.py
View File

@@ -296,11 +296,11 @@ class FrameProcessor(BaseObject):
await self.__cancel_push_task() await self.__cancel_push_task()
async def __pause(self, frame: FrameProcessorPauseFrame | FrameProcessorPauseUrgentFrame): async def __pause(self, frame: FrameProcessorPauseFrame | FrameProcessorPauseUrgentFrame):
if frame.name == self.name: if frame.processor.name == self.name:
await self.pause_processing_frames() await self.pause_processing_frames()
async def __resume(self, frame: FrameProcessorResumeFrame | FrameProcessorResumeUrgentFrame): async def __resume(self, frame: FrameProcessorResumeFrame | FrameProcessorResumeUrgentFrame):
if frame.name == self.name: if frame.processor.name == self.name:
await self.resume_processing_frames() await self.resume_processing_frames()
# #