PR feedback

examples/canonical-metrics/bot.py

@@ -20,8 +20,10 @@ from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
 from pipecat.processors.aggregators.llm_response import (
     LLMAssistantResponseAggregator, LLMUserResponseAggregator)
-from pipecat.processors.canonical_metrics_processor import CanonicalMetrics
+from pipecat.processors.audio.audio_buffer_processor import \
+    AudioBufferProcessor
 from pipecat.processors.user_marker_processor import UserMarkerProcessor
+from pipecat.services.canonical import CanonicalMetricsService
 from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -103,7 +105,12 @@ async def main():
         call completion, CanonicalMetrics will send the audio buffer to Canonical for
         analysis. Visit https://voice.canonical.chat to learn more.
         """
-        canonical = CanonicalMetrics(
+        audio_buffer_processor = AudioBufferProcessor()
+        canonical = CanonicalMetricsService(
+            audio_buffer_processor=audio_buffer_processor,
+            aiohttp_session=session,
+            api_key=os.getenv("CANONICAL_API_KEY"),
+            api_url=os.getenv("CANONICAL_API_URL"),
             call_id=str(uuid.uuid4()),
             assistant="pipecat-chatbot",
             assistant_speaks_first=True,
@@ -111,11 +118,12 @@ async def main():
         usermarker = UserMarkerProcessor()
         pipeline = Pipeline([
             transport.input(),  # microphone
-            usermarker,  # used to mark the user's audio in the pipeline
+            usermarker,
             user_response,
             llm,
             tts,
-            canonical,  # captures audio and uploads to Canonical AI for metrics
+            audio_buffer_processor, # captures audio into a buffer
+            canonical,  # uploads audio buffer to Canonical AI for metrics
             transport.output(),
             assistant_response,
         ])

examples/chatbot-audio-recording/bot.py

@@ -19,7 +19,7 @@ from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
 from pipecat.processors.aggregators.llm_response import (
     LLMAssistantResponseAggregator, LLMUserResponseAggregator)
-from pipecat.processors.audio_buffer_processor import AudioBufferProcessor
+from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor
 from pipecat.processors.user_marker_processor import UserMarkerProcessor
 from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.openai import OpenAILLMService

src/pipecat/processors/audio/audio_buffer_processor.py

@@ -1,6 +1,7 @@
 from pipecat.frames.frames import (AudioRawFrame, BotStartedSpeakingFrame,
                                    BotStoppedSpeakingFrame, Frame,
-                                   UserAudioFrame, UserStoppedSpeakingFrame)
+                                   UserAudioFrame, UserStartedSpeakingFrame,
+                                   UserStoppedSpeakingFrame)
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 
 
@@ -21,54 +22,61 @@ class AudioBufferProcessor(FrameProcessor):
         populated when the first audio frame is processed.
         """
         super().__init__()
-        self.audio_buffer = bytearray()
+        self._audio_buffer = bytearray()
-        self.num_channels = None
+        self._num_channels = None
-        self.sample_rate = None
+        self._sample_rate = None
-        self.assistant_audio = False
+        self._assistant_audio = False
-        self.user_audio = False
+        self._user_audio = False
+        print(f"ctor::AudioBufferProcessor object memory address: {id(self)}")
 
-    def has_audio(self):
+    def _has_audio(self):
         return (
-            self.audio_buffer and
+            self._audio_buffer is not None and
-            len(self.audio_buffer) > 0 and
+            len(self._audio_buffer) > 0 and
-            self.num_channels and
+            self._num_channels is not None and
-            self.sample_rate
+            self._sample_rate is not None
         )
 
+    def _reset_audio_buffer(self):
+        self._audio_buffer = bytearray()
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
-        if isinstance(frame, AudioRawFrame) or isinstance(frame, UserAudioFrame):
+        if isinstance(frame, AudioRawFrame):
-            if self.num_channels is None:
+            if self._num_channels is None:
-                self.num_channels = frame.num_channels
+                self._num_channels = frame.num_channels
-            if self.sample_rate is None:
+            if self._sample_rate is None:
-                self.sample_rate = frame.sample_rate
+                self._sample_rate = frame.sample_rate
 
