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

make all transports read/write functions async

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2024-05-30 16:28:17 -07:00
parent 219af71227
commit 7a6136fa9d
12 changed files with 195 additions and 282 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
examples/websocket-server/server.py
View File

@@ -31,7 +31,12 @@ logger.add(sys.stderr, level="DEBUG")
async def main(): async def main():
async with aiohttp.ClientSession() as session: async with aiohttp.ClientSession() as session:
transport = WebsocketServerTransport(params=WebsocketServerParams(add_wav_header=True)) transport = WebsocketServerTransport(
params=WebsocketServerParams(
audio_out_enabled=True,
add_wav_header=True
)
)
vad = SileroVAD(audio_passthrough=True) vad = SileroVAD(audio_passthrough=True)

5
src/pipecat/pipeline/pipeline.py
View File

@@ -4,8 +4,6 @@
# SPDX-License-Identifier: BSD 2-Clause License # SPDX-License-Identifier: BSD 2-Clause License
# #
import asyncio
from typing import Callable, Coroutine, List from typing import Callable, Coroutine, List
from pipecat.frames.frames import Frame from pipecat.frames.frames import Frame
@@ -67,7 +65,8 @@ class Pipeline(FrameProcessor):
await self._sink.process_frame(frame, FrameDirection.UPSTREAM) await self._sink.process_frame(frame, FrameDirection.UPSTREAM)
async def _cleanup_processors(self): async def _cleanup_processors(self):
await asyncio.gather(*[p.cleanup() for p in self._processors]) for p in self._processors:
await p.cleanup()
def _link_processors(self): def _link_processors(self):
prev = self._processors[0] prev = self._processors[0]

2
src/pipecat/pipeline/runner.py
View File

