Merge pull request #3181 from pipecat-ai/aleix/sync-to-utils-sync

move pipecat.sync to pipecat.utils.sync
This commit is contained in:
Aleix Conchillo Flaqué
2025-12-03 19:41:18 -08:00
committed by GitHub
15 changed files with 113 additions and 75 deletions

View File

@@ -40,6 +40,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Deprecated ### Deprecated
- Package `pipecat.sync` is deprecated, use `pipecat.utils.sync` instead.
- The `noise_gate_enable` parameter in `AICFilter` is deprecated and no longer - The `noise_gate_enable` parameter in `AICFilter` is deprecated and no longer
has any effect. Noise gating is now handled automatically by the AIC VAD has any effect. Noise gating is now handled automatically by the AIC VAD
system. Use `AICFilter.create_vad_analyzer()` for VAD functionality instead. system. Use `AICFilter.create_vad_analyzer()` for VAD functionality instead.

View File

@@ -119,7 +119,6 @@ def import_core_modules():
"pipecat.observers", "pipecat.observers",
"pipecat.runner", "pipecat.runner",
"pipecat.serializers", "pipecat.serializers",
"pipecat.sync",
"pipecat.transcriptions", "pipecat.transcriptions",
"pipecat.utils", "pipecat.utils",
] ]

View File

@@ -30,7 +30,6 @@ Quick Links
Runner <api/pipecat.runner> Runner <api/pipecat.runner>
Serializers <api/pipecat.serializers> Serializers <api/pipecat.serializers>
Services <api/pipecat.services> Services <api/pipecat.services>
Sync <api/pipecat.sync>
Transcriptions <api/pipecat.transcriptions> Transcriptions <api/pipecat.transcriptions>
Transports <api/pipecat.transports> Transports <api/pipecat.transports>
Utils <api/pipecat.utils> Utils <api/pipecat.utils>

View File

@@ -28,10 +28,10 @@ from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import LLMService from pipecat.services.llm_service import LLMService
from pipecat.services.openai.llm import OpenAIContextAggregatorPair, OpenAILLMService from pipecat.services.openai.llm import OpenAIContextAggregatorPair, OpenAILLMService
from pipecat.sync.event_notifier import EventNotifier
from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams from pipecat.transports.daily.transport import DailyParams
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
from pipecat.utils.sync.event_notifier import EventNotifier
load_dotenv(override=True) load_dotenv(override=True)

View File

@@ -45,11 +45,11 @@ from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams, LLMService from pipecat.services.llm_service import FunctionCallParams, LLMService
from pipecat.services.openai.llm import OpenAILLMService from pipecat.services.openai.llm import OpenAILLMService
from pipecat.sync.base_notifier import BaseNotifier
from pipecat.sync.event_notifier import EventNotifier
from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams from pipecat.transports.daily.transport import DailyParams
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
from pipecat.utils.sync.base_notifier import BaseNotifier
from pipecat.utils.sync.event_notifier import EventNotifier
from pipecat.utils.time import time_now_iso8601 from pipecat.utils.time import time_now_iso8601
load_dotenv(override=True) load_dotenv(override=True)

View File

@@ -46,11 +46,11 @@ from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams, LLMService from pipecat.services.llm_service import FunctionCallParams, LLMService
from pipecat.services.openai.llm import OpenAILLMService from pipecat.services.openai.llm import OpenAILLMService
from pipecat.sync.base_notifier import BaseNotifier
from pipecat.sync.event_notifier import EventNotifier
from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams from pipecat.transports.daily.transport import DailyParams
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
from pipecat.utils.sync.base_notifier import BaseNotifier
from pipecat.utils.sync.event_notifier import EventNotifier
from pipecat.utils.time import time_now_iso8601 from pipecat.utils.time import time_now_iso8601
load_dotenv(override=True) load_dotenv(override=True)

View File

@@ -47,11 +47,11 @@ from pipecat.runner.utils import create_transport
from pipecat.services.cartesia.tts import CartesiaTTSService from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.google.llm import GoogleLLMService from pipecat.services.google.llm import GoogleLLMService
from pipecat.services.llm_service import LLMService from pipecat.services.llm_service import LLMService
from pipecat.sync.base_notifier import BaseNotifier
from pipecat.sync.event_notifier import EventNotifier
from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams from pipecat.transports.daily.transport import DailyParams
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
from pipecat.utils.sync.base_notifier import BaseNotifier
from pipecat.utils.sync.event_notifier import EventNotifier
from pipecat.utils.time import time_now_iso8601 from pipecat.utils.time import time_now_iso8601
load_dotenv(override=True) load_dotenv(override=True)

View File

@@ -40,8 +40,8 @@ from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup
from pipecat.services.llm_service import LLMService from pipecat.services.llm_service import LLMService
from pipecat.sync.base_notifier import BaseNotifier from pipecat.utils.sync.base_notifier import BaseNotifier
from pipecat.sync.event_notifier import EventNotifier from pipecat.utils.sync.event_notifier import EventNotifier
class NotifierGate(FrameProcessor): class NotifierGate(FrameProcessor):

