From a1d46cb26bbb489d9ddcc414c59dced7fe6983b8 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 6 May 2025 21:23:23 -0400 Subject: [PATCH] Removing CanonicalMetricsService --- CHANGELOG.md | 4 + README.md | 2 +- docs/api/requirements.txt | 1 - examples/canonical-metrics/.gitignore | 161 -------------- examples/canonical-metrics/Dockerfile | 10 - examples/canonical-metrics/README.md | 66 ------ examples/canonical-metrics/bot.py | 146 ------------- examples/canonical-metrics/env.example | 6 - examples/canonical-metrics/requirements.txt | 5 - examples/canonical-metrics/runner.py | 55 ----- examples/canonical-metrics/server.py | 139 ------------ pyproject.toml | 1 - src/pipecat/services/canonical/__init__.py | 13 -- src/pipecat/services/canonical/metrics.py | 230 -------------------- 14 files changed, 5 insertions(+), 834 deletions(-) delete mode 100644 examples/canonical-metrics/.gitignore delete mode 100644 examples/canonical-metrics/Dockerfile delete mode 100644 examples/canonical-metrics/README.md delete mode 100644 examples/canonical-metrics/bot.py delete mode 100644 examples/canonical-metrics/env.example delete mode 100644 examples/canonical-metrics/requirements.txt delete mode 100644 examples/canonical-metrics/runner.py delete mode 100644 examples/canonical-metrics/server.py delete mode 100644 src/pipecat/services/canonical/__init__.py delete mode 100644 src/pipecat/services/canonical/metrics.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 1a6274667..101cc7f58 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -41,6 +41,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Remove custom audio tracks from `DailyTransport` before leaving. +### Removed + +- Removed `CanonicalMetricsService` as it's no longer maintained. + ## [0.0.66] - 2025-05-02 ### Added diff --git a/README.md b/README.md index ac2444f87..7f95bb664 100644 --- a/README.md +++ b/README.md @@ -60,7 +60,7 @@ You can connect to Pipecat from any platform using our official SDKs: | Memory | [mem0](https://docs.pipecat.ai/server/services/memory/mem0) | | Vision & Image | [fal](https://docs.pipecat.ai/server/services/image-generation/fal), [Google Imagen](https://docs.pipecat.ai/server/services/image-generation/fal), [Moondream](https://docs.pipecat.ai/server/services/vision/moondream) | | Audio Processing | [Silero VAD](https://docs.pipecat.ai/server/utilities/audio/silero-vad-analyzer), [Krisp](https://docs.pipecat.ai/server/utilities/audio/krisp-filter), [Koala](https://docs.pipecat.ai/server/utilities/audio/koala-filter), [Noisereduce](https://docs.pipecat.ai/server/utilities/audio/noisereduce-filter) | -| Analytics & Metrics | [Canonical AI](https://docs.pipecat.ai/server/services/analytics/canonical), [Sentry](https://docs.pipecat.ai/server/services/analytics/sentry) | +| Analytics & Metrics | [Sentry](https://docs.pipecat.ai/server/services/analytics/sentry) | 📚 [View full services documentation →](https://docs.pipecat.ai/server/services/supported-services) diff --git a/docs/api/requirements.txt b/docs/api/requirements.txt index 9badccd8f..a77ff1084 100644 --- a/docs/api/requirements.txt +++ b/docs/api/requirements.txt @@ -10,7 +10,6 @@ pipecat-ai[anthropic] pipecat-ai[assemblyai] pipecat-ai[aws] pipecat-ai[azure] -pipecat-ai[canonical] pipecat-ai[cartesia] pipecat-ai[cerebras] pipecat-ai[deepseek] diff --git a/examples/canonical-metrics/.gitignore b/examples/canonical-metrics/.gitignore deleted file mode 100644 index 50d9d205e..000000000 --- a/examples/canonical-metrics/.gitignore +++ /dev/null @@ -1,161 +0,0 @@ -# Byte-compiled / optimized / DLL files -__pycache__/ -*.py[cod] -*$py.class -recordings/ -# C extensions -*.so - -# Distribution / packaging -.Python -build/ -develop-eggs/ -dist/ -downloads/ -eggs/ -.eggs/ -lib/ -lib64/ -parts/ -sdist/ -var/ -wheels/ -share/python-wheels/ -*.egg-info/ -.installed.cfg -*.egg -MANIFEST - -# PyInstaller -# Usually these files are written by a python script from a template -# before PyInstaller builds the exe, so as to inject date/other infos into it. -*.manifest -*.spec - -# Installer logs -pip-log.txt -pip-delete-this-directory.txt - -# Unit test / coverage reports -htmlcov/ -.tox/ -.nox/ -.coverage -.coverage.* -.cache -nosetests.xml -coverage.xml -*.cover -*.py,cover -.hypothesis/ -.pytest_cache/ -cover/ - -# Translations -*.mo -*.pot - -# Django stuff: -*.log -local_settings.py -db.sqlite3 -db.sqlite3-journal - -# Flask stuff: -instance/ -.webassets-cache - -# Scrapy stuff: -.scrapy - -# Sphinx documentation -docs/_build/ - -# PyBuilder -.pybuilder/ -target/ - -# Jupyter Notebook -.ipynb_checkpoints - -# IPython -profile_default/ -ipython_config.py - -# pyenv -# For a library or package, you might want to ignore these files since the code is -# intended to run in multiple environments; otherwise, check them in: -# .python-version - -# pipenv -# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. -# However, in case of collaboration, if having platform-specific dependencies or dependencies -# having no cross-platform support, pipenv may install dependencies that don't work, or not -# install all needed dependencies. -#Pipfile.lock - -# poetry -# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control. -# This is especially recommended for binary packages to ensure reproducibility, and is more -# commonly ignored for libraries. -# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control -#poetry.lock - -# pdm -# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control. -#pdm.lock -# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it -# in version control. -# https://pdm.fming.dev/#use-with-ide -.pdm.toml - -# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm -__pypackages__/ - -# Celery stuff -celerybeat-schedule -celerybeat.pid - -# SageMath parsed files -*.sage.py - -# Environments -.env -.venv -env/ -venv/ -ENV/ -env.bak/ -venv.bak/ - -# Spyder project settings -.spyderproject -.spyproject - -# Rope project settings -.ropeproject - -# mkdocs documentation -/site - -# mypy -.mypy_cache/ -.dmypy.json -dmypy.json - -# Pyre type checker -.pyre/ - -# pytype static type analyzer -.pytype/ - -# Cython debug symbols -cython_debug/ - -# PyCharm -# JetBrains specific template is maintained in a separate JetBrains.gitignore that can -# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore -# and can be added to the global gitignore or merged into this file. For a more nuclear -# option (not recommended) you can uncomment the following to ignore the entire idea folder. -#.idea/ -runpod.toml diff --git a/examples/canonical-metrics/Dockerfile b/examples/canonical-metrics/Dockerfile deleted file mode 100644 index a5b4668c6..000000000 --- a/examples/canonical-metrics/Dockerfile +++ /dev/null @@ -1,10 +0,0 @@ -FROM python:3.10-bullseye -RUN mkdir /app -COPY *.py /app/ -COPY requirements.txt /app/ -WORKDIR /app -RUN pip3 install -r requirements.txt - -EXPOSE 7860 - -CMD ["python3", "server.py"] diff --git a/examples/canonical-metrics/README.md b/examples/canonical-metrics/README.md deleted file mode 100644 index 068655d2b..000000000 --- a/examples/canonical-metrics/README.md +++ /dev/null @@ -1,66 +0,0 @@ -# Chatbot with canonical-metrics - -This project implements a chatbot using a pipeline architecture that integrates audio processing, transcription, and a language model for conversational interactions. The chatbot operates within a daily communication environment, utilizing various services for text-to-speech and language model responses. - -## Features - -- **Audio Input and Output**: Captures microphone input and plays back audio responses. -- **Voice Activity Detection**: Utilizes Silero VAD to manage audio input intelligently. -- **Text-to-Speech**: Integrates ElevenLabs TTS service to convert text responses into audio. -- **Language Model Interaction**: Uses OpenAI's GPT-4 model to generate responses based on user input. -- **Transcription Services**: Captures and transcribes participant speech for analytics. -- **Metrics Collection**: Sends audio data for analysis via Canonical Metrics Service. - -## Requirements - -- Python 3.10+ -- `python-dotenv` -- Additional libraries from the `pipecat` package. - -## Setup - -1. Clone the repository. -2. Install the required packages. -3. Set up environment variables for API keys: - - `OPENAI_API_KEY` - - `ELEVENLABS_API_KEY` - - `CANONICAL_API_KEY` - - `CANONICAL_API_URL` -4. Run the script. - -## Usage - -The chatbot introduces itself and engages in conversations, providing brief and creative responses. Designed for flexibility, it can support multiple languages with appropriate configuration. - -## Events - -- Participants joining or leaving the call are handled dynamically, adjusting the chatbot's behavior accordingly. - - -ℹ️ The first time, things might take extra time to get started since VAD (Voice Activity Detection) model needs to be downloaded. - -## Get started - -```python -python3 -m venv venv -source venv/bin/activate -pip install -r requirements.txt - -cp env.example .env # and add your credentials - -``` - -## Run the server - -```bash -python server.py -``` - -Then, visit `http://localhost:7860/` in your browser to start a chatbot session. - -## Build and test the Docker image - -``` -docker build -t chatbot . -docker run --env-file .env -p 7860:7860 chatbot -``` diff --git a/examples/canonical-metrics/bot.py b/examples/canonical-metrics/bot.py deleted file mode 100644 index 871d0542d..000000000 --- a/examples/canonical-metrics/bot.py +++ /dev/null @@ -1,146 +0,0 @@ -# -# Copyright (c) 2024–2025, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -import asyncio -import os -import sys -import uuid - -import aiohttp -from dotenv import load_dotenv -from loguru import logger -from runner import configure - -from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame -from pipecat.pipeline.pipeline import Pipeline -from pipecat.pipeline.runner import PipelineRunner -from pipecat.pipeline.task import PipelineParams, PipelineTask -from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext -from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor -from pipecat.services.canonical.metrics import CanonicalMetricsService -from pipecat.services.elevenlabs.tts import ElevenLabsTTSService -from pipecat.services.openai.llm import OpenAILLMService -from pipecat.transports.services.daily import DailyParams, DailyTransport - -load_dotenv(override=True) - -logger.remove(0) -logger.add(sys.stderr, level="DEBUG") - - -async def main(): - async with aiohttp.ClientSession() as session: - (room_url, token) = await configure(session) - - transport = DailyTransport( - room_url, - token, - "Chatbot", - DailyParams( - audio_out_enabled=True, - audio_in_enabled=True, - video_out_enabled=False, - vad_analyzer=SileroVADAnalyzer(), - transcription_enabled=True, - # - # Spanish - # - # transcription_settings=DailyTranscriptionSettings( - # language="es", - # tier="nova", - # model="2-general" - # ) - ), - ) - - tts = ElevenLabsTTSService( - api_key=os.getenv("ELEVENLABS_API_KEY"), - # - # English - # - voice_id="cgSgspJ2msm6clMCkdW9", - # - # Spanish - # - # model="eleven_multilingual_v2", - # voice_id="gD1IexrzCvsXPHUuT0s3", - ) - - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) - - messages = [ - { - "role": "system", - # - # English - # - "content": "You are Chatbot, a friendly, helpful robot. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way, but keep your responses brief. Start by introducing yourself. Keep all your responses to 12 words or fewer.", - # - # Spanish - # - # "content": "Eres Chatbot, un amigable y útil robot. Tu objetivo es demostrar tus capacidades de una manera breve. Tus respuestas se convertiran a audio así que nunca no debes incluir caracteres especiales. Contesta a lo que el usuario pregunte de una manera creativa, útil y breve. Empieza por presentarte a ti mismo.", - }, - ] - - context = OpenAILLMContext(messages) - context_aggregator = llm.create_context_aggregator(context) - - """ - CanonicalMetrics uses AudioBufferProcessor under the hood to buffer the audio. On - call completion, CanonicalMetrics will send the audio buffer to Canonical for - analysis. Visit https://voice.canonical.chat to learn more. - """ - audio_buffer_processor = AudioBufferProcessor(num_channels=2) - canonical = CanonicalMetricsService( - audio_buffer_processor=audio_buffer_processor, - aiohttp_session=session, - api_key=os.getenv("CANONICAL_API_KEY"), - call_id=str(uuid.uuid4()), - assistant="pipecat-chatbot", - assistant_speaks_first=True, - context=context, - ) - pipeline = Pipeline( - [ - transport.input(), # microphone - context_aggregator.user(), - llm, - tts, - transport.output(), - canonical, # uploads audio buffer to Canonical AI for metrics - audio_buffer_processor, # captures audio into a buffer - context_aggregator.assistant(), - ] - ) - - task = PipelineTask(pipeline, params=PipelineParams(allow_interruptions=True)) - - @transport.event_handler("on_first_participant_joined") - async def on_first_participant_joined(transport, participant): - await audio_buffer_processor.start_recording() - await transport.capture_participant_transcription(participant["id"]) - await task.queue_frames([context_aggregator.user().get_context_frame()]) - - @transport.event_handler("on_participant_left") - async def on_participant_left(transport, participant, reason): - print(f"Participant left: {participant}") - await task.cancel() - - @transport.event_handler("on_call_state_updated") - async def on_call_state_updated(transport, state): - if state == "left": - # Here we don't want to cancel, we just want to finish sending - # whatever is queued, so we use an EndFrame(). - await task.queue_frame(EndFrame()) - - runner = PipelineRunner() - - await runner.run(task) - - -if __name__ == "__main__": - asyncio.run(main()) diff --git a/examples/canonical-metrics/env.example b/examples/canonical-metrics/env.example deleted file mode 100644 index 6b865401a..000000000 --- a/examples/canonical-metrics/env.example +++ /dev/null @@ -1,6 +0,0 @@ -DAILY_SAMPLE_ROOM_URL=https://yourdomain.daily.co/yourroom # (for joining the bot to the same room repeatedly for local dev) -DAILY_API_KEY=7df... -OPENAI_API_KEY=sk-PL... -ELEVENLABS_API_KEY=aeb... -CANONICAL_API_KEY=can... -CANONICAL_API_URL= diff --git a/examples/canonical-metrics/requirements.txt b/examples/canonical-metrics/requirements.txt deleted file mode 100644 index 7e53edc6b..000000000 --- a/examples/canonical-metrics/requirements.txt +++ /dev/null @@ -1,5 +0,0 @@ -python-dotenv -fastapi[all] -uvicorn -pipecat-ai[daily,openai,silero,elevenlabs,canonical] - diff --git a/examples/canonical-metrics/runner.py b/examples/canonical-metrics/runner.py deleted file mode 100644 index ad39a3ac4..000000000 --- a/examples/canonical-metrics/runner.py +++ /dev/null @@ -1,55 +0,0 @@ -# -# Copyright (c) 2024–2025, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -import argparse -import os - -import aiohttp - -from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper - - -async def configure(aiohttp_session: aiohttp.ClientSession): - parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample") - parser.add_argument( - "-u", "--url", type=str, required=False, help="URL of the Daily room to join" - ) - parser.add_argument( - "-k", - "--apikey", - type=str, - required=False, - help="Daily API Key (needed to create an owner token for the room)", - ) - - args, unknown = parser.parse_known_args() - - url = args.url or os.getenv("DAILY_SAMPLE_ROOM_URL") - key = args.apikey or os.getenv("DAILY_API_KEY") - - if not url: - raise Exception( - "No Daily room specified. use the -u/--url option from the command line, or set DAILY_SAMPLE_ROOM_URL in your environment to specify a Daily room URL." - ) - - if not key: - raise Exception( - "No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers." - ) - - daily_rest_helper = DailyRESTHelper( - daily_api_key=key, - daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"), - aiohttp_session=aiohttp_session, - ) - - # Create a meeting token for the given room with an expiration 1 hour in - # the future. - expiry_time: float = 60 * 60 - - token = await daily_rest_helper.get_token(url, expiry_time) - - return (url, token) diff --git a/examples/canonical-metrics/server.py b/examples/canonical-metrics/server.py deleted file mode 100644 index a0f38854c..000000000 --- a/examples/canonical-metrics/server.py +++ /dev/null @@ -1,139 +0,0 @@ -# -# Copyright (c) 2024–2025, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -import argparse -import os -import subprocess -from contextlib import asynccontextmanager - -import aiohttp -from dotenv import load_dotenv -from fastapi import FastAPI, HTTPException, Request -from fastapi.middleware.cors import CORSMiddleware -from fastapi.responses import JSONResponse, RedirectResponse - -from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper, DailyRoomParams - -MAX_BOTS_PER_ROOM = 1 - -# Bot sub-process dict for status reporting and concurrency control -bot_procs = {} - -daily_helpers = {} - -load_dotenv(override=True) - - -def cleanup(): - # Clean up function, just to be extra safe - for entry in bot_procs.values(): - proc = entry[0] - proc.terminate() - proc.wait() - - -@asynccontextmanager -async def lifespan(app: FastAPI): - aiohttp_session = aiohttp.ClientSession() - daily_helpers["rest"] = DailyRESTHelper( - daily_api_key=os.getenv("DAILY_API_KEY", ""), - daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"), - aiohttp_session=aiohttp_session, - ) - yield - await aiohttp_session.close() - cleanup() - - -app = FastAPI(lifespan=lifespan) - -app.add_middleware( - CORSMiddleware, - allow_origins=["*"], - allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], -) - - -@app.get("/") -async def start_agent(request: Request): - print(f"!!! Creating room") - room = await daily_helpers["rest"].create_room(DailyRoomParams()) - print(f"!!! Room URL: {room.url}") - # Ensure the room property is present - if not room.url: - raise HTTPException( - status_code=500, - detail="Missing 'room' property in request data. Cannot start agent without a target room!", - ) - - # Check if there is already an existing process running in this room - num_bots_in_room = sum( - 1 for proc in bot_procs.values() if proc[1] == room.url and proc[0].poll() is None - ) - if num_bots_in_room >= MAX_BOTS_PER_ROOM: - raise HTTPException(status_code=500, detail=f"Max bot limited reach for room: {room.url}") - - # Get the token for the room - token = await daily_helpers["rest"].get_token(room.url) - - if not token: - raise HTTPException(status_code=500, detail=f"Failed to get token for room: {room.url}") - - # Spawn a new agent, and join the user session - # Note: this is mostly for demonstration purposes (refer to 'deployment' in README) - try: - proc = subprocess.Popen( - [f"python3 -m bot -u {room.url} -t {token}"], - shell=True, - bufsize=1, - cwd=os.path.dirname(os.path.abspath(__file__)), - ) - bot_procs[proc.pid] = (proc, room.url) - except Exception as e: - raise HTTPException(status_code=500, detail=f"Failed to start subprocess: {e}") - - return RedirectResponse(room.url) - - -@app.get("/status/{pid}") -def get_status(pid: int): - # Look up the subprocess - proc = bot_procs.get(pid) - - # If the subprocess doesn't exist, return an error - if not proc: - raise HTTPException(status_code=404, detail=f"Bot with process id: {pid} not found") - - # Check the status of the subprocess - if proc[0].poll() is None: - status = "running" - else: - status = "finished" - - return JSONResponse({"bot_id": pid, "status": status}) - - -if __name__ == "__main__": - import uvicorn - - default_host = os.getenv("HOST", "0.0.0.0") - default_port = int(os.getenv("FAST_API_PORT", "7860")) - - parser = argparse.ArgumentParser(description="Daily Storyteller FastAPI server") - parser.add_argument("--host", type=str, default=default_host, help="Host address") - parser.add_argument("--port", type=int, default=default_port, help="Port number") - parser.add_argument("--reload", action="store_true", help="Reload code on change") - - config = parser.parse_args() - - uvicorn.run( - "server:app", - host=config.host, - port=config.port, - reload=config.reload, - ) diff --git a/pyproject.toml b/pyproject.toml index ecddb0902..910c8d066 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -43,7 +43,6 @@ anthropic = [ "anthropic~=0.49.0" ] assemblyai = [ "assemblyai~=0.37.0" ] aws = [ "boto3~=1.37.16" ] azure = [ "azure-cognitiveservices-speech~=1.42.0"] -canonical = [ "aiofiles~=24.1.0" ] cartesia = [ "cartesia~=1.4.0", "websockets~=13.1" ] cerebras = [] deepseek = [] diff --git a/src/pipecat/services/canonical/__init__.py b/src/pipecat/services/canonical/__init__.py deleted file mode 100644 index f47b99c4e..000000000 --- a/src/pipecat/services/canonical/__init__.py +++ /dev/null @@ -1,13 +0,0 @@ -# -# Copyright (c) 2024–2025, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -import sys - -from pipecat.services import DeprecatedModuleProxy - -from .metrics import * - -sys.modules[__name__] = DeprecatedModuleProxy(globals(), "canonical", "canonical.metrics") diff --git a/src/pipecat/services/canonical/metrics.py b/src/pipecat/services/canonical/metrics.py deleted file mode 100644 index 012cd4ab7..000000000 --- a/src/pipecat/services/canonical/metrics.py +++ /dev/null @@ -1,230 +0,0 @@ -# -# Copyright (c) 2024–2025, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -import io -import os -import uuid -import wave -from datetime import datetime -from typing import Dict, List, Optional, Tuple - -import aiohttp -from loguru import logger - -from pipecat.frames.frames import CancelFrame, EndFrame, Frame -from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext -from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor -from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.ai_service import AIService - -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}") - - -# Multipart upload part size in bytes, cannot be smaller than 5MB -PART_SIZE = 1024 * 1024 * 5 - - -class CanonicalMetricsService(AIService): - """Initialize a CanonicalAudioProcessor instance. - - 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. - 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. - """ - - def __init__( - self, - *, - aiohttp_session: aiohttp.ClientSession, - 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", - audio_buffer_processor: Optional[AudioBufferProcessor] = None, - context: Optional[OpenAILLMContext] = None, - **kwargs, - ): - super().__init__(**kwargs) - # Validate that at least one of audio_buffer_processor or context is provided - if audio_buffer_processor is None and context is None: - raise ValueError("At least one of audio_buffer_processor or context must be specified") - - 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 - self._context = context - - async def stop(self, frame: EndFrame): - await super().stop(frame) - await self._process_completion() - - async def cancel(self, frame: CancelFrame): - await super().cancel(frame) - await self._process_completion() - - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - await self.push_frame(frame, direction) - - async def _process_completion(self): - if self._audio_buffer_processor is not None: - await self._process_audio() - elif self._context is not None: - await self._process_transcript() - - async def _process_transcript(self): - params = { - "callId": self._call_id, - "assistant": {"id": self._assistant, "speaksFirst": self._assistant_speaks_first}, - "transcript": self._context.messages, - } - response = await self._aiohttp_session.post( - f"{self._api_url}/call", - headers=self._request_headers(), - json=params, - ) - if not response.ok: - logger.error(f"Failed to process transcript: {await response.text()}") - - async def _process_audio(self): - audio_buffer_processor = self._audio_buffer_processor - - if not audio_buffer_processor.has_audio(): - return - - os.makedirs(self._output_dir, exist_ok=True) - filename = self._get_output_filename() - audio = audio_buffer_processor.merge_audio_buffers() - - with io.BytesIO() as buffer: - with wave.open(buffer, "wb") as wf: - wf.setsampwidth(2) - wf.setnchannels(audio_buffer_processor.num_channels) - wf.setframerate(audio_buffer_processor.sample_rate) - wf.writeframes(audio) - async with aiofiles.open(filename, "wb") as file: - await file.write(buffer.getvalue()) - - try: - await self._multipart_upload(filename) - 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}, - } - if self._context is not None: - params["transcript"] = self._context.messages - - 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