diff --git a/.gitignore b/.gitignore index 7eaaa0ca0..85fe741a6 100644 --- a/.gitignore +++ b/.gitignore @@ -22,4 +22,6 @@ share/python-wheels/ *.egg-info/ .installed.cfg *.egg -MANIFEST \ No newline at end of file +MANIFEST + +.DS_Store \ No newline at end of file diff --git a/src/dailyai/services/ai_services.py b/src/dailyai/services/ai_services.py index d95bfe10d..8bc0b8e52 100644 --- a/src/dailyai/services/ai_services.py +++ b/src/dailyai/services/ai_services.py @@ -1,6 +1,8 @@ import logging from abc import abstractmethod +from collections.abc import AsyncGenerator + from dataclasses import dataclass from typing import Generator from PIL import Image @@ -13,18 +15,17 @@ class AIService: def close(self): pass - class LLMService(AIService): # Generate a set of responses to a prompt. Yields a list of responses. @abstractmethod - def run_llm_async( + async def run_llm_async( self, messages - ) -> Generator[str, None, None]: + ) -> AsyncGenerator[str, None, None]: pass # Generate a responses to a prompt. Returns the response @abstractmethod - def run_llm( + async def run_llm( self, messages ) -> str or None: pass @@ -38,14 +39,14 @@ class TTSService(AIService): # Converts the sentence to audio. Yields a list of audio frames that can # be sent to the microphone device @abstractmethod - def run_tts(self, sentence) -> Generator[bytes, None, None]: + async def run_tts(self, sentence) -> AsyncGenerator[bytes, None, None]: pass class ImageGenService(AIService): # Renders the image. Returns an Image object. @abstractmethod - def run_image_gen(self, sentence) -> tuple[str, Image.Image]: + async def run_image_gen(self, sentence) -> tuple[str, Image.Image]: pass diff --git a/src/dailyai/services/azure_ai_services.py b/src/dailyai/services/azure_ai_services.py index 968e8d3b1..cf5ba2898 100644 --- a/src/dailyai/services/azure_ai_services.py +++ b/src/dailyai/services/azure_ai_services.py @@ -1,11 +1,13 @@ -import json +import aiohttp +import asyncio import io +import json from openai import AzureOpenAI import os import requests -from typing import Generator +from collections.abc import AsyncGenerator from dailyai.services.ai_services import LLMService, TTSService, ImageGenService from PIL import Image @@ -23,7 +25,7 @@ class AzureTTSService(TTSService): self.speech_config = SpeechConfig(subscription=speech_key, region=speech_region) self.speech_synthesizer = SpeechSynthesizer(speech_config=self.speech_config, audio_config=None) - def run_tts(self, sentence) -> Generator[bytes, None, None]: + async def run_tts(self, sentence) -> AsyncGenerator[bytes, None, None]: self.logger.info("Running azure tts") ssml = "" \ @@ -33,7 +35,7 @@ class AzureTTSService(TTSService): "" \ f"{sentence}" \ " " - result = self.speech_synthesizer.speak_ssml(ssml) + result = await asyncio.to_thread(self.speech_synthesizer.speak_ssml, (ssml)) self.logger.info("Got azure tts result") if result.reason == ResultReason.SynthesizingAudioCompleted: self.logger.info("Returning result") @@ -65,7 +67,7 @@ class AzureLLMService(LLMService): model=self.model, ) - def run_llm_async(self, messages) -> Generator[str, None, None]: + async def run_llm_async(self, messages) -> AsyncGenerator[str, None, None]: messages_for_log = json.dumps(messages) self.logger.debug(f"Generating chat via azure: {messages_for_log}") @@ -78,7 +80,7 @@ class AzureLLMService(LLMService): if chunk.choices[0].delta.content: yield chunk.choices[0].delta.content - def run_llm(self, messages) -> str | None: + async def run_llm(self, messages) -> str | None: messages_for_log = json.dumps(messages) self.logger.debug(f"Generating chat via azure: {messages_for_log}") @@ -88,6 +90,49 @@ class AzureLLMService(LLMService): else: return None +class AzureImageGenServiceREST(ImageGenService): + + def __init__(self, api_key=None, azure_endpoint=None, api_version=None, model=None): + super().__init__() + self.api_key = api_key or os.getenv("AZURE_DALLE_KEY") + self.azure_endpoint = azure_endpoint or os.getenv("AZURE_DALLE_ENDPOINT") + self.api_version = api_version or "2023-06-01-preview" + self.model = model or os.getenv("AZURE_DALLE_DEPLOYMENT_ID") + + async def run_image_gen(self, sentence, size) -> tuple[str, Image.Image]: + # TODO hoist the session to app-level + async with aiohttp.ClientSession() as session: + url = f"{self.azure_endpoint}openai/images/generations:submit?api-version={self.api_version}" + headers= { "api-key": self.api_key, "Content-Type": "application/json" } + body = { + # Enter your prompt text here + "prompt": sentence, + "size": size, + "n": 1, + } + async with session.post(url, headers=headers, json=body) as submission: + operation_location = submission.headers['operation-location'] + + status = "" + attempts_left = 120 + while status != "succeeded": + attempts_left -= 1 + if attempts_left == 0: + raise Exception("Image generation timed out") + + await asyncio.sleep(1) + response = await session.get(operation_location, headers=headers) + json_response = await response.json() + status = json_response["status"] + + image_url = json_response["result"]["data"][0]["url"] + + # Load the image from the url + async with session.get(image_url) as response: + image_stream = io.BytesIO(await response.content.read()) + image = Image.open(image_stream) + return (image_url, image.tobytes()) + class AzureImageGenService(ImageGenService): @@ -96,7 +141,7 @@ class AzureImageGenService(ImageGenService): api_key = api_key or os.getenv("AZURE_DALLE_KEY") azure_endpoint = azure_endpoint or os.getenv("AZURE_DALLE_ENDPOINT") - api_version = api_version or "2023-12-01-preview" + api_version = api_version or "2023-06-01-preview" self.model = model or os.getenv("AZURE_DALLE_DEPLOYMENT_ID") self.client = AzureOpenAI( @@ -105,7 +150,7 @@ class AzureImageGenService(ImageGenService): api_version=api_version, ) - def run_image_gen(self, sentence) -> tuple[str, Image.Image]: + async def run_image_gen(self, sentence) -> tuple[str, Image.Image]: self.logger.info("Generating azure image", sentence) image = self.client.images.generate( diff --git a/src/dailyai/services/daily_transport_service.py b/src/dailyai/services/daily_transport_service.py new file mode 100644 index 000000000..63567b1b7 --- /dev/null +++ b/src/dailyai/services/daily_transport_service.py @@ -0,0 +1,267 @@ +import inspect +import logging +import time +import types + +from functools import partial +from queue import Queue, Empty + +from dailyai.output_queue import OutputQueueFrame, FrameType + +from threading import Thread, Event, Timer + +from daily import ( + EventHandler, + CallClient, + Daily, + VirtualCameraDevice, + VirtualMicrophoneDevice, + VirtualSpeakerDevice, +) + +class DailyTransportService(EventHandler): + def __init__( + self, + room_url: str, + token: str, + bot_name: str, + duration: float = 10, + ): + super().__init__() + self.bot_name: str = bot_name + self.room_url: str = room_url + self.token: str = token + self.duration: float = duration + self.expiration = time.time() + duration * 60 + + self.output_queue = Queue() + self.is_interrupted = Event() + self.stop_threads = Event() + self.story_started = False + + self.logger: logging.Logger = logging.getLogger("dailyai") + + self.event_handlers = {} + + def monkeypatch(self, event_name, *args): + for handler in self.event_handlers[event_name]: + handler(*args) + + 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_'") + + methods = inspect.getmembers(self, predicate=inspect.ismethod) + if event_name not in [method[0] for method in methods]: + raise Exception(f"Event handler {event_name} not found") + + if not event_name in self.event_handlers: + self.event_handlers[event_name] = [getattr(self, event_name), types.MethodType(handler, self)] + setattr(self, event_name, partial(self.monkeypatch, event_name)) + else: + self.event_handlers[event_name].append(types.MethodType(handler, self)) + + def configure_daily(self): + Daily.init() + self.client = CallClient(event_handler=self) + + if self.mic_enabled: + self.mic: VirtualMicrophoneDevice = Daily.create_microphone_device( + "mic", sample_rate=self.mic_sample_rate, channels=1 + ) + + if self.camera_enabled: + self.camera: VirtualCameraDevice = Daily.create_camera_device( + "camera", width=self.camera_width, height=self.camera_height, color_format="RGB" + ) + + self.speaker: VirtualSpeakerDevice = Daily.create_speaker_device( + "speaker", sample_rate=16000, channels=1 + ) + + Daily.select_speaker_device("speaker") + + self.client.set_user_name(self.bot_name) + self.client.join(self.room_url, self.token, completion=self.call_joined) + + self.client.update_inputs( + { + "camera": { + "isEnabled": True, + "settings": { + "deviceId": "camera", + }, + }, + "microphone": { + "isEnabled": True, + "settings": { + "deviceId": "mic", + "customConstraints": { + "autoGainControl": {"exact": False}, + "echoCancellation": {"exact": False}, + "noiseSuppression": {"exact": False}, + }, + }, + }, + } + ) + + self.client.update_publishing( + { + "camera": { + "sendSettings": { + "maxQuality": "low", + "encodings": { + "low": { + "maxBitrate": 250000, + "scaleResolutionDownBy": 1.333, + "maxFramerate": 8, + } + }, + } + } + } + ) + + self.my_participant_id = self.client.participants()["local"]["id"] + + def run(self) -> None: + self.configure_daily() + self.running_thread = Thread(target=self.run_daily, daemon=True) + self.running_thread.start() + + def run_daily(self): + # TODO: this loop could, I think, be replaced with a timer and an event + self.participant_left = False + + try: + participant_count: int = len(self.client.participants()) + self.logger.info(f"{participant_count} participants in room") + while time.time() < self.expiration and not self.participant_left: + # all handling of incoming transcriptions happens in on_transcription_message + time.sleep(1) + except Exception as e: + self.logger.error(f"Exception {e}") + finally: + self.client.leave() + + def stop(self): + self.stop_threads.set() + self.camera_thread.join() + self.output_queue.put(OutputQueueFrame(FrameType.END_STREAM, None)) + self.frame_consumer_thread.join() + self.client.leave() + + def call_joined(self, join_data, client_error): + self.logger.info(f"Call_joined: {join_data}, {client_error}") + + self.image: bytes | None = None + self.camera_thread = Thread(target=self.run_camera, daemon=True) + self.camera_thread.start() + + self.logger.info("Starting frame consumer thread") + self.frame_consumer_thread = Thread(target=self.frame_consumer, daemon=True) + self.frame_consumer_thread.start() + + if self.token: + self.client.start_transcription( + { + "language": "en", + "tier": "nova", + "model": "2-conversationalai", + "profanity_filter": True, + "redact": False, + "extra": { + "endpointing": True, + "punctuate": False, + }, + } + ) + + def on_participant_joined(self, participant): + pass + + def on_participant_left(self, participant, reason): + pass + + def on_app_message(self, message, sender): + pass + + def on_transcription_message(self, message): + with self.tracer.start_as_current_span( + "on_transcription_message", context=self.ctx + ): + if message["session_id"] != self.my_participant_id: + self.handle_transcription_fragment(message["text"]) + + def on_transcription_stopped(self, stopped_by, stopped_by_error): + self.logger.info(f"Transcription stopped {stopped_by}, {stopped_by_error}") + + def on_transcription_error(self, message): + self.logger.error(f"Transcription error {message}") + + def on_transcription_started(self, status): + self.logger.info(f"Transcription started {status}") + + def set_image(self, image: bytes): + self.image: bytes | None = image + + def run_camera(self): + try: + while not self.stop_threads.is_set(): + if self.image: + self.camera.write_frame(self.image) + + time.sleep(1.0 / 8.0) # 8 fps + except Exception as e: + self.logger.error(f"Exception {e} in camera thread.") + + def frame_consumer(self): + self.logger.info("🎬 Starting frame consumer thread") + b = bytearray() + smallest_write_size = 3200 + all_audio_frames = bytearray() + while True: + try: + frame: OutputQueueFrame = self.output_queue.get() + if frame.frame_type == FrameType.END_STREAM: + self.logger.info("Stopping frame consumer thread") + return + + # if interrupted, we just pull frames off the queue and discard them + if not self.is_interrupted.is_set(): + if frame: + if frame.frame_type == FrameType.AUDIO_FRAME: + chunk = frame.frame_data + + all_audio_frames.extend(chunk) + + b.extend(chunk) + l = len(b) - (len(b) % smallest_write_size) + if l: + self.mic.write_frames(bytes(b[:l])) + b = b[l:] + elif frame.frame_type == FrameType.IMAGE_FRAME: + self.set_image(frame.frame_data) + elif len(b): + self.mic.write_frames(bytes(b)) + b = bytearray() + else: + if self.interrupt_time: + self.logger.info( + f"Lag to stop stream after interruption {time.perf_counter() - self.interrupt_time}" + ) + self.interrupt_time = None + + if frame.frame_type == FrameType.START_STREAM: + self.is_interrupted.clear() + + self.output_queue.task_done() + except Empty: + try: + if len(b): + self.mic.write_frames(bytes(b)) + except Exception as e: + self.logger.error(f"Exception in frame_consumer: {e}, {len(b)}") + + b = bytearray() diff --git a/src/samples/05-sync-speech-and-text.py b/src/samples/05-sync-speech-and-text.py new file mode 100644 index 000000000..de3f0cc80 --- /dev/null +++ b/src/samples/05-sync-speech-and-text.py @@ -0,0 +1,60 @@ +import asyncio + +from dailyai.output_queue import OutputQueueFrame, FrameType +from dailyai.services.azure_ai_services import AzureLLMService, AzureTTSService, AzureImageGenServiceREST +from dailyai.services.daily_transport_service import DailyTransportService + +async def main(room_url, token): + class Sample05Transport(DailyTransportService): + def on_participant_joined(self, participant): + super().on_participant_joined(participant) + + meeting_duration_minutes = 4 + transport = Sample05Transport( + room_url, + token, + "Simple Bot", + meeting_duration_minutes, + ) + transport.mic_enabled = True + transport.camera_enabled = True + transport.mic_sample_rate = 16000 + transport.camera_width = 1024 + transport.camera_height = 1024 + + llm = AzureLLMService() + tts = AzureTTSService() + dalle = AzureImageGenServiceREST() + + inference_text_process = llm.run_llm( + [ + { + "role": "system", + "content": f"Describe a nature photograph suitable for use in a calendar, for the month of January. Include only the image description with no preamble." + } + ] + ) + + try: + transport.run() + + inference_text = await inference_text_process + + tts_iterator = tts.run_tts(inference_text) + (image, audio) = await asyncio.gather( + *[dalle.run_image_gen(inference_text, "1024x1024"), anext(tts_iterator)] + ) + transport.output_queue.put(OutputQueueFrame(FrameType.IMAGE_FRAME, image[1])) + transport.output_queue.put(OutputQueueFrame(FrameType.AUDIO_FRAME, audio)) + async for audio in tts_iterator: + transport.output_queue.put( + OutputQueueFrame(FrameType.AUDIO_FRAME, audio) + ) + + await asyncio.sleep(meeting_duration_minutes * 60) + finally: + transport.stop() + print("Done") + +if __name__=="__main__": + asyncio.run(main("https://moishe.daily.co/Lettvins", None))