-        elif isinstance(frame, UserStoppedSpeakingFrame):
+        if isinstance(frame, UserStoppedSpeakingFrame):
-            self.user_audio = False
+            self._user_audio = False
 
         if isinstance(frame, BotStartedSpeakingFrame):
-            self.assistant_audio = True
+            self._assistant_audio = True
-            self.user_audio = False  # do not capture user audio if assistant is speaking
+            self._user_audio = False  # do not capture user audio if assistant is speaking
+
         if isinstance(frame, BotStoppedSpeakingFrame):
-            self.assistant_audio = False
+            self._assistant_audio = False
             # Capture user audio if assistant is not speaking, even if it's silence, the point
             # here is to capture so that the conversation is as close to reality as possible.
             # This is important for evaluation and metrics capture.
-            self.user_audio = True
+            self._user_audio = True
 
         # only include audio from the user if the user is speaking, this is because audio from the user's
         # mic is always coming in. if we include all the user's audio there will be a long latency before
         # the user starts speaking because all of the user's silence during the assistant's speech will have been
         # added to the buffer.
-        if isinstance(frame, UserAudioFrame) and self.user_audio:
+        #
-            self.audio_buffer.extend(frame.audio)
+        # and include all audio from the assistant
+        if isinstance(frame, UserAudioFrame) and self._user_audio:
+            self._audio_buffer.extend(frame.audio)
 
         # include all audio from the assistant
         if (
             isinstance(frame, AudioRawFrame)
             and not isinstance(frame, UserAudioFrame)
         ):
-            self.audio_buffer.extend(frame.audio)
+            self._audio_buffer.extend(frame.audio)
 
         # do not push the user's audio frame, doing so will result in echo
         if not isinstance(frame, UserAudioFrame):

src/pipecat/processors/canonical_metrics_processor.py

