diff --git a/src/pipecat/processors/audio/audio_buffer_processor.py b/src/pipecat/processors/audio/audio_buffer_processor.py index 0c07d6815..25a1f0237 100644 --- a/src/pipecat/processors/audio/audio_buffer_processor.py +++ b/src/pipecat/processors/audio/audio_buffer_processor.py @@ -1,18 +1,17 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + import wave from io import BytesIO from pipecat.frames.frames import ( AudioRawFrame, - BotInterruptionFrame, - BotStartedSpeakingFrame, - BotStoppedSpeakingFrame, Frame, InputAudioRawFrame, OutputAudioRawFrame, - StartInterruptionFrame, - StopInterruptionFrame, - UserStartedSpeakingFrame, - UserStoppedSpeakingFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor @@ -39,18 +38,18 @@ class AudioBufferProcessor(FrameProcessor): def _buffer_has_audio(self, buffer: bytearray): return buffer is not None and len(buffer) > 0 - def _has_audio(self): + def has_audio(self): return ( self._buffer_has_audio(self._user_audio_buffer) and self._buffer_has_audio(self._assistant_audio_buffer) and self._sample_rate is not None ) - def _reset_audio_buffer(self): + def reset_audio_buffer(self): self._user_audio_buffer = bytearray() self._assistant_audio_buffer = bytearray() - def _merge_audio_buffers(self): + def merge_audio_buffers(self): with BytesIO() as buffer: with wave.open(buffer, "wb") as wf: wf.setnchannels(2) diff --git a/src/pipecat/services/canonical.py b/src/pipecat/services/canonical.py index 0f4fb6a2f..048a6a4ee 100644 --- a/src/pipecat/services/canonical.py +++ b/src/pipecat/services/canonical.py @@ -1,9 +1,21 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import aiohttp import os import uuid + from datetime import datetime from typing import Dict, List, Tuple -import aiohttp +from pipecat.frames.frames import CancelFrame, EndFrame, Frame +from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor +from pipecat.processors.frame_processor import FrameDirection +from pipecat.services.ai_services import AIService + from loguru import logger try: @@ -18,27 +30,18 @@ except ModuleNotFoundError as e: raise Exception(f"Missing module: {e}") -from pipecat.frames.frames import CancelFrame, EndFrame, Frame -from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor -from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.ai_services import AIService - # Multipart upload part size in bytes, cannot be smaller than 5MB PART_SIZE = 1024 * 1024 * 5 -""" -This class extends AudioBufferProcessor to handle audio processing and uploading -for the Canonical Voice API. -""" class CanonicalMetricsService(AIService): - """ - Initialize a CanonicalAudioProcessor instance. + """Initialize a CanonicalAudioProcessor instance. - This class extends AudioBufferProcessor to handle audio processing and uploading - for the Canonical Voice API. + This class uses an AudioBufferProcessor to get the conversation audio and + uploads it to Canonical Voice API for audio processing. Args: + call_id (str): Your unique identifier for the call. This is used to match the call in the Canonical Voice system to the call in your system. assistant (str): Identifier for the AI assistant. This can be whatever you want, it's intended for you convenience so you can distinguish between different assistants and a grouping mechanism for calls. @@ -52,7 +55,6 @@ class CanonicalMetricsService(AIService): output_dir (str): Directory path for saving temporary audio files. The constructor also ensures that the output directory exists. - This class requires a Canonical API key to be set in the CANONICAL_API_KEY environment variable. """ def __init__( @@ -90,17 +92,17 @@ class CanonicalMetricsService(AIService): async def _process_audio(self): pipeline = self._audio_buffer_processor - if pipeline._has_audio(): + if pipeline.has_audio(): os.makedirs(self._output_dir, exist_ok=True) filename = self._get_output_filename() - wave_data = pipeline._merge_audio_buffers() + wave_data = pipeline.merge_audio_buffers() async with aiofiles.open(filename, "wb") as file: await file.write(wave_data) try: await self._multipart_upload(filename) - pipeline._reset_audio_buffer() + pipeline.reset_audio_buffer() await aiofiles.os.remove(filename) except FileNotFoundError: pass diff --git a/src/pipecat/transports/services/livekit.py b/src/pipecat/transports/services/livekit.py index d8aff2ecb..25678db4e 100644 --- a/src/pipecat/transports/services/livekit.py +++ b/src/pipecat/transports/services/livekit.py @@ -1,3 +1,9 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + import asyncio from dataclasses import dataclass from typing import Any, Awaitable, Callable, List