From 79c8aa2c4a089fe3ad3ccbd9199b817467465ff2 Mon Sep 17 00:00:00 2001 From: Adrian Cowham Date: Fri, 11 Oct 2024 11:35:02 -0700 Subject: [PATCH] ruff formatting --- examples/canonical-metrics/bot.py | 38 ++++----- examples/canonical-metrics/runner.py | 20 +++-- examples/canonical-metrics/server.py | 44 ++++------ examples/chatbot-audio-recording/bot.py | 36 ++++---- examples/chatbot-audio-recording/runner.py | 20 +++-- examples/chatbot-audio-recording/server.py | 44 ++++------ .../audio/audio_buffer_processor.py | 40 ++++----- src/pipecat/services/canonical.py | 85 ++++++++----------- 8 files changed, 147 insertions(+), 180 deletions(-) diff --git a/examples/canonical-metrics/bot.py b/examples/canonical-metrics/bot.py index 5b46b50fc..efad7710c 100644 --- a/examples/canonical-metrics/bot.py +++ b/examples/canonical-metrics/bot.py @@ -19,9 +19,10 @@ from pipecat.pipeline.pipeline import Pipeline 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.audio_buffer_processor import \ - AudioBufferProcessor + LLMAssistantResponseAggregator, + LLMUserResponseAggregator, +) +from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor from pipecat.services.canonical import CanonicalMetricsService from pipecat.services.elevenlabs import ElevenLabsTTSService from pipecat.services.openai import OpenAILLMService @@ -58,18 +59,16 @@ async def main(): # tier="nova", # model="2-general" # ) - ) + ), ) tts = ElevenLabsTTSService( api_key=os.getenv("ELEVENLABS_API_KEY"), - # # English # voice_id="cgSgspJ2msm6clMCkdW9", aiohttp_session=session, - # # Spanish # @@ -77,9 +76,7 @@ async def main(): # voice_id="gD1IexrzCvsXPHUuT0s3", ) - llm = OpenAILLMService( - api_key=os.getenv("OPENAI_API_KEY"), - model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") messages = [ { @@ -88,7 +85,6 @@ async def main(): # 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 # @@ -114,16 +110,18 @@ async def main(): assistant="pipecat-chatbot", assistant_speaks_first=True, ) - pipeline = Pipeline([ - transport.input(), # microphone - user_response, - llm, - tts, - transport.output(), - audio_buffer_processor, # captures audio into a buffer - canonical, # uploads audio buffer to Canonical AI for metrics - assistant_response, - ]) + pipeline = Pipeline( + [ + transport.input(), # microphone + user_response, + llm, + tts, + transport.output(), + audio_buffer_processor, # captures audio into a buffer + canonical, # uploads audio buffer to Canonical AI for metrics + assistant_response, + ] + ) task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) diff --git a/examples/canonical-metrics/runner.py b/examples/canonical-metrics/runner.py index 7507d28d6..a0b46ca36 100644 --- a/examples/canonical-metrics/runner.py +++ b/examples/canonical-metrics/runner.py @@ -4,21 +4,19 @@ # SPDX-License-Identifier: BSD 2-Clause License # -import aiohttp 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") + "-u", "--url", type=str, required=False, help="URL of the Daily room to join" + ) parser.add_argument( "-k", "--apikey", @@ -34,15 +32,18 @@ async def configure(aiohttp_session: aiohttp.ClientSession): 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.") + "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.") + 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 + aiohttp_session=aiohttp_session, ) # Create a meeting token for the given room with an expiration 1 hour in @@ -52,3 +53,4 @@ async def configure(aiohttp_session: aiohttp.ClientSession): token = await daily_rest_helper.get_token(url, expiry_time) return (url, token) + return (url, token) diff --git a/examples/canonical-metrics/server.py b/examples/canonical-metrics/server.py index 2c717ffa2..62ce899be 100644 --- a/examples/canonical-metrics/server.py +++ b/examples/canonical-metrics/server.py @@ -15,8 +15,7 @@ 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) +from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper, DailyRoomParams MAX_BOTS_PER_ROOM = 1 @@ -41,13 +40,14 @@ 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 + 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( @@ -68,37 +68,34 @@ async def start_agent(request: Request): if not room.url: raise HTTPException( status_code=500, - detail="Missing 'room' property in request data. Cannot start agent without a target room!") + 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) + 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}") + 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}") + 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}" - ], + [f"python3 -m bot -u {room.url} -t {token}"], shell=True, bufsize=1, - cwd=os.path.dirname(os.path.abspath(__file__)) + 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}") + raise HTTPException(status_code=500, detail=f"Failed to start subprocess: {e}") return RedirectResponse(room.url) @@ -110,8 +107,7 @@ def get_status(pid: int): # 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") + 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: @@ -128,14 +124,10 @@ if __name__ == "__main__": 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") + 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() diff --git a/examples/chatbot-audio-recording/bot.py b/examples/chatbot-audio-recording/bot.py index 2d572fb90..6acdb08e6 100644 --- a/examples/chatbot-audio-recording/bot.py +++ b/examples/chatbot-audio-recording/bot.py @@ -18,9 +18,10 @@ from pipecat.pipeline.pipeline import Pipeline 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.audio_buffer_processor import \ - AudioBufferProcessor + LLMAssistantResponseAggregator, + LLMUserResponseAggregator, +) +from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor from pipecat.services.elevenlabs import ElevenLabsTTSService from pipecat.services.openai import OpenAILLMService from pipecat.transports.services.daily import DailyParams, DailyTransport @@ -56,18 +57,16 @@ async def main(): # tier="nova", # model="2-general" # ) - ) + ), ) tts = ElevenLabsTTSService( api_key=os.getenv("ELEVENLABS_API_KEY"), - # # English # voice_id="cgSgspJ2msm6clMCkdW9", aiohttp_session=session, - # # Spanish # @@ -75,9 +74,7 @@ async def main(): # voice_id="gD1IexrzCvsXPHUuT0s3", ) - llm = OpenAILLMService( - api_key=os.getenv("OPENAI_API_KEY"), - model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") messages = [ { @@ -86,7 +83,6 @@ async def main(): # 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 response to 12 words or fewer.", - # # Spanish # @@ -98,15 +94,17 @@ async def main(): assistant_response = LLMAssistantResponseAggregator() audiobuffer = AudioBufferProcessor() - pipeline = Pipeline([ - transport.input(), # microphone - user_response, - llm, - tts, - transport.output(), - audiobuffer, # used to buffer the audio in the pipeline - assistant_response, - ]) + pipeline = Pipeline( + [ + transport.input(), # microphone + user_response, + llm, + tts, + transport.output(), + audiobuffer, # used to buffer the audio in the pipeline + assistant_response, + ] + ) task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) diff --git a/examples/chatbot-audio-recording/runner.py b/examples/chatbot-audio-recording/runner.py index 7507d28d6..a0b46ca36 100644 --- a/examples/chatbot-audio-recording/runner.py +++ b/examples/chatbot-audio-recording/runner.py @@ -4,21 +4,19 @@ # SPDX-License-Identifier: BSD 2-Clause License # -import aiohttp 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") + "-u", "--url", type=str, required=False, help="URL of the Daily room to join" + ) parser.add_argument( "-k", "--apikey", @@ -34,15 +32,18 @@ async def configure(aiohttp_session: aiohttp.ClientSession): 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.") + "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.") + 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 + aiohttp_session=aiohttp_session, ) # Create a meeting token for the given room with an expiration 1 hour in @@ -52,3 +53,4 @@ async def configure(aiohttp_session: aiohttp.ClientSession): token = await daily_rest_helper.get_token(url, expiry_time) return (url, token) + return (url, token) diff --git a/examples/chatbot-audio-recording/server.py b/examples/chatbot-audio-recording/server.py index 2c717ffa2..62ce899be 100644 --- a/examples/chatbot-audio-recording/server.py +++ b/examples/chatbot-audio-recording/server.py @@ -15,8 +15,7 @@ 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) +from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper, DailyRoomParams MAX_BOTS_PER_ROOM = 1 @@ -41,13 +40,14 @@ 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 + 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( @@ -68,37 +68,34 @@ async def start_agent(request: Request): if not room.url: raise HTTPException( status_code=500, - detail="Missing 'room' property in request data. Cannot start agent without a target room!") + 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) + 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}") + 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}") + 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}" - ], + [f"python3 -m bot -u {room.url} -t {token}"], shell=True, bufsize=1, - cwd=os.path.dirname(os.path.abspath(__file__)) + 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}") + raise HTTPException(status_code=500, detail=f"Failed to start subprocess: {e}") return RedirectResponse(room.url) @@ -110,8 +107,7 @@ def get_status(pid: int): # 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") + 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: @@ -128,14 +124,10 @@ if __name__ == "__main__": 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") + 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() diff --git a/src/pipecat/processors/audio/audio_buffer_processor.py b/src/pipecat/processors/audio/audio_buffer_processor.py index dabc24e38..0c07d6815 100644 --- a/src/pipecat/processors/audio/audio_buffer_processor.py +++ b/src/pipecat/processors/audio/audio_buffer_processor.py @@ -1,19 +1,23 @@ import wave from io import BytesIO -from pipecat.frames.frames import (AudioRawFrame, BotInterruptionFrame, - BotStartedSpeakingFrame, - BotStoppedSpeakingFrame, Frame, - InputAudioRawFrame, OutputAudioRawFrame, - StartInterruptionFrame, - StopInterruptionFrame, - UserStartedSpeakingFrame, - UserStoppedSpeakingFrame) +from pipecat.frames.frames import ( + AudioRawFrame, + BotInterruptionFrame, + BotStartedSpeakingFrame, + BotStoppedSpeakingFrame, + Frame, + InputAudioRawFrame, + OutputAudioRawFrame, + StartInterruptionFrame, + StopInterruptionFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, +) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor class AudioBufferProcessor(FrameProcessor): - def __init__(self, **kwargs): """ Initialize the AudioBufferProcessor. @@ -33,16 +37,13 @@ class AudioBufferProcessor(FrameProcessor): self._sample_rate = None def _buffer_has_audio(self, buffer: bytearray): - return ( - buffer is not None and - len(buffer) > 0 - ) + return buffer is not None and len(buffer) > 0 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 + 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): @@ -51,13 +52,12 @@ class AudioBufferProcessor(FrameProcessor): def _merge_audio_buffers(self): with BytesIO() as buffer: - with wave.open(buffer, 'wb') as wf: + with wave.open(buffer, "wb") as wf: wf.setnchannels(2) wf.setsampwidth(2) wf.setframerate(self._sample_rate) # Interleave the two audio streams - max_length = max(len(self._user_audio_buffer), - len(self._assistant_audio_buffer)) + max_length = max(len(self._user_audio_buffer), len(self._assistant_audio_buffer)) interleaved = bytearray(max_length * 2) for i in range(0, max_length, 2): @@ -89,7 +89,7 @@ class AudioBufferProcessor(FrameProcessor): # Sync the assistant's buffer to the user's buffer by adding silence if needed if len(self._user_audio_buffer) > len(self._assistant_audio_buffer): silence_length = len(self._user_audio_buffer) - len(self._assistant_audio_buffer) - silence = b'\x00' * silence_length + silence = b"\x00" * silence_length self._assistant_audio_buffer.extend(silence) # if the assistant is speaking, include all audio from the assistant, diff --git a/src/pipecat/services/canonical.py b/src/pipecat/services/canonical.py index 8fc415c95..412e06dab 100644 --- a/src/pipecat/services/canonical.py +++ b/src/pipecat/services/canonical.py @@ -12,14 +12,14 @@ try: 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.") + "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.audio.audio_buffer_processor import AudioBufferProcessor from pipecat.processors.frame_processor import FrameDirection from pipecat.services.ai_services import AIService @@ -56,15 +56,16 @@ class CanonicalMetricsService(AIService): """ 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"): + 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 @@ -92,7 +93,7 @@ class CanonicalMetricsService(AIService): filename = self._get_output_filename() wave_data = pipeline._merge_audio_buffers() - async with aiofiles.open(filename, 'wb') as file: + async with aiofiles.open(filename, "wb") as file: await file.write(wave_data) try: @@ -109,10 +110,7 @@ class CanonicalMetricsService(AIService): 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 - } + 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) @@ -129,19 +127,14 @@ class CanonicalMetricsService(AIService): 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 - } + "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 + 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()}") @@ -149,15 +142,11 @@ class CanonicalMetricsService(AIService): 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'] + 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: + 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: @@ -168,35 +157,29 @@ class CanonicalMetricsService(AIService): logger.error(f"Failed to upload part {partnum}: {await response.text()}") return None - etag = response.headers['ETag'] - parts.append({'partnum': str(partnum), 'etag': etag}) + 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): - + 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 - } + "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 + json=params, ) if not response.ok: logger.error(f"Failed to complete upload: {await response.text()}")