@@ -46,7 +46,7 @@ class PipelineRunner:
return self._running return self._running
def _setup_sigint(self): def _setup_sigint(self):
loop = asyncio.get_event_loop() loop = asyncio.get_running_loop()
loop.add_signal_handler( loop.add_signal_handler(
signal.SIGINT, signal.SIGINT,
lambda *args: asyncio.create_task(self._sigint_handler()) lambda *args: asyncio.create_task(self._sigint_handler())

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

@@ -5,7 +5,7 @@
# #
import asyncio import asyncio
from asyncio import AbstractEventLoop
from enum import Enum from enum import Enum
from pipecat.frames.frames import ErrorFrame, Frame from pipecat.frames.frames import ErrorFrame, Frame
@@ -21,12 +21,12 @@ class FrameDirection(Enum):
class FrameProcessor: class FrameProcessor:
def __init__(self): def __init__(self, loop: asyncio.AbstractEventLoop | None = None):
self.id: int = obj_id() self.id: int = obj_id()
self.name = f"{self.__class__.__name__}#{obj_count(self)}" self.name = f"{self.__class__.__name__}#{obj_count(self)}"
self._prev: "FrameProcessor" | None = None self._prev: "FrameProcessor" | None = None
self._next: "FrameProcessor" | None = None self._next: "FrameProcessor" | None = None
self._loop: AbstractEventLoop = asyncio.get_event_loop() self._loop: asyncio.AbstractEventLoop = loop or asyncio.get_running_loop()
async def cleanup(self): async def cleanup(self):
pass pass
@@ -36,7 +36,7 @@ class FrameProcessor:
processor._prev = self processor._prev = self
logger.debug(f"Linking {self} -> {self._next}") logger.debug(f"Linking {self} -> {self._next}")
def get_event_loop(self) -> AbstractEventLoop: def get_event_loop(self) -> asyncio.AbstractEventLoop:
return self._loop return self._loop
async def process_frame(self, frame: Frame, direction: FrameDirection): async def process_frame(self, frame: Frame, direction: FrameDirection):

94
src/pipecat/transports/base_input.py
View File

@@ -5,9 +5,6 @@
# #
import asyncio import asyncio
import queue
from concurrent.futures import ThreadPoolExecutor
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.frames.frames import ( from pipecat.frames.frames import (
@@ -33,14 +30,11 @@ class BaseInputTransport(FrameProcessor):
self._params = params self._params = params
self._running = False
self._allow_interruptions = False self._allow_interruptions = False
self._in_executor = ThreadPoolExecutor(max_workers=5)
# Create audio input queue if needed. # Create audio input queue if needed.
if self._params.audio_in_enabled or self._params.vad_enabled: if self._params.audio_in_enabled or self._params.vad_enabled:
self._audio_in_queue = queue.Queue() self._audio_in_queue = asyncio.Queue()
# Create push frame task. This is the task that will push frames in # Create push frame task. This is the task that will push frames in
# order. We also guarantee that all frames are pushed in the same task. # order. We also guarantee that all frames are pushed in the same task.
@@ -52,45 +46,27 @@ class BaseInputTransport(FrameProcessor):
# for example. # for example.
self._allow_interruptions = frame.allow_interruptions self._allow_interruptions = frame.allow_interruptions
if self._running:
return
self._running = True
if self._params.audio_in_enabled or self._params.vad_enabled: if self._params.audio_in_enabled or self._params.vad_enabled:
loop = self.get_event_loop() loop = self.get_event_loop()
self._audio_in_thread = loop.run_in_executor( self._audio_task = loop.create_task(self._audio_task_handler())
self._in_executor, self._audio_in_thread_handler)
self._audio_out_thread = loop.run_in_executor(
self._in_executor, self._audio_out_thread_handler)
async def stop(self): async def stop(self):
if not self._running: # Wait for the tasks to finish.
return
# This will exit all threads.
self._running = False
# Wait for the threads to finish.
if self._params.audio_in_enabled or self._params.vad_enabled: if self._params.audio_in_enabled or self._params.vad_enabled:
await self._audio_in_thread self._audio_task.cancel()
await self._audio_out_thread
self._push_frame_task.cancel() self._push_frame_task.cancel()
def vad_analyze(self, audio_frames: bytes) -> VADState: async def vad_analyze(self, audio_frames: bytes) -> VADState:
pass pass
def read_raw_audio_frames(self, frame_count: int) -> bytes: async def read_raw_audio_frames(self, frame_count: int) -> bytes:
pass pass
# #
# Frame processor # Frame processor
# #
async def cleanup(self):
pass
async def process_frame(self, frame: Frame, direction: FrameDirection): async def process_frame(self, frame: Frame, direction: FrameDirection):
if isinstance(frame, CancelFrame): if isinstance(frame, CancelFrame):
await self.stop() await self.stop()
@@ -150,8 +126,8 @@ class BaseInputTransport(FrameProcessor):
# Audio input # Audio input
# #
def _handle_vad(self, audio_frames: bytes, vad_state: VADState): async def _handle_vad(self, audio_frames: bytes, vad_state: VADState):
new_vad_state = self.vad_analyze(audio_frames) new_vad_state = await self.vad_analyze(audio_frames)
if new_vad_state != vad_state and new_vad_state != VADState.STARTING and new_vad_state != VADState.STOPPING: if new_vad_state != vad_state and new_vad_state != VADState.STARTING and new_vad_state != VADState.STOPPING:
frame = None frame = None
if new_vad_state == VADState.SPEAKING: if new_vad_state == VADState.SPEAKING:
@@ -160,51 +136,39 @@ class BaseInputTransport(FrameProcessor):
frame = UserStoppedSpeakingFrame() frame = UserStoppedSpeakingFrame()
if frame: if frame:
future = asyncio.run_coroutine_threadsafe( await self._handle_interruptions(frame)
self._handle_interruptions(frame), self.get_event_loop())
future.result()
vad_state = new_vad_state vad_state = new_vad_state
return vad_state return vad_state
def _audio_in_thread_handler(self): async def _audio_task_handler(self):
vad_state: VADState = VADState.QUIET
sample_rate = self._params.audio_in_sample_rate sample_rate = self._params.audio_in_sample_rate
num_channels = self._params.audio_in_channels num_channels = self._params.audio_in_channels
num_frames = int(sample_rate / 100) # 10ms of audio num_frames = int(sample_rate / 100) # 10ms of audio
while self._running:
while True:
try: try:
audio_frames = self.read_raw_audio_frames(num_frames) audio_frames = await self.read_raw_audio_frames(num_frames)
if len(audio_frames) > 0: if len(audio_frames) > 0:
frame = AudioRawFrame( frame = AudioRawFrame(
audio=audio_frames, audio=audio_frames,
sample_rate=sample_rate, sample_rate=sample_rate,
num_channels=num_channels) num_channels=num_channels)
self._audio_in_queue.put(frame)
audio_passthrough = True
# Check VAD and push event if necessary. We just care about
# changes from QUIET to SPEAKING and vice versa.
if self._params.vad_enabled:
vad_state = await self._handle_vad(frame.audio, vad_state)
audio_passthrough = self._params.vad_audio_passthrough
# Push audio downstream if passthrough.
if audio_passthrough:
await self._internal_push_frame(frame)
except asyncio.CancelledError:
break
except BaseException as e: except BaseException as e:
logger.error(f"Error reading audio frames: {e}") logger.error(f"Error reading audio frames: {e}")
def _audio_out_thread_handler(self):
vad_state: VADState = VADState.QUIET
while self._running:
try:
frame = self._audio_in_queue.get(timeout=1)
audio_passthrough = True
# Check VAD and push event if necessary. We just care about changes
# from QUIET to SPEAKING and vice versa.
if self._params.vad_enabled:
vad_state = self._handle_vad(frame.audio, vad_state)
audio_passthrough = self._params.vad_audio_passthrough
# Push audio downstream if passthrough.
if audio_passthrough:
future = asyncio.run_coroutine_threadsafe(
self._internal_push_frame(frame), self.get_event_loop())
future.result()
self._audio_in_queue.task_done()
except queue.Empty:
pass
except BaseException as e:
logger.error(f"Error pushing audio frames: {e}")

130
src/pipecat/transports/base_output.py
View File

@@ -1,17 +1,11 @@
# #
# Copyright (c) 2024, Daily # Copyright (c) 2024, Daily
# #
# SPDX-License-Identifier: BSD 2-Clause License # SPDX-License-Identifier: BSD 2-Clause License
# #
import asyncio import asyncio
import itertools import itertools
import queue
import time
import threading
from concurrent.futures import ThreadPoolExecutor
from PIL import Image from PIL import Image
from typing import List from typing import List
@@ -40,22 +34,19 @@ class BaseOutputTransport(FrameProcessor):
self._params = params self._params = params
self._running = False
self._allow_interruptions = False self._allow_interruptions = False
self._out_executor = ThreadPoolExecutor(max_workers=5)
# These are the images that we should send to the camera at our desired # These are the images that we should send to the camera at our desired
# framerate. # framerate.
self._camera_images = None self._camera_images = None
# Create media threads queues. # Create media task queues.
if self._params.camera_out_enabled: if self._params.camera_out_enabled:
self._camera_out_queue = queue.Queue() self._camera_out_queue = asyncio.Queue()
self._sink_queue = queue.Queue() self._sink_queue = asyncio.Queue()
self._stopped_event = asyncio.Event() self._stopped_event = asyncio.Event()
self._is_interrupted = threading.Event() self._is_interrupted = asyncio.Event()
async def start(self, frame: StartFrame): async def start(self, frame: StartFrame):
# Make sure we have the latest params. Note that this transport might # Make sure we have the latest params. Note that this transport might
@@ -63,52 +54,40 @@ class BaseOutputTransport(FrameProcessor):
# for example. # for example.
self._allow_interruptions = frame.allow_interruptions self._allow_interruptions = frame.allow_interruptions
if self._running:
return
self._running = True
loop = self.get_event_loop() loop = self.get_event_loop()
if self._params.camera_out_enabled: if self._params.camera_out_enabled:
self._camera_out_thread = loop.run_in_executor( self._camera_out_task = loop.create_task(self._camera_out_task_handler())
self._out_executor, self._camera_out_thread_handler)
self._sink_thread = loop.run_in_executor(self._out_executor, self._sink_thread_handler) self._sink_task = loop.create_task(self._sink_task_handler())
# Create push frame task. This is the task that will push frames in # Create push frame task. This is the task that will push frames in
# order. We also guarantee that all frames are pushed in the same task. # order. We also guarantee that all frames are pushed in the same task.
self._create_push_task() self._create_push_task()
async def stop(self): async def stop(self):
if not self._running:
return
# This will exit all threads.
self._running = False
self._stopped_event.set() self._stopped_event.set()
def send_message(self, frame: TransportMessageFrame): async def cleanup(self):
if self._params.camera_out_enabled:
self._camera_out_task.cancel()
self._sink_task.cancel()
self._push_frame_task.cancel()
async def send_message(self, frame: TransportMessageFrame):
pass pass
def write_frame_to_camera(self, frame: ImageRawFrame): async def write_frame_to_camera(self, frame: ImageRawFrame):
pass pass
def write_raw_audio_frames(self, frames: bytes): async def write_raw_audio_frames(self, frames: bytes):
pass pass
# #
# Frame processor # Frame processor
# #
async def cleanup(self):
# Wait on the threads to finish.
if self._params.camera_out_enabled:
await self._camera_out_thread
await self._sink_thread
async def process_frame(self, frame: Frame, direction: FrameDirection): async def process_frame(self, frame: Frame, direction: FrameDirection):
# #
# Out-of-band frames like (CancelFrame or StartInterruptionFrame) are # Out-of-band frames like (CancelFrame or StartInterruptionFrame) are
@@ -117,7 +96,7 @@ class BaseOutputTransport(FrameProcessor):
# #
if isinstance(frame, StartFrame): if isinstance(frame, StartFrame):
await self.start(frame) await self.start(frame)
self._sink_queue.put(frame) await self._sink_queue.put(frame)
# EndFrame is managed in the queue handler. # EndFrame is managed in the queue handler.
elif isinstance(frame, CancelFrame): elif isinstance(frame, CancelFrame):
await self.stop() await self.stop()
@@ -126,11 +105,11 @@ class BaseOutputTransport(FrameProcessor):
await self._handle_interruptions(frame) await self._handle_interruptions(frame)
await self.push_frame(frame, direction) await self.push_frame(frame, direction)
else: else:
self._sink_queue.put(frame) await self._sink_queue.put(frame)
# If we are finishing, wait here until we have stopped, otherwise we might # If we are finishing, wait here until we have stopped, otherwise we might
# close things too early upstream. We need this event because we don't # close things too early upstream. We need this event because we don't
# know when the internal threads will finish. # know when the internal tasks will finish.
if isinstance(frame, CancelFrame) or isinstance(frame, EndFrame): if isinstance(frame, CancelFrame) or isinstance(frame, EndFrame):
await self._stopped_event.wait() await self._stopped_event.wait()
@@ -145,7 +124,7 @@ class BaseOutputTransport(FrameProcessor):
elif isinstance(frame, StopInterruptionFrame): elif isinstance(frame, StopInterruptionFrame):
self._is_interrupted.clear() self._is_interrupted.clear()
def _sink_thread_handler(self): async def _sink_task_handler(self):
# 10ms bytes # 10ms bytes
bytes_size_10ms = int(self._params.audio_out_sample_rate / 100) * \ bytes_size_10ms = int(self._params.audio_out_sample_rate / 100) * \
self._params.audio_out_channels * 2 self._params.audio_out_channels * 2
@@ -155,37 +134,34 @@ class BaseOutputTransport(FrameProcessor):
# Audio accumlation buffer # Audio accumlation buffer
buffer = bytearray() buffer = bytearray()
while self._running: while True:
try: try:
frame = self._sink_queue.get(timeout=1) frame = await self._sink_queue.get()
if not self._is_interrupted.is_set(): if not self._is_interrupted.is_set():
if isinstance(frame, AudioRawFrame): if isinstance(frame, AudioRawFrame):
if self._params.audio_out_enabled: if self._params.audio_out_enabled:
buffer.extend(frame.audio) buffer.extend(frame.audio)
buffer = self._send_audio_truncated(buffer, smallest_write_size) buffer = await self._send_audio_truncated(buffer, smallest_write_size)
elif isinstance(frame, ImageRawFrame) and self._params.camera_out_enabled: elif isinstance(frame, ImageRawFrame) and self._params.camera_out_enabled:
self._set_camera_image(frame) await self._set_camera_image(frame)
elif isinstance(frame, SpriteFrame) and self._params.camera_out_enabled: elif isinstance(frame, SpriteFrame) and self._params.camera_out_enabled:
self._set_camera_images(frame.images) await self._set_camera_images(frame.images)
elif isinstance(frame, TransportMessageFrame): elif isinstance(frame, TransportMessageFrame):
self.send_message(frame) await self.send_message(frame)
else: else:
future = asyncio.run_coroutine_threadsafe( await self._internal_push_frame(frame)
self._internal_push_frame(frame), self.get_event_loop())
future.result()
else: else:
# If we get interrupted just clear the output buffer. # If we get interrupted just clear the output buffer.
buffer = bytearray() buffer = bytearray()
if isinstance(frame, EndFrame): if isinstance(frame, EndFrame):
# Send all remaining audio before stopping (multiple of 10ms of audio). # Send all remaining audio before stopping (multiple of 10ms of audio).
self._send_audio_truncated(buffer, bytes_size_10ms) await self._send_audio_truncated(buffer, bytes_size_10ms)
future = asyncio.run_coroutine_threadsafe(self.stop(), self.get_event_loop()) await self.stop()
future.result()
self._sink_queue.task_done() self._sink_queue.task_done()
except queue.Empty: except asyncio.CancelledError:
pass break
except BaseException as e: except BaseException as e:
logger.error(f"Error processing sink queue: {e}") logger.error(f"Error processing sink queue: {e}")
@@ -219,7 +195,7 @@ class BaseOutputTransport(FrameProcessor):
async def send_image(self, frame: ImageRawFrame | SpriteFrame): async def send_image(self, frame: ImageRawFrame | SpriteFrame):
await self.process_frame(frame, FrameDirection.DOWNSTREAM) await self.process_frame(frame, FrameDirection.DOWNSTREAM)
def _draw_image(self, frame: ImageRawFrame): async def _draw_image(self, frame: ImageRawFrame):
desired_size = (self._params.camera_out_width, self._params.camera_out_height) desired_size = (self._params.camera_out_width, self._params.camera_out_height)
if frame.size != desired_size: if frame.size != desired_size:
@@ -229,32 +205,32 @@ class BaseOutputTransport(FrameProcessor):
f"{frame} does not have the expected size {desired_size}, resizing") f"{frame} does not have the expected size {desired_size}, resizing")
frame = ImageRawFrame(resized_image.tobytes(), resized_image.size, resized_image.format) frame = ImageRawFrame(resized_image.tobytes(), resized_image.size, resized_image.format)
self.write_frame_to_camera(frame) await self.write_frame_to_camera(frame)
def _set_camera_image(self, image: ImageRawFrame): async def _set_camera_image(self, image: ImageRawFrame):
if self._params.camera_out_is_live: if self._params.camera_out_is_live:
self._camera_out_queue.put(image) await self._camera_out_queue.put(image)
else: else:
self._camera_images = itertools.cycle([image]) self._camera_images = itertools.cycle([image])
def _set_camera_images(self, images: List[ImageRawFrame]): async def _set_camera_images(self, images: List[ImageRawFrame]):
self._camera_images = itertools.cycle(images) self._camera_images = itertools.cycle(images)
def _camera_out_thread_handler(self): async def _camera_out_task_handler(self):
while self._running: while True:
try: try:
if self._params.camera_out_is_live: if self._params.camera_out_is_live:
image = self._camera_out_queue.get(timeout=1) image = await self._camera_out_queue.get()
self._draw_image(image) await self._draw_image(image)
self._camera_out_queue.task_done() self._camera_out_queue.task_done()
elif self._camera_images: elif self._camera_images:
image = next(self._camera_images) image = next(self._camera_images)
self._draw_image(image) await self._draw_image(image)
time.sleep(1.0 / self._params.camera_out_framerate) await asyncio.sleep(1.0 / self._params.camera_out_framerate)
else: else:
time.sleep(1.0 / self._params.camera_out_framerate) await asyncio.sleep(1.0 / self._params.camera_out_framerate)
except queue.Empty: except asyncio.CancelledError:
pass break
except Exception as e: except Exception as e:
logger.error(f"Error writing to camera: {e}") logger.error(f"Error writing to camera: {e}")
@@ -265,13 +241,9 @@ class BaseOutputTransport(FrameProcessor):
async def send_audio(self, frame: AudioRawFrame): async def send_audio(self, frame: AudioRawFrame):
await self.process_frame(frame, FrameDirection.DOWNSTREAM) await self.process_frame(frame, FrameDirection.DOWNSTREAM)
def _send_audio_truncated(self, buffer: bytearray, smallest_write_size: int) -> bytearray: async def _send_audio_truncated(self, buffer: bytearray, smallest_write_size: int) -> bytearray:
try: truncated_length: int = len(buffer) - (len(buffer) % smallest_write_size)
truncated_length: int = len(buffer) - (len(buffer) % smallest_write_size) if truncated_length:
if truncated_length: await self.write_raw_audio_frames(bytes(buffer[:truncated_length]))
self.write_raw_audio_frames(bytes(buffer[:truncated_length])) buffer = buffer[truncated_length:]
buffer = buffer[truncated_length:] return buffer
return buffer
except BaseException as e:
logger.error(f"Error writing audio frames: {e}")
return buffer

4
src/pipecat/transports/base_transport.py
View File

@@ -41,8 +41,8 @@ class TransportParams(BaseModel):
class BaseTransport(ABC): class BaseTransport(ABC):
def __init__(self, loop: asyncio.AbstractEventLoop): def __init__(self, loop: asyncio.AbstractEventLoop | None):
self._loop = loop self._loop = loop or asyncio.get_running_loop()
self._event_handlers: dict = {} self._event_handlers: dict = {}
@abstractmethod @abstractmethod

22
src/pipecat/transports/local/tk.py
View File

@@ -6,6 +6,8 @@
import asyncio import asyncio
from concurrent.futures import ThreadPoolExecutor
import numpy as np import numpy as np
import tkinter as tk import tkinter as tk
@@ -38,6 +40,8 @@ class TkInputTransport(BaseInputTransport):
def __init__(self, py_audio: pyaudio.PyAudio, params: TransportParams): def __init__(self, py_audio: pyaudio.PyAudio, params: TransportParams):
super().__init__(params) super().__init__(params)
self._executor = ThreadPoolExecutor(max_workers=5)
self._in_stream = py_audio.open( self._in_stream = py_audio.open(
format=py_audio.get_format_from_width(2), format=py_audio.get_format_from_width(2),
channels=params.audio_in_channels, channels=params.audio_in_channels,
@@ -45,8 +49,11 @@ class TkInputTransport(BaseInputTransport):
frames_per_buffer=params.audio_in_sample_rate, frames_per_buffer=params.audio_in_sample_rate,
input=True) input=True)
def read_raw_audio_frames(self, frame_count: int) -> bytes: async def read_raw_audio_frames(self, frame_count: int) -> bytes:
return self._in_stream.read(frame_count, exception_on_overflow=False) def read_audio(frame_count: int) -> bytes:
return self._in_stream.read(frame_count, exception_on_overflow=False)
return await self.get_event_loop().run_in_executor(self._executor, read_audio, frame_count)
async def start(self, frame: StartFrame): async def start(self, frame: StartFrame):
await super().start(frame) await super().start(frame)
@@ -70,6 +77,8 @@ class TkOutputTransport(BaseOutputTransport):
def __init__(self, tk_root: tk.Tk, py_audio: pyaudio.PyAudio, params: TransportParams): def __init__(self, tk_root: tk.Tk, py_audio: pyaudio.PyAudio, params: TransportParams):
super().__init__(params) super().__init__(params)
self._executor = ThreadPoolExecutor(max_workers=5)
self._out_stream = py_audio.open( self._out_stream = py_audio.open(
format=py_audio.get_format_from_width(2), format=py_audio.get_format_from_width(2),
channels=params.audio_out_channels, channels=params.audio_out_channels,
@@ -83,10 +92,13 @@ class TkOutputTransport(BaseOutputTransport):
self._image_label = tk.Label(tk_root, image=photo) self._image_label = tk.Label(tk_root, image=photo)
self._image_label.pack() self._image_label.pack()
def write_raw_audio_frames(self, frames: bytes): async def write_raw_audio_frames(self, frames: bytes):
self._out_stream.write(frames) def write_audio(frames: bytes):
self._out_stream.write(frames)
def write_frame_to_camera(self, frame: ImageRawFrame): await self.get_event_loop().run_in_executor(self._executor, write_audio, frames)
async def write_frame_to_camera(self, frame: ImageRawFrame):
self.get_event_loop().call_soon(self._write_frame_to_tk, frame) self.get_event_loop().call_soon(self._write_frame_to_tk, frame)
async def start(self, frame: StartFrame): async def start(self, frame: StartFrame):

100
src/pipecat/transports/network/websocket_server.py
View File

@@ -4,7 +4,6 @@
# SPDX-License-Identifier: BSD 2-Clause License # SPDX-License-Identifier: BSD 2-Clause License
# #
import asyncio import asyncio
import io import io
import wave import wave
@@ -18,12 +17,11 @@ from pipecat.frames.frames import (
CancelFrame, CancelFrame,
EndFrame, EndFrame,
Frame, Frame,
StartFrame, StartFrame)
TTSStartedFrame,
TTSStoppedFrame)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.serializers.base_serializer import FrameSerializer from pipecat.serializers.base_serializer import FrameSerializer
from pipecat.serializers.protobuf import ProtobufFrameSerializer from pipecat.serializers.protobuf import ProtobufFrameSerializer
from pipecat.transports.base_output import BaseOutputTransport
from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.base_transport import BaseTransport, TransportParams
from loguru import logger from loguru import logger
@@ -134,24 +132,15 @@ class WebsocketServerInputTransport(FrameProcessor):
break break
class WebsocketServerOutputTransport(FrameProcessor): class WebsocketServerOutputTransport(BaseOutputTransport):
def __init__(self, params: WebsocketServerParams): def __init__(self, params: WebsocketServerParams):
super().__init__() super().__init__(params)
self._params = params self._params = params
self._websocket = None
self._audio_buffer = bytes()
self._websocket: websockets.WebSocketServerProtocol | None = None self._websocket: websockets.WebSocketServerProtocol | None = None
loop = self.get_event_loop()
self._send_queue_task = loop.create_task(self._send_queue_task_handler())
self._send_queue = asyncio.Queue()
self._audio_buffer = bytes() self._audio_buffer = bytes()
self._in_tts_audio = False
async def set_client_connection(self, websocket: websockets.WebSocketServerProtocol): async def set_client_connection(self, websocket: websockets.WebSocketServerProtocol):
if self._websocket: if self._websocket:
@@ -159,61 +148,34 @@ class WebsocketServerOutputTransport(FrameProcessor):
logger.warning("Only one client allowed, using new connection") logger.warning("Only one client allowed, using new connection")
self._websocket = websocket self._websocket = websocket
async def _send_queue_task_handler(self): async def write_raw_audio_frames(self, frames: bytes):
running = True self._audio_buffer += frames
while running: while len(self._audio_buffer) >= self._params.audio_frame_size:
frame = await self._send_queue.get() frame = AudioRawFrame(
if self._websocket and frame: audio=self._audio_buffer[:self._params.audio_frame_size],
# We send WAV data so we can easily decoded in the browser. sample_rate=self._params.audio_out_sample_rate,
if self._params.add_wav_header: num_channels=self._params.audio_out_channels
content = io.BytesIO() )
ww = wave.open(content, "wb")
ww.setsampwidth(2)
ww.setnchannels(frame.num_channels)
ww.setframerate(frame.sample_rate)
ww.writeframes(frame.audio)
ww.close()
content.seek(0)
wav_frame = AudioRawFrame(
content.read(),
sample_rate=frame.sample_rate,
num_channels=frame.num_channels)
frame = wav_frame
proto = self._params.serializer.serialize(frame)
await self._websocket.send(proto)
async def _stop(self): if self._params.add_wav_header:
self._send_queue_task.cancel() content = io.BytesIO()
ww = wave.open(content, "wb")
ww.setsampwidth(2)
ww.setnchannels(frame.num_channels)
ww.setframerate(frame.sample_rate)
ww.writeframes(frame.audio)
ww.close()
content.seek(0)
wav_frame = AudioRawFrame(
content.read(),
sample_rate=frame.sample_rate,
num_channels=frame.num_channels)
frame = wav_frame
async def process_frame(self, frame: Frame, direction: FrameDirection): proto = self._params.serializer.serialize(frame)
if isinstance(frame, CancelFrame): await self._websocket.send(proto)
await self._stop()
await self.push_frame(frame, direction) self._audio_buffer = self._audio_buffer[self._params.audio_frame_size:]
elif isinstance(frame, TTSStartedFrame):
self._in_tts_audio = True
elif isinstance(frame, AudioRawFrame):
if self._in_tts_audio:
self._audio_buffer += frame.audio
while len(self._audio_buffer) >= self._params.audio_frame_size:
frame = AudioRawFrame(
audio=self._audio_buffer[:self._params.audio_frame_size],
sample_rate=self._params.audio_out_sample_rate,
num_channels=self._params.audio_out_channels
)
await self._send_queue.put(frame)
self._audio_buffer = self._audio_buffer[self._params.audio_frame_size:]
elif isinstance(frame, TTSStoppedFrame):
self._in_tts_audio = False
if self._audio_buffer:
frame = AudioRawFrame(
audio=self._audio_buffer,
sample_rate=self._params.audio_out_sample_rate,
num_channels=self._params.audio_out_channels
)
await self._send_queue.put(frame)
self._audio_buffer = bytes()
else:
await self.push_frame(frame, direction)
class WebsocketServerTransport(BaseTransport): class WebsocketServerTransport(BaseTransport):
@@ -223,7 +185,7 @@ class WebsocketServerTransport(BaseTransport):
host: str = "localhost", host: str = "localhost",
port: int = 8765, port: int = 8765,
params: WebsocketServerParams = WebsocketServerParams(), params: WebsocketServerParams = WebsocketServerParams(),
loop: asyncio.AbstractEventLoop = asyncio.get_event_loop()): loop: asyncio.AbstractEventLoop | None = None):
super().__init__(loop) super().__init__(loop)
self._host = host self._host = host
self._port = port self._port = port

95
src/pipecat/transports/services/daily.py
View File

@@ -182,9 +182,6 @@ class DailyTransportClient(EventHandler):
def participant_id(self) -> str: def participant_id(self) -> str:
return self._participant_id return self._participant_id
def set_callbacks(self, callbacks: DailyCallbacks):
self._callbacks = callbacks
def send_message(self, frame: DailyTransportMessageFrame): def send_message(self, frame: DailyTransportMessageFrame):
self._client.send_app_message(frame.message, frame.participant_id) self._client.send_app_message(frame.message, frame.participant_id)
@@ -464,8 +461,10 @@ class DailyInputTransport(BaseInputTransport):
self._client = client self._client = client
self._executor = ThreadPoolExecutor(max_workers=5)
self._video_renderers = {} self._video_renderers = {}
self._camera_in_queue = queue.Queue() self._camera_in_queue = asyncio.Queue()
self._vad_analyzer = params.vad_analyzer self._vad_analyzer = params.vad_analyzer
if params.vad_enabled and not params.vad_analyzer: if params.vad_enabled and not params.vad_analyzer:
@@ -474,38 +473,34 @@ class DailyInputTransport(BaseInputTransport):
num_channels=self._params.audio_in_channels) num_channels=self._params.audio_in_channels)
async def start(self, frame: StartFrame): async def start(self, frame: StartFrame):
if self._running:
return
# Join the room. # Join the room.
await self._client.join() await self._client.join()
# This will set _running=True # Create camera in task
self._camera_in_task = self.get_event_loop().create_task(self._camera_in_task_handler())
await super().start(frame) await super().start(frame)
# Create camera in thread (runs if _running is true).
self._camera_in_thread = self._loop.run_in_executor(
self._in_executor, self._camera_in_thread_handler)
async def stop(self): async def stop(self):
if not self._running:
return
# Leave the room. # Leave the room.
await self._client.leave() await self._client.leave()
# This will set _running=False # The task will stop.
self._camera_in_task.cancel()
await super().stop() await super().stop()
# The thread will stop.
await self._camera_in_thread
async def cleanup(self): async def cleanup(self):
await super().cleanup()
await self._client.cleanup() await self._client.cleanup()
await super().cleanup()
def vad_analyze(self, audio_frames: bytes) -> VADState: async def vad_analyze(self, audio_frames: bytes) -> VADState:
state = VADState.QUIET state = VADState.QUIET
if self._vad_analyzer: if self._vad_analyzer:
state = self._vad_analyzer.analyze_audio(audio_frames) state = await self._vad_analyzer.analyze_audio(audio_frames)
return state return state
def read_raw_audio_frames(self, frame_count: int) -> bytes: async def read_raw_audio_frames(self, frame_count: int) -> bytes:
return self._client.read_raw_audio_frames(frame_count) def read_audio(frame_count: int) -> bytes:
return self._client.read_raw_audio_frames(frame_count)
return await self.get_event_loop().run_in_executor(self._executor, read_audio, frame_count)
# #
# FrameProcessor # FrameProcessor
@@ -580,20 +575,19 @@ class DailyInputTransport(BaseInputTransport):
image=buffer, image=buffer,
size=size, size=size,
format=format) format=format)
self._camera_in_queue.put(frame) asyncio.run_coroutine_threadsafe(
self._camera_in_queue.put(frame), self.get_event_loop())
self._video_renderers[participant_id]["timestamp"] = curr_time self._video_renderers[participant_id]["timestamp"] = curr_time
def _camera_in_thread_handler(self): async def _camera_in_task_handler(self):
while self._running: while True:
try: try:
frame = self._camera_in_queue.get(timeout=1) frame = await self._camera_in_queue.get()
future = asyncio.run_coroutine_threadsafe( await self._internal_push_frame(frame)
self._internal_push_frame(frame), self.get_event_loop())
future.result()
self._camera_in_queue.task_done() self._camera_in_queue.task_done()
except queue.Empty: except asyncio.CancelledError:
pass break
except BaseException as e: except BaseException as e:
logger.error(f"Error capturing video: {e}") logger.error(f"Error capturing video: {e}")
@@ -605,34 +599,39 @@ class DailyOutputTransport(BaseOutputTransport):
self._client = client self._client = client
self._executor = ThreadPoolExecutor(max_workers=5)
async def start(self, frame: StartFrame): async def start(self, frame: StartFrame):
if self._running:
return
# This will set _running=True
await super().start(frame)
# Join the room. # Join the room.
await self._client.join() await self._client.join()
await super().start(frame)
async def stop(self): async def stop(self):
if not self._running:
return
# This will set _running=False
await super().stop()
# Leave the room. # Leave the room.
await self._client.leave() await self._client.leave()
await super().stop()
async def cleanup(self): async def cleanup(self):
await super().cleanup()
await self._client.cleanup() await self._client.cleanup()
await super().cleanup()
def send_message(self, frame: DailyTransportMessageFrame): async def send_message(self, frame: DailyTransportMessageFrame):
self._client.send_message(frame) def send_message(frame: DailyTransportMessageFrame):
self._client.send_message(frame)
def write_raw_audio_frames(self, frames: bytes): await self.get_event_loop().run_in_executor(self._executor, send_message, frame)
self._client.write_raw_audio_frames(frames)
def write_frame_to_camera(self, frame: ImageRawFrame): async def write_raw_audio_frames(self, frames: bytes):
self._client.write_frame_to_camera(frame) def write_audio(frames: bytes):
self._client.write_raw_audio_frames(frames)
await self.get_event_loop().run_in_executor(self._executor, write_audio, frames)
async def write_frame_to_camera(self, frame: ImageRawFrame):
def write_image(frame: ImageRawFrame):
self._client.write_frame_to_camera(frame)
await self.get_event_loop().run_in_executor(self._executor, write_image, frame)
class DailyTransport(BaseTransport): class DailyTransport(BaseTransport):
@@ -643,7 +642,7 @@ class DailyTransport(BaseTransport):
token: str | None, token: str | None,
bot_name: str, bot_name: str,
params: DailyParams, params: DailyParams,
loop: asyncio.AbstractEventLoop = asyncio.get_event_loop()): loop: asyncio.AbstractEventLoop | None = None):
super().__init__(loop) super().__init__(loop)
callbacks = DailyCallbacks( callbacks = DailyCallbacks(
@@ -663,7 +662,8 @@ class DailyTransport(BaseTransport):
) )
self._params = params self._params = params
self._client = DailyTransportClient(room_url, token, bot_name, params, callbacks, loop) self._client = DailyTransportClient(
room_url, token, bot_name, params, callbacks, self._loop)
self._input: DailyInputTransport | None = None self._input: DailyInputTransport | None = None
self._output: DailyOutputTransport | None = None self._output: DailyOutputTransport | None = None
@@ -831,6 +831,5 @@ class DailyTransport(BaseTransport):
def _call_async_event_handler(self, event_name: str, *args, **kwargs): def _call_async_event_handler(self, event_name: str, *args, **kwargs):
future = asyncio.run_coroutine_threadsafe( future = asyncio.run_coroutine_threadsafe(
self._call_event_handler( self._call_event_handler(event_name, *args, **kwargs), self._loop)
event_name, args, kwargs), self._loop)
future.result() future.result()