View File

@@ -9,7 +9,7 @@
from pipecat.frames.frames import CancelFrame, EndFrame, Frame, LLMContextFrame, StartFrame from pipecat.frames.frames import CancelFrame, EndFrame, Frame, LLMContextFrame, StartFrame
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.sync.base_notifier import BaseNotifier from pipecat.utils.sync.base_notifier import BaseNotifier
class GatedLLMContextAggregator(FrameProcessor): class GatedLLMContextAggregator(FrameProcessor):

View File

@@ -10,7 +10,7 @@ from typing import Awaitable, Callable, Tuple, Type
from pipecat.frames.frames import Frame from pipecat.frames.frames import Frame
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.sync.base_notifier import BaseNotifier from pipecat.utils.sync.base_notifier import BaseNotifier
class WakeNotifierFilter(FrameProcessor): class WakeNotifierFilter(FrameProcessor):

View File

@@ -6,31 +6,14 @@
"""Base notifier interface for Pipecat.""" """Base notifier interface for Pipecat."""
from abc import ABC, abstractmethod import warnings
from pipecat.utils.sync.base_notifier import BaseNotifier
class BaseNotifier(ABC): with warnings.catch_warnings():
"""Abstract base class for notification mechanisms. warnings.simplefilter("always")
warnings.warn(
Provides a standard interface for implementing notification and waiting "Package pipecat.sync is deprecated, use pipecat.utils.sync instead.",
patterns used for event coordination and signaling between components DeprecationWarning,
in the Pipecat framework. stacklevel=2,
""" )
@abstractmethod
async def notify(self):
"""Send a notification signal.
Implementations should trigger any waiting coroutines or processes
that are blocked on this notifier.
"""
pass
@abstractmethod
async def wait(self):
"""Wait for a notification signal.
Implementations should block until a notification is received
from the corresponding notify() call.
"""
pass

View File

@@ -6,40 +6,14 @@
"""Event-based notifier implementation using asyncio Event primitives.""" """Event-based notifier implementation using asyncio Event primitives."""
import asyncio import warnings
from pipecat.sync.base_notifier import BaseNotifier from pipecat.utils.sync.event_notifier import EventNotifier
with warnings.catch_warnings():
class EventNotifier(BaseNotifier): warnings.simplefilter("always")
"""Event-based notifier using asyncio.Event for task synchronization. warnings.warn(
"Package pipecat.sync is deprecated, use pipecat.utils.sync instead.",
Provides a simple notification mechanism where one task can signal DeprecationWarning,
an event and other tasks can wait for that event to occur. The event stacklevel=2,
is automatically cleared after each wait operation. )
"""
def __init__(self):
"""Initialize the event notifier.
Creates an internal asyncio.Event for managing notifications.
"""
self._event = asyncio.Event()
async def notify(self):
"""Signal the event to notify waiting tasks.
Sets the internal event, causing any tasks waiting on this
notifier to be awakened.
"""
self._event.set()
async def wait(self):
"""Wait for the event to be signaled.
Blocks until another task calls notify(). Automatically clears
the event after being awakened so subsequent calls will wait
for the next notification.
"""
await self._event.wait()
self._event.clear()

View File

View File

@@ -0,0 +1,36 @@
#
# Copyright (c) 20242025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Base notifier interface for Pipecat."""
from abc import ABC, abstractmethod
class BaseNotifier(ABC):
"""Abstract base class for notification mechanisms.
Provides a standard interface for implementing notification and waiting
patterns used for event coordination and signaling between components
in the Pipecat framework.
"""
@abstractmethod
async def notify(self):
"""Send a notification signal.
Implementations should trigger any waiting coroutines or processes
that are blocked on this notifier.
"""
pass
@abstractmethod
async def wait(self):
"""Wait for a notification signal.
Implementations should block until a notification is received
from the corresponding notify() call.
"""
pass

View File

@@ -0,0 +1,45 @@
#
# Copyright (c) 20242025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Event-based notifier implementation using asyncio Event primitives."""
import asyncio
from pipecat.utils.sync.base_notifier import BaseNotifier
class EventNotifier(BaseNotifier):
"""Event-based notifier using asyncio.Event for task synchronization.
Provides a simple notification mechanism where one task can signal
an event and other tasks can wait for that event to occur. The event
is automatically cleared after each wait operation.
"""
def __init__(self):
"""Initialize the event notifier.
Creates an internal asyncio.Event for managing notifications.
"""
self._event = asyncio.Event()
async def notify(self):
"""Signal the event to notify waiting tasks.
Sets the internal event, causing any tasks waiting on this
notifier to be awakened.
"""
self._event.set()
async def wait(self):
"""Wait for the event to be signaled.
Blocks until another task calls notify(). Automatically clears
the event after being awakened so subsequent calls will wait
for the next notification.
"""
await self._event.wait()
self._event.clear()