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

autopep8 formatting

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2024-03-18 11:28:32 -07:00
parent 2914e43350
commit 9385270775
44 changed files with 400 additions and 341 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/dailyai/services/ai_services.py
View File

@@ -28,6 +28,7 @@ class AIService(FrameProcessor):
def __init__(self):
self.logger = logging.getLogger("dailyai")
class LLMService(AIService):
"""This class is a no-op but serves as a base class for LLM services."""
@@ -76,7 +77,8 @@ class TTSService(AIService):
async for audio_chunk in self.run_tts(text):
yield AudioFrame(audio_chunk)
# note we pass along the text frame *after* the audio, so the text frame is completed after the audio is processed.
# note we pass along the text frame *after* the audio, so the text
# frame is completed after the audio is processed.
yield TextFrame(text)

6
src/dailyai/services/anthropic_llm_service.py
View File

@@ -9,7 +9,11 @@ from dailyai.services.ai_services import LLMService
class AnthropicLLMService(LLMService):
def __init__(self, api_key, model="claude-3-opus-20240229", max_tokens=1024):
def __init__(
self,
api_key,
model="claude-3-opus-20240229",
max_tokens=1024):
super().__init__()
self.client = AsyncAnthropic(api_key=api_key)
self.model = model

29
src/dailyai/services/azure_ai_services.py
View File

@@ -44,8 +44,7 @@ class AzureTTSService(TTSService):
"<mstts:express-as style='lyrical' styledegree='2' role='SeniorFemale'>"
"<prosody rate='1.05'>"
f"{sentence}"
"</prosody></mstts:express-as></voice></speak> "
)
"</prosody></mstts:express-as></voice></speak> ")
result = await asyncio.to_thread(self.speech_synthesizer.speak_ssml, (ssml))
self.logger.info("Got azure tts result")
if result.reason == ResultReason.SynthesizingAudioCompleted:
@@ -55,16 +54,22 @@ class AzureTTSService(TTSService):
elif result.reason == ResultReason.Canceled:
cancellation_details = result.cancellation_details
self.logger.info(
"Speech synthesis canceled: {}".format(cancellation_details.reason)
)
"Speech synthesis canceled: {}".format(
cancellation_details.reason))
if cancellation_details.reason == CancellationReason.Error:
self.logger.info(
"Error details: {}".format(cancellation_details.error_details)
)
"Error details: {}".format(
cancellation_details.error_details))
class AzureLLMService(BaseOpenAILLMService):
def __init__(self, *, api_key, endpoint, api_version="2023-12-01-preview", model):
def __init__(
self,
*,
api_key,
endpoint,
api_version="2023-12-01-preview",
model):
self._endpoint = endpoint
self._api_version = api_version
@@ -101,7 +106,9 @@ class AzureImageGenServiceREST(ImageGenService):
async def run_image_gen(self, sentence) -> tuple[str, bytes]:
url = f"{self._azure_endpoint}openai/images/generations:submit?api-version={self._api_version}"
headers = {"api-key": self._api_key, "Content-Type": "application/json"}
headers = {
"api-key": self._api_key,
"Content-Type": "application/json"}
body = {
# Enter your prompt text here
"prompt": sentence,
@@ -112,7 +119,8 @@ class AzureImageGenServiceREST(ImageGenService):
url, headers=headers, json=body
) as submission:
# We never get past this line, because this header isn't
# defined on a 429 response, but something is eating our exceptions!
# defined on a 429 response, but something is eating our
# exceptions!
operation_location = submission.headers["operation-location"]
status = ""
attempts_left = 120
@@ -130,8 +138,7 @@ class AzureImageGenServiceREST(ImageGenService):
status = json_response["status"]
image_url = (
json_response["result"]["data"][0]["url"] if json_response else None
)
json_response["result"]["data"][0]["url"] if json_response else None)
if not image_url:
raise Exception("Image generation failed")
# Load the image from the url

58
src/dailyai/services/base_transport_service.py
View File