4
src/pipecat/vad/silero.py
View File

@@ -46,7 +46,7 @@ class SileroVADAnalyzer(VADAnalyzer):
def num_frames_required(self) -> int: def num_frames_required(self) -> int:
return int(self.sample_rate / 100) * 4 # 40ms return int(self.sample_rate / 100) * 4 # 40ms
def voice_confidence(self, buffer) -> float: async def voice_confidence(self, buffer) -> float:
try: try:
audio_int16 = np.frombuffer(buffer, np.int16) audio_int16 = np.frombuffer(buffer, np.int16)
# Divide by 32768 because we have signed 16-bit data. # Divide by 32768 because we have signed 16-bit data.
@@ -88,7 +88,7 @@ class SileroVAD(FrameProcessor):
async def _analyze_audio(self, frame: AudioRawFrame): async def _analyze_audio(self, frame: AudioRawFrame):
# Check VAD and push event if necessary. We just care about changes # Check VAD and push event if necessary. We just care about changes
# from QUIET to SPEAKING and vice versa. # from QUIET to SPEAKING and vice versa.
new_vad_state = self._vad_analyzer.analyze_audio(frame.audio) new_vad_state = await self._vad_analyzer.analyze_audio(frame.audio)
if new_vad_state != self._processor_vad_state and new_vad_state != VADState.STARTING and new_vad_state != VADState.STOPPING: if new_vad_state != self._processor_vad_state and new_vad_state != VADState.STARTING and new_vad_state != VADState.STOPPING:
new_frame = None new_frame = None