@@ -1,202 +0,0 @@
-import os
-import uuid
-import wave
-from datetime import datetime
-from io import BytesIO
-from typing import Dict, List, Tuple
-
-import aiohttp
-from loguru import logger
-
-try:
-    import aiofiles
-    import aiofiles.os
-except ModuleNotFoundError as e:
-    logger.error(f"Exception: {e}")
-    logger.error(
-        "In order to use Canonical Metrics, you need to `pip install pipecat-ai[canonical]`. " +
-        "Also, set the `CANONICAL_API_KEY` environment variable.")
-    raise Exception(f"Missing module: {e}")
-
-
-from pipecat.frames.frames import CancelFrame, EndFrame, Frame
-from pipecat.processors.audio_buffer_processor import AudioBufferProcessor
-from pipecat.processors.frame_processor import FrameDirection
-
-"""
-This class extends AudioBufferProcessor to handle audio processing and uploading
-for the Canonical Voice API.
-"""
-
-
-class CanonicalMetrics(AudioBufferProcessor):
-    """
-    Initialize a CanonicalAudioProcessor instance.
-
-    This class extends AudioBufferProcessor to handle audio processing and uploading
-    for the Canonical Voice API.
-
-    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.
-        assistant_speaks_first (bool, optional): Indicates if the assistant speaks first in the conversation. Defaults to True.
-        output_dir (str, optional): Directory to save temporary audio files. Defaults to "recordings".
-        default_part_size (int, optional): Default size for multipart upload parts in bytes. Defaults to 1MB (1024 * 1024 * 1).
-
-    Attributes:
-        call_id (str): Stores the unique call identifier.
-        assistant (str): Stores the assistant identifier.
-        assistant_speaks_first (bool): Indicates whether the assistant speaks first.
-        output_dir (str): Directory path for saving temporary audio files.
-        partsize (int): Size of each part for multipart uploads.
-
-    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__(
-            self,
-            call_id: str,
-            assistant: str,
-            assistant_speaks_first: bool = True,
-            output_dir: str = "recordings",
-            default_part_size: int = 1024 * 1024 * 1):
-        super().__init__()
-        if not os.environ.get("CANONICAL_API_KEY"):
-            raise ValueError(
-                "CANONICAL_API_KEY is not set, a Canonical API key is required to use this class")
-        self.call_id = call_id
-        self.assistant = assistant
-        self.assistant_speaks_first = assistant_speaks_first
-        self.output_dir = output_dir
-        self.partsize = default_part_size
-        self.end_of_call = False
-
-    async def process_frame(self, frame: Frame, direction: FrameDirection):
-        await super().process_frame(frame, direction)
-        if self.end_of_call:
-            return
-
-        if (isinstance(frame, EndFrame) or isinstance(frame, CancelFrame)):
-            self.end_of_call = True
-            if self.has_audio():
-                os.makedirs(self.output_dir, exist_ok=True)
-                filename = self.get_output_filename()
-                with BytesIO() as buffer:
-                    with wave.open(buffer, 'wb') as wf:
-                        wf.setnchannels(self.num_channels)
-                        wf.setsampwidth(self.sample_rate // 8000)
-                        wf.setframerate(self.sample_rate)
-                        wf.writeframes(self.audio_buffer)
-                    wave_data = buffer.getvalue()
-
-                async with aiofiles.open(filename, 'wb') as file:
-                    await file.write(wave_data)
-
-                try:
-                    await self.multipart_upload(filename)
-                    await aiofiles.os.remove(filename)
-                except FileNotFoundError:
-                    pass
-                except Exception as e:
-                    raise e
-                self.audio_buffer = bytearray()
-
-    def get_output_filename(self):
-        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
-        return f"{self.output_dir}/{timestamp}-{uuid.uuid4().hex}.wav"
-
-    def canonical_api_url(self):
-        return os.environ.get("CANONICAL_API_URL", "https://voiceapp.canonical.chat/api/v1")
-
-    def request_headers(self):
-        return {
-            "Content-Type": "application/json",
-            "X-Canonical-Api-Key": os.environ.get("CANONICAL_API_KEY")
-        }
-
-    async def multipart_upload(self, file_path: str):
-        upload_request, upload_response = await self.request_upload(file_path)
-        parts = await self.upload_parts(file_path, upload_request, upload_response)
-        await self.upload_complete(parts, upload_request, upload_response)
-
-    async def request_upload(self, file_path: str) -> Tuple[Dict, Dict]:
-        filename = os.path.basename(file_path)
-        filename = f"{str(uuid.uuid4())}-{filename}"
-        filesize = os.path.getsize(file_path)
-        numparts = int((filesize + self.partsize - 1) / self.partsize)
-
-        params = {
-            'filename': filename,
-            'parts': numparts,
-            'assistant': self.assistant,
-            'assistantSpeaksFirst': self.assistant_speaks_first
-        }
-        print(f"Requesting presigned URLs for {numparts} parts")
-        async with aiohttp.ClientSession() as session:
-            async with session.post(
-                f"{self.canonical_api_url()}/recording/uploadRequest",
-                headers=self.request_headers(),
-                json=params
-            ) as response:
-                if not response.ok:
-                    raise Exception(f"Failed to get presigned URLs: {await response.text()}")
-                response_json = await response.json()
-        return params, response_json
-
-    async def upload_parts(
-            self,
-            file_path: str,
-            upload_request: Dict,
-            upload_response: Dict) -> List[Dict]:
-
-        urls = upload_response['urls']
-        parts = []
-        try:
-            async with aiofiles.open(file_path, 'rb') as file:
-                async with aiohttp.ClientSession() as session:
-                    for partnum, upload_url in enumerate(urls, start=1):
-                        data = await file.read(self.partsize)
-                        if not data:
-                            break
-
-                        async with session.put(upload_url, data=data) as response:
-                            if not response.ok:
-                                logger.error(f"Failed to upload part {partnum}: {await response.text()}")
-                                raise Exception(f"Failed to upload part {partnum}: {await response.text()}")
-
-                            etag = response.headers['ETag']
-                            parts.append({'partnum': str(partnum), 'etag': etag})
-
-        except Exception as e:
-            logger.error(f"Multipart upload aborted, an error occurred: {str(e)}")
-        return parts
-
-    async def upload_complete(
-            self,
-            parts: List[Dict],
-            upload_request: Dict,
-            upload_response: Dict):
-
-        params = {
-            'filename': upload_request['filename'],
-            'parts': parts,
-            'slug': upload_response['slug'],
-            'callId': self.call_id,
-            'assistant': {
-                'id': self.assistant,
-                'speaksFirst': self.assistant_speaks_first
-            }
-        }
-        print(f"Completing upload for {params['filename']}")
-        print(f"Slug: {params['slug']}")
-        async with aiohttp.ClientSession() as session:
-            async with session.post(
-                f"{self.canonical_api_url()}/recording/uploadComplete",
-                headers=self.request_headers(),
-                json=params
-            ) as response:
-                if not response.ok:
-                    logger.error(f"Failed to complete upload: {await response.text()}")
-                    raise Exception(f"Failed to complete upload: {await response.text()}")

src/pipecat/services/canonical.py

@@ -0,0 +1,212 @@
+import os
+import uuid
+import wave
+from datetime import datetime
+from io import BytesIO
+from typing import Dict, List, Tuple
+
+import aiohttp
+from loguru import logger
+
+try:
+    import aiofiles
+    import aiofiles.os
+except ModuleNotFoundError as e:
+    logger.error(f"Exception: {e}")
+    logger.error(
+        "In order to use Canonical Metrics, you need to `pip install pipecat-ai[canonical]`. " +
+        "Also, set the `CANONICAL_API_KEY` environment variable.")
+    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.
+
+    This class extends AudioBufferProcessor to handle audio processing and uploading
+    for the Canonical Voice API.
+
+    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.
+        assistant_speaks_first (bool, optional): Indicates if the assistant speaks first in the conversation. Defaults to True.
+        output_dir (str, optional): Directory to save temporary audio files. Defaults to "recordings".
+
+    Attributes:
+        call_id (str): Stores the unique call identifier.
+        assistant (str): Stores the assistant identifier.
+        assistant_speaks_first (bool): Indicates whether the assistant speaks first.
+        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__(
+            self,
+            aiohttp_session: aiohttp.ClientSession,
+            audio_buffer_processor: AudioBufferProcessor,
+            call_id: str,
+            assistant: str,
+            api_key: str,
+            api_url: str = "https://voiceapp.canonical.chat/api/v1",
+            assistant_speaks_first: bool = True,
+            output_dir: str = "recordings"):
+        super().__init__()
+        self._aiohttp_session = aiohttp_session
+        self._audio_buffer_processor = audio_buffer_processor
+        self._api_key = api_key
+        self._api_url = api_url
+        self._call_id = call_id
+        self._assistant = assistant
+        self._assistant_speaks_first = assistant_speaks_first
+        self._output_dir = output_dir
+
+    async def stop(self, frame: EndFrame):
+        await self._process_audio()
+
+    async def cancel(self, frame: CancelFrame):
+        await self._process_audio()
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+        await self.push_frame(frame, direction)
+
+    async def _process_audio(self):
+        pipeline = self._audio_buffer_processor
+        if pipeline._has_audio():
+            os.makedirs(self._output_dir, exist_ok=True)
+            filename = self._get_output_filename()
+            with BytesIO() as buffer:
+                with wave.open(buffer, 'wb') as wf:
+                    wf.setnchannels(pipeline._num_channels)
+                    wf.setsampwidth(pipeline._sample_rate // 8000)
+                    wf.setframerate(pipeline._sample_rate)
+                    wf.writeframes(pipeline._audio_buffer)
+                wave_data = buffer.getvalue()
+
+            async with aiofiles.open(filename, 'wb') as file:
+                await file.write(wave_data)
+
+            try:
+                await self._multipart_upload(filename)
+                pipeline._reset_audio_buffer()
+                # await aiofiles.os.remove(filename)
+            except FileNotFoundError:
+                pass
+            except Exception as e:
+                logger.error(f"Failed to upload recording: {e}")
+
+    def _get_output_filename(self):
+        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+        return f"{self._output_dir}/{timestamp}-{uuid.uuid4().hex}.wav"
+
+    def _request_headers(self):
+        return {
+            "Content-Type": "application/json",
+            "X-Canonical-Api-Key": self._api_key
+        }
+
+    async def _multipart_upload(self, file_path: str):
+        upload_request, upload_response = await self._request_upload(file_path)
+        if upload_request is None or upload_response is None:
+            return
+        parts = await self._upload_parts(file_path, upload_response)
+        if parts is None:
+            return
+        await self._upload_complete(parts, upload_request, upload_response)
+
+    async def _request_upload(self, file_path: str) -> Tuple[Dict, Dict]:
+        filename = os.path.basename(file_path)
+        filesize = os.path.getsize(file_path)
+        numparts = int((filesize + PART_SIZE - 1) / PART_SIZE)
+
+        params = {
+            'filename': filename,
+            'parts': numparts,
+            'callId': self._call_id,
+            'assistant': {
+                'id': self._assistant,
+                'speaksFirst': self._assistant_speaks_first
+            }
+        }
+        logger.debug(f"Requesting presigned URLs for {numparts} parts")
+        response = await self._aiohttp_session.post(
+            f"{self._api_url}/recording/uploadRequest",
+            headers=self._request_headers(),
+            json=params
+        )
+        if not response.ok:
+            logger.error(f"Failed to get presigned URLs: {await response.text()}")
+            return None, None
+        response_json = await response.json()
+        return params, response_json
+
+    async def _upload_parts(
+            self,
+            file_path: str,
+            upload_response: Dict) -> List[Dict]:
+
+        urls = upload_response['urls']
+        parts = []
+        try:
+            async with aiofiles.open(file_path, 'rb') as file:
+                for partnum, upload_url in enumerate(urls, start=1):
+                    data = await file.read(PART_SIZE)
+                    if not data:
+                        break
+
+                    response = await self._aiohttp_session.put(upload_url, data=data)
+                    if not response.ok:
+                        logger.error(f"Failed to upload part {partnum}: {await response.text()}")
+                        return None
+
+                    etag = response.headers['ETag']
+                    parts.append({'partnum': str(partnum), 'etag': etag})
+
+        except Exception as e:
+            logger.error(f"Multipart upload aborted, an error occurred: {str(e)}")
+        return parts
+
+    async def _upload_complete(
+            self,
+            parts: List[Dict],
+            upload_request: Dict,
+            upload_response: Dict):
+
+        params = {
+            'filename': upload_request['filename'],
+            'parts': parts,
+            'slug': upload_response['slug'],
+            'callId': self._call_id,
+            'assistant': {
+                'id': self._assistant,
+                'speaksFirst': self._assistant_speaks_first
+            }
+        }
+        logger.debug(f"Completing upload for {params['filename']}")
+        logger.debug(f"Slug: {params['slug']}")
+        response = await self._aiohttp_session.post(
+            f"{self._api_url}/recording/uploadComplete",
+            headers=self._request_headers(),
+            json=params
+        )
+        if not response.ok:
+            logger.error(f"Failed to complete upload: {await response.text()}")
+            return
+