@@ -127,12 +127,14 @@ class BaseTransportService:
self._logger: logging.Logger = logging.getLogger()
async def run(self, pipeline:Pipeline | None=None, override_pipeline_source_queue=True):
async def run(self, pipeline: Pipeline | None = None, override_pipeline_source_queue=True):
self._prerun()
async_output_queue_marshal_task = asyncio.create_task(self._marshal_frames())
async_output_queue_marshal_task = asyncio.create_task(
self._marshal_frames())
self._camera_thread = threading.Thread(target=self._run_camera, daemon=True)
self._camera_thread = threading.Thread(
target=self._run_camera, daemon=True)
self._camera_thread.start()
self._frame_consumer_thread = threading.Thread(
@@ -182,7 +184,7 @@ class BaseTransportService:
if self._vad_enabled:
self._vad_thread.join()
async def run_pipeline(self, pipeline:Pipeline, override_pipeline_source_queue=True):
async def run_pipeline(self, pipeline: Pipeline, override_pipeline_source_queue=True):
pipeline.set_sink(self.send_queue)
if override_pipeline_source_queue:
pipeline.set_source(self.receive_queue)
@@ -217,7 +219,8 @@ class BaseTransportService:
break
if post_processor:
post_process_task = asyncio.create_task(post_process(post_processor))
post_process_task = asyncio.create_task(
post_process(post_processor))
started = False
@@ -244,7 +247,7 @@ class BaseTransportService:
await asyncio.gather(pipeline_task, post_process_task)
async def say(self, text:str, tts:TTSService):
async def say(self, text: str, tts: TTSService):
"""Say a phrase. Use with caution; this bypasses any running pipelines."""
async for frame in tts.process_frame(TextFrame(text)):
await self.send_queue.put(frame)
@@ -290,7 +293,8 @@ class BaseTransportService:
audio_chunk = self.read_audio_frames(self._vad_samples)
audio_int16 = np.frombuffer(audio_chunk, np.int16)
audio_float32 = int2float(audio_int16)
new_confidence = model(torch.from_numpy(audio_float32), 16000).item()
new_confidence = model(
torch.from_numpy(audio_float32), 16000).item()
speaking = new_confidence > 0.5
if speaking:
@@ -320,8 +324,8 @@ class BaseTransportService:
):
if self._loop:
asyncio.run_coroutine_threadsafe(
self.receive_queue.put(UserStartedSpeakingFrame()), self._loop
)
self.receive_queue.put(
UserStartedSpeakingFrame()), self._loop)
# self.interrupt()
self._vad_state = VADState.SPEAKING
self._vad_starting_count = 0
@@ -331,8 +335,8 @@ class BaseTransportService:
):
if self._loop:
asyncio.run_coroutine_threadsafe(
self.receive_queue.put(UserStoppedSpeakingFrame()), self._loop
)
self.receive_queue.put(
UserStoppedSpeakingFrame()), self._loop)
self._vad_state = VADState.QUIET
self._vad_stopping_count = 0
@@ -370,7 +374,9 @@ class BaseTransportService:
self.receive_queue.put(frame), self._loop
)
asyncio.run_coroutine_threadsafe(self.receive_queue.put(EndFrame()), self._loop)
asyncio.run_coroutine_threadsafe(
self.receive_queue.put(
EndFrame()), self._loop)
def _set_image(self, image: bytes):
self._images = itertools.cycle([image])
@@ -378,7 +384,7 @@ class BaseTransportService:
def _set_images(self, images: list[bytes], start_frame=0):
self._images = itertools.cycle(images)
def send_app_message(self, message: Any, participantId:str|None):
def send_app_message(self, message: Any, participantId: str | None):
""" Child classes should override this to send a custom message to the room. """
pass
@@ -401,17 +407,18 @@ class BaseTransportService:
largest_write_size = 8000
while True:
try:
frames_or_frame: Frame | list[Frame] = self._threadsafe_send_queue.get()
frames_or_frame: Frame | list[Frame] = self._threadsafe_send_queue.get(
)
if (
isinstance(frames_or_frame, AudioFrame)
and len(frames_or_frame.data) > largest_write_size
):
# subdivide large audio frames to enable interruption
frames = []
for i in range(0, len(frames_or_frame.data), largest_write_size):
frames.append(
AudioFrame(frames_or_frame.data[i : i + largest_write_size])
)
for i in range(0, len(frames_or_frame.data),
largest_write_size):
frames.append(AudioFrame(
frames_or_frame.data[i: i + largest_write_size]))
elif isinstance(frames_or_frame, Frame):
frames: list[Frame] = [frames_or_frame]
elif isinstance(frames_or_frame, list):
@@ -430,7 +437,8 @@ class BaseTransportService:
)
return
# if interrupted, we just pull frames off the queue and discard them
# if interrupted, we just pull frames off the queue and
# discard them
if not self._is_interrupted.is_set():
if frame:
if isinstance(frame, AudioFrame):
@@ -441,14 +449,16 @@ class BaseTransportService:
len(b) % smallest_write_size
)
if truncated_length:
self.write_frame_to_mic(bytes(b[:truncated_length]))
self.write_frame_to_mic(
bytes(b[:truncated_length]))
b = b[truncated_length:]
elif isinstance(frame, ImageFrame):
self._set_image(frame.image)
elif isinstance(frame, SpriteFrame):
self._set_images(frame.images)
elif isinstance(frame, SendAppMessageFrame):
self.send_app_message(frame.message, frame.participantId)
self.send_app_message(
frame.message, frame.participantId)
elif len(b):
self.write_frame_to_mic(bytes(b))
b = bytearray()
@@ -457,7 +467,8 @@ class BaseTransportService:
# can cause static in the audio stream.
if len(b):
truncated_length = len(b) - (len(b) % 160)
self.write_frame_to_mic(bytes(b[:truncated_length]))
self.write_frame_to_mic(
bytes(b[:truncated_length]))
b = bytearray()
if isinstance(frame, StartFrame):
@@ -479,5 +490,6 @@ class BaseTransportService:
b = bytearray()
except Exception as e:
self._logger.error(f"Exception in frame_consumer: {e}, {len(b)}")
self._logger.error(
f"Exception in frame_consumer: {e}, {len(b)}")
raise e