6
src/pipecat/vad/vad_analyzer.py
View File

@@ -58,14 +58,14 @@ class VADAnalyzer:
pass pass
@abstractmethod @abstractmethod
def voice_confidence(self, buffer) -> float: async def voice_confidence(self, buffer) -> float:
pass pass
def _get_smoothed_volume(self, audio: bytes) -> float: def _get_smoothed_volume(self, audio: bytes) -> float:
volume = calculate_audio_volume(audio, self._sample_rate) volume = calculate_audio_volume(audio, self._sample_rate)
return exp_smoothing(volume, self._prev_volume, self._smoothing_factor) return exp_smoothing(volume, self._prev_volume, self._smoothing_factor)
def analyze_audio(self, buffer) -> VADState: async def analyze_audio(self, buffer) -> VADState:
self._vad_buffer += buffer self._vad_buffer += buffer
num_required_bytes = self._vad_frames_num_bytes num_required_bytes = self._vad_frames_num_bytes
@@ -75,7 +75,7 @@ class VADAnalyzer:
audio_frames = self._vad_buffer[:num_required_bytes] audio_frames = self._vad_buffer[:num_required_bytes]
self._vad_buffer = self._vad_buffer[num_required_bytes:] self._vad_buffer = self._vad_buffer[num_required_bytes:]
confidence = self.voice_confidence(audio_frames) confidence = await self.voice_confidence(audio_frames)
volume = self._get_smoothed_volume(audio_frames) volume = self._get_smoothed_volume(audio_frames)
self._prev_volume = volume self._prev_volume = volume