26
src/dailyai/services/daily_transport_service.py
View File

@@ -48,7 +48,8 @@ class DailyTransportService(BaseTransportService, EventHandler):
start_transcription: bool = False,
**kwargs,
):
super().__init__(**kwargs) # This will call BaseTransportService.__init__ method, not EventHandler
# This will call BaseTransportService.__init__ method, not EventHandler
super().__init__(**kwargs)
self._room_url: str = room_url
self._bot_name: str = bot_name
@@ -83,9 +84,11 @@ class DailyTransportService(BaseTransportService, EventHandler):
for handler in self._event_handlers[event_name]:
if inspect.iscoroutinefunction(handler):
if self._loop:
future = asyncio.run_coroutine_threadsafe(handler(*args, **kwargs), self._loop)
future = asyncio.run_coroutine_threadsafe(
handler(*args, **kwargs), self._loop)
# wait for the coroutine to finish. This will also raise any exceptions raised by the coroutine.
# wait for the coroutine to finish. This will also
# raise any exceptions raised by the coroutine.
future.result()
else:
raise Exception(
@@ -98,7 +101,8 @@ class DailyTransportService(BaseTransportService, EventHandler):
def add_event_handler(self, event_name: str, handler):
if not event_name.startswith("on_"):
raise Exception(f"Event handler {event_name} must start with 'on_'")
raise Exception(
f"Event handler {event_name} must start with 'on_'")
methods = inspect.getmembers(self, predicate=inspect.ismethod)
if event_name not in [method[0] for method in methods]:
@@ -111,7 +115,8 @@ class DailyTransportService(BaseTransportService, EventHandler):
handler, self)]
setattr(self, event_name, partial(self._patch_method, event_name))
else:
self._event_handlers[event_name].append(types.MethodType(handler, self))
self._event_handlers[event_name].append(
types.MethodType(handler, self))
def event_handler(self, event_name: str):
def decorator(handler):
@@ -148,8 +153,7 @@ class DailyTransportService(BaseTransportService, EventHandler):
if self._camera_enabled:
self.camera: VirtualCameraDevice = Daily.create_camera_device(
"camera", width=self._camera_width, height=self._camera_height, color_format="RGB"
)
"camera", width=self._camera_width, height=self._camera_height, color_format="RGB")
if self._speaker_enabled or self._vad_enabled:
self._speaker: VirtualSpeakerDevice = Daily.create_speaker_device(
@@ -249,7 +253,7 @@ class DailyTransportService(BaseTransportService, EventHandler):
if len(self.client.participants()) < self._min_others_count + 1:
self._stop_threads.set()
def on_app_message(self, message:Any, sender:str):
def on_app_message(self, message: Any, sender: str):
if self._loop:
frame = ReceivedAppMessageFrame(message, sender)
print(frame)
@@ -265,8 +269,10 @@ class DailyTransportService(BaseTransportService, EventHandler):
elif "session_id" in message:
participantId = message["session_id"]
if self._my_participant_id and participantId != self._my_participant_id:
frame = TranscriptionQueueFrame(message["text"], participantId, message["timestamp"])
asyncio.run_coroutine_threadsafe(self.receive_queue.put(frame), self._loop)
frame = TranscriptionQueueFrame(
message["text"], participantId, message["timestamp"])
asyncio.run_coroutine_threadsafe(
self.receive_queue.put(frame), self._loop)
def on_transcription_error(self, message):
self._logger.error(f"Transcription error: {message}")

4
src/dailyai/services/deepgram_ai_service.py
View File

@@ -25,7 +25,9 @@ class DeepgramAIService(TTSService):
self.logger.info(f"Running deepgram tts for {sentence}")
base_url = "https://api.beta.deepgram.com/v1/speak"
request_url = f"{base_url}?model={self._voice}&encoding=linear16&container=none&sample_rate={self._sample_rate}"
headers = {"authorization": f"token {self._api_key}", "Content-Type": "application/json"}
headers = {
"authorization": f"token {self._api_key}",
"Content-Type": "application/json"}
data = {"text": sentence}
async with self._aiohttp_session.post(

7
src/dailyai/services/deepgram_ai_services.py
View File

@@ -9,7 +9,12 @@ from dailyai.services.ai_services import TTSService
class DeepgramTTSService(TTSService):
def __init__(self, *, aiohttp_session, api_key, voice="alpha-asteria-en-v2"):
def __init__(
self,
*,
aiohttp_session,
api_key,
voice="alpha-asteria-en-v2"):
super().__init__()
self._voice = voice

4
src/dailyai/services/elevenlabs_ai_service.py
View File

@@ -28,7 +28,9 @@ class ElevenLabsTTSService(TTSService):
async def run_tts(self, sentence) -> AsyncGenerator[bytes, None]:
url = f"https://api.elevenlabs.io/v1/text-to-speech/{self._voice_id}/stream"
payload = {"text": sentence, "model_id": self._model}
querystring = {"output_format": "pcm_16000", "optimize_streaming_latency": 2}
querystring = {
"output_format": "pcm_16000",
"optimize_streaming_latency": 2}
headers = {
"xi-api-key": self._api_key,
"Content-Type": "application/json",

2
src/dailyai/services/fal_ai_services.py
View File

@@ -33,7 +33,7 @@ class FalImageGenService(ImageGenService):
def get_image_url(sentence, size):
handler = fal.apps.submit(
"110602490-fast-sdxl",
#"fal-ai/fast-sdxl",
# "fal-ai/fast-sdxl",
arguments={"prompt": sentence},
)
for event in handler.iter_events():

18
src/dailyai/services/local_transport_service.py
View File

@@ -15,13 +15,15 @@ class LocalTransportService(BaseTransportService):
self._tk_root = kwargs.get("tk_root") or None
if self._camera_enabled and not self._tk_root:
raise ValueError("If camera is enabled, a tkinter root must be provided")
raise ValueError(
"If camera is enabled, a tkinter root must be provided")
if self._speaker_enabled:
self._speaker_buffer_pending = bytearray()
async def _write_frame_to_tkinter(self, frame: bytes):
data = f"P6 {self._camera_width} {self._camera_height} 255 ".encode() + frame
data = f"P6 {self._camera_width} {self._camera_height} 255 ".encode() + \
frame
photo = tk.PhotoImage(
width=self._camera_width,
height=self._camera_height,
@@ -29,7 +31,8 @@ class LocalTransportService(BaseTransportService):
format="PPM")
self._image_label.config(image=photo)
# This holds a reference to the photo, preventing it from being garbage collected.
# This holds a reference to the photo, preventing it from being garbage
# collected.
self._image_label.image = photo # type: ignore
def write_frame_to_camera(self, frame: bytes):
@@ -61,8 +64,13 @@ class LocalTransportService(BaseTransportService):
if self._camera_enabled:
# Start with a neutral gray background.
array = np.ones((1024, 1024, 3)) * 128
data = f"P5 {1024} {1024} 255 ".encode() + array.astype(np.uint8).tobytes()
photo = tk.PhotoImage(width=1024, height=1024, data=data, format="PPM")
data = f"P5 {1024} {1024} 255 ".encode(
) + array.astype(np.uint8).tobytes()
photo = tk.PhotoImage(
width=1024,
height=1024,
data=data,
format="PPM")
self._image_label = tk.Label(self._tk_root, image=photo)
self._image_label.pack()

3
src/dailyai/services/openai_api_llm_service.py
View File

@@ -110,7 +110,8 @@ class BaseOpenAILLMService(LLMService):
yield LLMFunctionStartFrame(function_name=tool_call.function.name)
if tool_call.function and tool_call.function.arguments:
# Keep iterating through the response to collect all the argument fragments and
# yield a complete LLMFunctionCallFrame after run_llm_async completes
# yield a complete LLMFunctionCallFrame after run_llm_async
# completes
arguments += tool_call.function.arguments
elif chunk.choices[0].delta.content:
yield TextFrame(chunk.choices[0].delta.content)

16
src/dailyai/services/openai_llm_context.py
View File

@@ -16,7 +16,8 @@ class OpenAILLMContext:
tools: List[ChatCompletionToolParam] | NotGiven = NOT_GIVEN,
tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven = NOT_GIVEN
):
self.messages: List[ChatCompletionMessageParam] = messages if messages else []
self.messages: List[ChatCompletionMessageParam] = messages if messages else [
]
self.tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven = tool_choice
self.tools: List[ChatCompletionToolParam] | NotGiven = tools
@@ -25,13 +26,13 @@ class OpenAILLMContext:
context = OpenAILLMContext()
for message in messages:
context.add_message({
"content":message["content"],
"role":message["role"],
"name":message["name"] if "name" in message else message["role"]
"content": message["content"],
"role": message["role"],
"name": message["name"] if "name" in message else message["role"]
})
return context
#def __deepcopy__(self, memo):
# def __deepcopy__(self, memo):
def add_message(self, message: ChatCompletionMessageParam):
self.messages.append(message)
@@ -44,9 +45,10 @@ class OpenAILLMContext:
):
self.tool_choice = tool_choice
def set_tools(self, tools:List[ChatCompletionToolParam] | NotGiven = NOT_GIVEN):
def set_tools(
self,
tools: List[ChatCompletionToolParam] | NotGiven = NOT_GIVEN):
if tools != NOT_GIVEN and len(tools) == 0:
tools = NOT_GIVEN
self.tools = tools

8
src/dailyai/services/to_be_updated/cloudflare_ai_service.py
View File

@@ -17,7 +17,10 @@ class CloudflareAIService(AIService):
# base endpoint, used by the others
def run(self, model, input):
response = requests.post(f"{self.api_base_url}{model}", headers=self.headers, json=input)
response = requests.post(
f"{self.api_base_url}{model}",
headers=self.headers,
json=input)
return response.json()
# https://developers.cloudflare.com/workers-ai/models/llm/
@@ -41,7 +44,8 @@ class CloudflareAIService(AIService):
# https://developers.cloudflare.com/workers-ai/models/sentiment-analysis/
def run_text_sentiment(self, sentence):
return self.run("@cf/huggingface/distilbert-sst-2-int8", {"text": sentence})
return self.run("@cf/huggingface/distilbert-sst-2-int8",
{"text": sentence})
# https://developers.cloudflare.com/workers-ai/models/image-classification/
def run_image_classification(self, image_url):