diff --git a/output.avi b/output.avi new file mode 100644 index 000000000..5ca933ad6 Binary files /dev/null and b/output.avi differ diff --git a/pyproject.toml b/pyproject.toml index 348652cfb..7603f8500 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,18 +7,17 @@ name = "daily_ai" version = "0.0.1" description = "Orchestrator for AI bots with Daily" dependencies = [ - "daily-python", - "python-dotenv", - "Pillow", - "typing-extensions", - "openai", - "google-cloud-texttospeech", - "azure-cognitiveservices-speech", - "pyht", - "opentelemetry-sdk", "aiohttp", + "azure-cognitiveservices-speech", + "daily-python", "fal", - "faster_whisper" + "faster_whisper", + "google-cloud-texttospeech", + "openai", + "Pillow", + "pyht", + "python-dotenv", + "typing-extensions" ] [tool.setuptools.packages.find] diff --git a/src/dailyai/services/ai_services.py b/src/dailyai/services/ai_services.py index a7a301ca5..29b7d6488 100644 --- a/src/dailyai/services/ai_services.py +++ b/src/dailyai/services/ai_services.py @@ -1,6 +1,7 @@ import asyncio import io import logging +import time import wave from dailyai.queue_frame import ( @@ -12,6 +13,7 @@ from dailyai.queue_frame import ( LLMResponseEndQueueFrame, QueueFrame, TextQueueFrame, + TranscriptionQueueFrame, ) from abc import abstractmethod @@ -188,7 +190,8 @@ class STTService(AIService): ww.close() content.seek(0) text = await self.run_stt(content) - yield TextQueueFrame(text) + yield TranscriptionQueueFrame(text, '', str(time.time())) + class FrameLogger(AIService): def __init__(self, prefix="Frame", **kwargs): @@ -202,10 +205,3 @@ class FrameLogger(AIService): print(f"{self.prefix}: {frame}") yield frame - -@dataclass -class AIServiceConfig: - tts: TTSService - image: ImageGenService - llm: LLMService - stt: STTService diff --git a/src/dailyai/services/base_transport_service.py b/src/dailyai/services/base_transport_service.py new file mode 100644 index 000000000..139edc6ff --- /dev/null +++ b/src/dailyai/services/base_transport_service.py @@ -0,0 +1,235 @@ +from abc import abstractmethod +import asyncio +import itertools +import logging +import queue +import threading +import time +from typing import AsyncGenerator + +from dailyai.queue_frame import ( + AudioQueueFrame, + EndStreamQueueFrame, + ImageQueueFrame, + QueueFrame, + SpriteQueueFrame, + StartStreamQueueFrame, +) + +class BaseTransportService(): + + def __init__( + self, + **kwargs, + ) -> None: + self._mic_enabled = kwargs.get("mic_enabled") or False + self._mic_sample_rate = kwargs.get("mic_sample_rate") or 16000 + self._camera_enabled = kwargs.get("camera_enabled") or False + self._camera_width = kwargs.get("camera_width") or 1024 + self._camera_height = kwargs.get("camera_height") or 768 + self._speaker_enabled = kwargs.get("speaker_enabled") or False + self._speaker_sample_rate = kwargs.get("speaker_sample_rate") or 16000 + self._fps = kwargs.get("fps") or 8 + + duration_minutes = kwargs.get("duration_minutes") or 10 + self._expiration = time.time() + duration_minutes * 60 + + self.send_queue = asyncio.Queue() + self.receive_queue = asyncio.Queue() + + self._threadsafe_send_queue = queue.Queue() + + self._images = None + + try: + self._loop: asyncio.AbstractEventLoop | None = asyncio.get_running_loop() + except RuntimeError: + self._loop = None + + self._stop_threads = threading.Event() + self._is_interrupted = threading.Event() + + self._logger: logging.Logger = logging.getLogger() + + async def run(self): + self._prerun() + + 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.start() + + self._frame_consumer_thread = threading.Thread(target=self._frame_consumer, daemon=True) + self._frame_consumer_thread.start() + + if self._speaker_enabled: + self._receive_audio_thread = threading.Thread(target=self._receive_audio, daemon=True) + self._receive_audio_thread.start() + + try: + while ( + time.time() < self._expiration + and not self._stop_threads.is_set() + ): + await asyncio.sleep(1) + except Exception as e: + self._logger.error(f"Exception {e}") + raise e + + self._stop_threads.set() + + await self.send_queue.put(EndStreamQueueFrame()) + await async_output_queue_marshal_task + await self.send_queue.join() + self._frame_consumer_thread.join() + + if self._speaker_enabled: + self._receive_audio_thread.join() + + def stop(self): + self._stop_threads.set() + + async def stop_when_done(self): + await self._wait_for_send_queue_to_empty() + self.stop() + + async def _wait_for_send_queue_to_empty(self): + await self.send_queue.join() + self._threadsafe_send_queue.join() + + @abstractmethod + def write_frame_to_camera(self, frame: bytes): + pass + + @abstractmethod + def write_frame_to_mic(self, frame: bytes): + pass + + @abstractmethod + def read_audio_frames(self, desired_frame_count): + return bytes() + + @abstractmethod + def _prerun(self): + pass + + async def _marshal_frames(self): + while True: + frame: QueueFrame | list = await self.send_queue.get() + self._threadsafe_send_queue.put(frame) + self.send_queue.task_done() + if isinstance(frame, EndStreamQueueFrame): + break + + def interrupt(self): + self._is_interrupted.set() + + async def get_receive_frames(self) -> AsyncGenerator[QueueFrame, None]: + while True: + frame = await self.receive_queue.get() + yield frame + if isinstance(frame, EndStreamQueueFrame): + break + + def _receive_audio(self): + if not self._loop: + self._logger.error("No loop available for audio thread") + return + + seconds = 1 + desired_frame_count = self._speaker_sample_rate * seconds + while not self._stop_threads.is_set(): + buffer = self.read_audio_frames(desired_frame_count) + if len(buffer) > 0: + frame = AudioQueueFrame(buffer) + asyncio.run_coroutine_threadsafe( + self.receive_queue.put(frame), self._loop + ) + + asyncio.run_coroutine_threadsafe( + self.receive_queue.put(EndStreamQueueFrame()), self._loop + ) + + def _set_image(self, image: bytes): + self._images = itertools.cycle([image]) + + def _set_images(self, images: list[bytes], start_frame=0): + self._images = itertools.cycle(images) + + def _run_camera(self): + try: + while not self._stop_threads.is_set(): + if self._images: + this_frame = next(self._images) + self.write_frame_to_camera(this_frame) + + time.sleep(1.0 / self._fps) + except Exception as e: + self._logger.error(f"Exception {e} in camera thread.") + raise e + + def _frame_consumer(self): + self._logger.info("🎬 Starting frame consumer thread") + b = bytearray() + smallest_write_size = 3200 + all_audio_frames = bytearray() + while True: + try: + frames_or_frame: QueueFrame | list[QueueFrame] = ( + self._threadsafe_send_queue.get() + ) + if isinstance(frames_or_frame, QueueFrame): + frames: list[QueueFrame] = [frames_or_frame] + elif isinstance(frames_or_frame, list): + frames: list[QueueFrame] = frames_or_frame + else: + raise Exception("Unknown type in output queue") + + for frame in frames: + if isinstance(frame, EndStreamQueueFrame): + self._logger.info("Stopping frame consumer thread") + self._threadsafe_send_queue.task_done() + return + + # if interrupted, we just pull frames off the queue and discard them + if not self._is_interrupted.is_set(): + if frame: + if isinstance(frame, AudioQueueFrame): + chunk = frame.data + + all_audio_frames.extend(chunk) + + b.extend(chunk) + truncated_length: int = len(b) - ( + len(b) % smallest_write_size + ) + if truncated_length: + self.write_frame_to_mic(bytes(b[:truncated_length])) + b = b[truncated_length:] + elif isinstance(frame, ImageQueueFrame): + self._set_image(frame.image) + elif isinstance(frame, SpriteQueueFrame): + self._set_images(frame.images) + elif len(b): + self.write_frame_to_mic(bytes(b)) + b = bytearray() + else: + # if there are leftover audio bytes, write them now; failing to do so + # 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])) + b = bytearray() + + if isinstance(frame, StartStreamQueueFrame): + self._is_interrupted.clear() + + self._threadsafe_send_queue.task_done() + except queue.Empty: + if len(b): + self.write_frame_to_mic(bytes(b)) + + b = bytearray() + except Exception as e: + self._logger.error(f"Exception in frame_consumer: {e}, {len(b)}") + raise e diff --git a/src/dailyai/services/daily_transport_service.py b/src/dailyai/services/daily_transport_service.py index f253db9f6..575a6560b 100644 --- a/src/dailyai/services/daily_transport_service.py +++ b/src/dailyai/services/daily_transport_service.py @@ -1,22 +1,13 @@ import asyncio import inspect -import itertools import logging import threading import time import types from functools import partial -from queue import Queue, Empty -from typing import AsyncGenerator from dailyai.queue_frame import ( - AudioQueueFrame, - EndStreamQueueFrame, - ImageQueueFrame, - SpriteQueueFrame, - QueueFrame, - StartStreamQueueFrame, TranscriptionQueueFrame, ) @@ -31,59 +22,42 @@ from daily import ( VirtualSpeakerDevice, ) +from dailyai.services.base_transport_service import BaseTransportService -class DailyTransportService(EventHandler): + +class DailyTransportService(BaseTransportService, EventHandler): _daily_initialized = False _lock = threading.Lock() - speaker_enabled: bool - speaker_sample_rate: int + _speaker_enabled: bool + _speaker_sample_rate: int + + # This is necessary to override EventHandler's __new__ method. + def __new__(cls, *args, **kwargs): + return super().__new__(cls) def __init__( self, room_url: str, token: str | None, bot_name: str, - duration: float = 10, min_others_count: int = 1, - start_transcription: bool = True, - speaker_enabled: bool = False, - speaker_sample_rate: int = 16000, + start_transcription: bool = False, + **kwargs, ): - super().__init__() - self.bot_name: str = bot_name - self.room_url: str = room_url - self.token: str | None = token - self.duration: float = duration - self.expiration = time.time() + duration * 60 - self.min_others_count = min_others_count - self.start_transcription = start_transcription + super().__init__(**kwargs) # This will call BaseTransportService.__init__ method, not EventHandler - # This queue is used to marshal frames from the async send queue to the thread that emits audio & video. - # We need this to maintain the asynchronous behavior of asyncio queues -- to give async functions - # a chance to run while waiting for queue items -- but also to maintain thread safety and have a threaded - # handler to send frames, to ensure that sending isn't subject to pauses - # in the async thread. - self.threadsafe_send_queue = Queue() + self._room_url: str = room_url + self._bot_name: str = bot_name + self._token: str | None = token + self._min_others_count = min_others_count + self._start_transcription = start_transcription self._is_interrupted = Event() self._stop_threads = Event() - self.mic_enabled = False - self.mic_sample_rate = 16000 - self.camera_width = 1024 - self.camera_height = 768 - self.camera_enabled = False - self.speaker_enabled = speaker_enabled - self.speaker_sample_rate = speaker_sample_rate - - self.send_queue = asyncio.Queue() - self.receive_queue = asyncio.Queue() self._other_participant_has_joined = False - self.my_participant_id = None - - self._camera_thread = None - self._frame_consumer_thread = None + self._my_participant_id = None self.transcription_settings = { "language": "en", @@ -101,11 +75,6 @@ class DailyTransportService(EventHandler): self._event_handlers = {} - try: - self._loop = asyncio.get_running_loop() - except RuntimeError: - self._loop = None - def _patch_method(self, event_name, *args, **kwargs): try: for handler in self._event_handlers[event_name]: @@ -145,7 +114,17 @@ class DailyTransportService(EventHandler): return decorator - def _configure_daily(self): + def write_frame_to_camera(self, frame: bytes): + self.camera.write_frame(frame) + + def write_frame_to_mic(self, frame: bytes): + self.mic.write_frames(frame) + + def read_audio_frames(self, desired_frame_count): + bytes = self._speaker.read_frames(desired_frame_count) + return bytes + + def _prerun(self): # Only initialize Daily once if not DailyTransportService._daily_initialized: with DailyTransportService._lock: @@ -153,34 +132,25 @@ class DailyTransportService(EventHandler): DailyTransportService._daily_initialized = True self.client = CallClient(event_handler=self) - if self.mic_enabled: + if self._mic_enabled: self.mic: VirtualMicrophoneDevice = Daily.create_microphone_device( - "mic", sample_rate=self.mic_sample_rate, channels=1 + "mic", sample_rate=self._mic_sample_rate, channels=1 ) - if self.camera_enabled: + 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: - self.speaker: VirtualSpeakerDevice = Daily.create_speaker_device( - "speaker", sample_rate=self.speaker_sample_rate, channels=1 + if self._speaker_enabled: + self._speaker: VirtualSpeakerDevice = Daily.create_speaker_device( + "speaker", sample_rate=self._speaker_sample_rate, channels=1 ) Daily.select_speaker_device("speaker") - self._images = 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() - - self.client.set_user_name(self.bot_name) - self.client.join(self.room_url, self.token, completion=self.call_joined) - self.my_participant_id = self.client.participants()["local"]["id"] + self.client.set_user_name(self._bot_name) + self.client.join(self._room_url, self._token, completion=self.call_joined) + self._my_participant_id = self.client.participants()["local"]["id"] self.client.update_inputs( { @@ -221,89 +191,14 @@ class DailyTransportService(EventHandler): } ) - if self.token and self.start_transcription: + if self._token and self._start_transcription: self.client.start_transcription(self.transcription_settings) - def _receive_audio(self): - """Receive audio from the Daily call and put it on the receive queue""" - seconds = 1 - desired_frame_count = self.speaker_sample_rate * seconds - while True: - buffer = self.speaker.read_frames(desired_frame_count) - if len(buffer) > 0: - frame = AudioQueueFrame(buffer) - if self._loop: - asyncio.run_coroutine_threadsafe(self.receive_queue.put(frame), self._loop) - - def interrupt(self): - self._is_interrupted.set() - - async def get_receive_frames(self) -> AsyncGenerator[QueueFrame, None]: - while True: - frame = await self.receive_queue.get() - yield frame - if isinstance(frame, EndStreamQueueFrame): - break - - def get_async_send_queue(self): - return self.send_queue - - async def _marshal_frames(self): - while True: - frame: QueueFrame | list = await self.send_queue.get() - self.threadsafe_send_queue.put(frame) - self.send_queue.task_done() - if isinstance(frame, EndStreamQueueFrame): - break - - async def _wait_for_send_queue_to_empty(self): - await self.send_queue.join() - self.threadsafe_send_queue.join() - - async def stop_when_done(self): - await self._wait_for_send_queue_to_empty() - self.stop() - - async def run(self) -> None: - self._configure_daily() - - self._do_shutdown = False - - async_output_queue_marshal_task = asyncio.create_task(self._marshal_frames()) - - 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._do_shutdown and not self._stop_threads.is_set(): - await asyncio.sleep(1) - except Exception as e: - self._logger.error(f"Exception {e}") - raise e - finally: - self.client.leave() - - self._stop_threads.set() - - await self.receive_queue.put(EndStreamQueueFrame()) - await self.send_queue.put(EndStreamQueueFrame()) - await async_output_queue_marshal_task - - if self._camera_thread and self._camera_thread.is_alive(): - self._camera_thread.join() - if self._frame_consumer_thread and self._frame_consumer_thread.is_alive(): - self._frame_consumer_thread.join() - - def stop(self): - self._stop_threads.set() - def on_first_other_participant_joined(self): pass def call_joined(self, join_data, client_error): self._logger.info(f"Call_joined: {join_data}, {client_error}") - if self.speaker_enabled: - t = Thread(target=self._receive_audio, daemon=True) - t.start() def dialout(self, number): self.client.start_dialout({"phoneNumber": number}) @@ -318,14 +213,13 @@ class DailyTransportService(EventHandler): pass def on_participant_joined(self, participant): - if not self._other_participant_has_joined and participant["id"] != self.my_participant_id: + if not self._other_participant_has_joined and participant["id"] != self._my_participant_id: self._other_participant_has_joined = True self.on_first_other_participant_joined() def on_participant_left(self, participant, reason): - if len(self.client.participants()) < self.min_others_count + 1: - self._do_shutdown = True - pass + if len(self.client.participants()) < self._min_others_count + 1: + self._stop_threads.set() def on_app_message(self, message, sender): pass @@ -348,83 +242,3 @@ class DailyTransportService(EventHandler): def on_transcription_started(self, status): pass - - def _set_image(self, image: bytes): - self._images = itertools.cycle([image]) - - def _set_images(self, images: list[bytes], start_frame=0): - self._images = itertools.cycle(images) - - def _run_camera(self): - try: - while not self._stop_threads.is_set(): - if self._images: - this_frame = next(self._images) - self.camera.write_frame(this_frame) - - time.sleep(1.0 / 8) # 8 fps - except Exception as e: - self._logger.error(f"Exception {e} in camera thread.") - raise e - - def _frame_consumer(self): - self._logger.info("🎬 Starting frame consumer thread") - b = bytearray() - smallest_write_size = 3200 - all_audio_frames = bytearray() - while True: - try: - frames_or_frame: QueueFrame | list[QueueFrame] = self.threadsafe_send_queue.get() - if isinstance(frames_or_frame, QueueFrame): - frames: list[QueueFrame] = [frames_or_frame] - elif isinstance(frames_or_frame, list): - frames: list[QueueFrame] = frames_or_frame - else: - raise Exception("Unknown type in output queue") - - for frame in frames: - if isinstance(frame, EndStreamQueueFrame): - self._logger.info("Stopping frame consumer thread") - self.threadsafe_send_queue.task_done() - return - - # if interrupted, we just pull frames off the queue and discard them - if not self._is_interrupted.is_set(): - if frame: - if isinstance(frame, AudioQueueFrame): - chunk = frame.data - - all_audio_frames.extend(chunk) - - b.extend(chunk) - truncated_length: int = len(b) - (len(b) % smallest_write_size) - if truncated_length: - self.mic.write_frames(bytes(b[:truncated_length])) - b = b[truncated_length:] - elif isinstance(frame, ImageQueueFrame): - self._set_image(frame.image) - elif isinstance(frame, SpriteQueueFrame): - self._set_images(frame.images) - elif len(b): - self.mic.write_frames(bytes(b)) - b = bytearray() - else: - # if there are leftover audio bytes, write them now; failing to do so - # can cause static in the audio stream. - if len(b): - truncated_length = len(b) - (len(b) % 160) - self.mic.write_frames(bytes(b[:truncated_length])) - b = bytearray() - - if isinstance(frame, StartStreamQueueFrame): - self._is_interrupted.clear() - - self.threadsafe_send_queue.task_done() - except Empty: - if len(b): - self.mic.write_frames(bytes(b)) - - b = bytearray() - except Exception as e: - self._logger.error(f"Exception in frame_consumer: {e}, {len(b)}") - raise e diff --git a/src/dailyai/services/fal_ai_services.py b/src/dailyai/services/fal_ai_services.py index 2f56f3be7..ca1f93d89 100644 --- a/src/dailyai/services/fal_ai_services.py +++ b/src/dailyai/services/fal_ai_services.py @@ -3,11 +3,13 @@ import aiohttp import asyncio import io import os -import json from PIL import Image +from dailyai.services.ai_services import ImageGenService -from dailyai.services.ai_services import LLMService, TTSService, ImageGenService + +from dailyai.services.ai_services import ImageGenService +# Fal expects FAL_KEY_ID and FAL_KEY_SECRET to be set in the env class FalImageGenService(ImageGenService): @@ -44,5 +46,3 @@ class FalImageGenService(ImageGenService): image_stream = io.BytesIO(await response.content.read()) image = Image.open(image_stream) return (image_url, image.tobytes()) - - # return (image_url, dalle_im.tobytes()) diff --git a/src/dailyai/services/local_stt_service.py b/src/dailyai/services/local_stt_service.py index 866d77cac..21359ea23 100644 --- a/src/dailyai/services/local_stt_service.py +++ b/src/dailyai/services/local_stt_service.py @@ -1,9 +1,10 @@ import array import io import math +import time from typing import AsyncGenerator import wave -from dailyai.queue_frame import AudioQueueFrame, QueueFrame, TextQueueFrame +from dailyai.queue_frame import AudioQueueFrame, QueueFrame, TranscriptionQueueFrame from dailyai.services.ai_services import STTService @@ -59,7 +60,7 @@ class LocalSTTService(STTService): self._content.seek(0) text = await self.run_stt(self._content) self._new_wave() - yield TextQueueFrame(text) + yield TranscriptionQueueFrame(text, '', str(time.time())) # If we get this far, this is a frame of silence self._current_silence_frames += 1 diff --git a/src/dailyai/services/local_transport_service.py b/src/dailyai/services/local_transport_service.py new file mode 100644 index 000000000..91af6763c --- /dev/null +++ b/src/dailyai/services/local_transport_service.py @@ -0,0 +1,72 @@ +import asyncio +import time +import numpy as np +import tkinter as tk +import pyaudio + +from dailyai.services.base_transport_service import BaseTransportService + + +class LocalTransportService(BaseTransportService): + def __init__(self, **kwargs): + super().__init__(**kwargs) + self._sample_width = kwargs.get("sample_width") or 2 + self._n_channels = kwargs.get("n_channels") or 1 + 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") + + 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 + photo = tk.PhotoImage(width=self._camera_width, height=self._camera_height, data=data, format="PPM") + self._image_label.config(image=photo) + + # 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): + if self._camera_enabled and self._loop: + asyncio.run_coroutine_threadsafe( + self._write_frame_to_tkinter(frame), self._loop + ) + + def write_frame_to_mic(self, frame: bytes): + self._audio_stream.write(frame) + + def read_frames(self, desired_frame_count): + bytes = self._speaker_stream.read( + desired_frame_count, + exception_on_overflow=False, + ) + return bytes + + def _prerun(self): + if self._mic_enabled: + self._pyaudio = pyaudio.PyAudio() + self._audio_stream = self._pyaudio.open( + format=self._pyaudio.get_format_from_width(self._sample_width), + channels=self._n_channels, + rate=self._speaker_sample_rate, + output=True, + ) + + 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") + self._image_label = tk.Label(self._tk_root, image=photo) + self._image_label.pack() + + if self._speaker_enabled: + self._speaker_stream = self._pyaudio.open( + format=self._pyaudio.get_format_from_width(self._sample_width), + channels=self._n_channels, + rate=self._speaker_sample_rate, + frames_per_buffer=self._speaker_sample_rate, + input=True + ) diff --git a/src/dailyai/services/whisper_ai_services.py b/src/dailyai/services/whisper_ai_services.py index 88bb6f5d4..cc657e6cf 100644 --- a/src/dailyai/services/whisper_ai_services.py +++ b/src/dailyai/services/whisper_ai_services.py @@ -46,7 +46,7 @@ class WhisperSTTService(LocalSTTService): compute_type=self._compute_type) self._model = model - async def run_stt(self, audio: BinaryIO = None) -> str: + async def run_stt(self, audio: BinaryIO) -> str: """Transcribes given audio using Whisper""" segments, _ = await asyncio.to_thread(self._model.transcribe, audio) res: str = "" diff --git a/src/dailyai/tests/test_daily_transport_service.py b/src/dailyai/tests/test_daily_transport_service.py index 47ad6ac37..469b1bb1b 100644 --- a/src/dailyai/tests/test_daily_transport_service.py +++ b/src/dailyai/tests/test_daily_transport_service.py @@ -46,9 +46,14 @@ class TestDailyTransport(unittest.IsolatedAsyncioTestCase): @patch("dailyai.services.daily_transport_service.Daily") async def test_run_with_camera_and_mic(self, daily_mock, callclient_mock): from dailyai.services.daily_transport_service import DailyTransportService - transport = DailyTransportService("https://mock.daily.co/mock", "token", "bot", 0.01) - transport.mic_enabled = True - transport.camera_enabled = True + transport = DailyTransportService( + "https://mock.daily.co/mock", + "token", + "bot", + mic_enabled=True, + camera_enabled=True, + duration_minutes=0.01, + ) mic = MagicMock() camera = MagicMock() diff --git a/src/samples/foundational/01-say-one-thing.py b/src/samples/foundational/01-say-one-thing.py index 39a492fae..a6e3f820b 100644 --- a/src/samples/foundational/01-say-one-thing.py +++ b/src/samples/foundational/01-say-one-thing.py @@ -24,7 +24,7 @@ async def main(room_url): "Say One Thing", meeting_duration_minutes, ) - transport.mic_enabled = True + transport._mic_enabled = True tts = ElevenLabsTTSService(aiohttp_session=session, api_key=os.getenv("ELEVENLABS_API_KEY"), voice_id=os.getenv("ELEVENLABS_VOICE_ID")) # Register an event handler so we can play the audio when the participant joins. diff --git a/src/samples/foundational/01a-local-transport.py b/src/samples/foundational/01a-local-transport.py new file mode 100644 index 000000000..3b7df528a --- /dev/null +++ b/src/samples/foundational/01a-local-transport.py @@ -0,0 +1,34 @@ +import asyncio +import aiohttp +import os + +from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService +from dailyai.services.local_transport_service import LocalTransportService + + +async def main(): + async with aiohttp.ClientSession() as session: + meeting_duration_minutes = 1 + transport = LocalTransportService( + duration_minutes=meeting_duration_minutes, + mic_enabled=True + ) + tts = ElevenLabsTTSService( + aiohttp_session=session, + api_key=os.getenv("ELEVENLABS_API_KEY"), + voice_id=os.getenv("ELEVENLABS_VOICE_ID"), + ) + + async def say_something(): + await asyncio.sleep(1) + await tts.say( + "Hello there.", + transport.send_queue, + ) + await transport.stop_when_done() + + await asyncio.gather(transport.run(), say_something()) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/samples/foundational/02-llm-say-one-thing.py b/src/samples/foundational/02-llm-say-one-thing.py index d182ebb7b..b27bd7205 100644 --- a/src/samples/foundational/02-llm-say-one-thing.py +++ b/src/samples/foundational/02-llm-say-one-thing.py @@ -19,16 +19,16 @@ async def main(room_url): room_url, None, "Say One Thing From an LLM", - meeting_duration_minutes, + duration_minutes=meeting_duration_minutes, ) - transport.mic_enabled = True + transport._mic_enabled = True tts = ElevenLabsTTSService(aiohttp_session=session, api_key=os.getenv("ELEVENLABS_API_KEY"), voice_id=os.getenv("ELEVENLABS_VOICE_ID")) # tts = AzureTTSService(api_key=os.getenv("AZURE_SPEECH_API_KEY"), region=os.getenv("AZURE_SPEECH_REGION")) # tts = DeepgramTTSService(aiohttp_session=session, api_key=os.getenv("DEEPGRAM_API_KEY"), voice=os.getenv("DEEPGRAM_VOICE")) - - # llm = AzureLLMService(api_key=os.getenv("AZURE_CHATGPT_API_KEY"), endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"), model=os.getenv("AZURE_CHATGPT_MODEL")) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_CHATGPT_API_KEY")) + + llm = AzureLLMService(api_key=os.getenv("AZURE_CHATGPT_API_KEY"), endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"), model=os.getenv("AZURE_CHATGPT_MODEL")) + #llm = OpenAILLMService(api_key=os.getenv("OPENAI_CHATGPT_API_KEY")) messages = [{ "role": "system", "content": "You are an LLM in a WebRTC session, and this is a 'hello world' demo. Say hello to the world." diff --git a/src/samples/foundational/03-still-frame.py b/src/samples/foundational/03-still-frame.py index d5cb2124b..e18f20bb6 100644 --- a/src/samples/foundational/03-still-frame.py +++ b/src/samples/foundational/03-still-frame.py @@ -22,17 +22,17 @@ async def main(room_url): room_url, None, "Show a still frame image", - meeting_duration_minutes, + duration_minutes=meeting_duration_minutes, ) - transport.mic_enabled = False - transport.camera_enabled = True - transport.camera_width = 1024 - transport.camera_height = 1024 + transport._mic_enabled = False + transport._camera_enabled = True + transport._camera_width = 1024 + transport._camera_height = 1024 imagegen = FalImageGenService(image_size="1024x1024", aiohttp_session=session, key_id=os.getenv("FAL_KEY_ID"), key_secret=os.getenv("FAL_KEY_SECRET")) # imagegen = OpenAIImageGenService(aiohttp_session=session, api_key=os.getenv("OPENAI_DALLE_API_KEY"), image_size="1024x1024") # imagegen = AzureImageGenServiceREST(image_size="1024x1024", aiohttp_session=session, api_key=os.getenv("AZURE_DALLE_API_KEY"), endpoint=os.getenv("AZURE_DALLE_ENDPOINT"), model=os.getenv("AZURE_DALLE_MODEL")) - + image_task = asyncio.create_task( imagegen.run_to_queue( transport.send_queue, [ diff --git a/src/samples/foundational/03a-image-local.py b/src/samples/foundational/03a-image-local.py new file mode 100644 index 000000000..40cb2b245 --- /dev/null +++ b/src/samples/foundational/03a-image-local.py @@ -0,0 +1,50 @@ +import asyncio +import aiohttp +import os + +import tkinter as tk + +from dailyai.queue_frame import TextQueueFrame +from dailyai.services.fal_ai_services import FalImageGenService +from dailyai.services.local_transport_service import LocalTransportService + +local_joined = False +participant_joined = False + + +async def main(): + async with aiohttp.ClientSession() as session: + meeting_duration_minutes = 2 + tk_root = tk.Tk() + tk_root.title("Calendar") + transport = LocalTransportService( + tk_root=tk_root, + mic_enabled=True, + camera_enabled=True, + camera_width=1024, + camera_height=1024, + duration_minutes=meeting_duration_minutes, + ) + + imagegen = FalImageGenService( + image_size="1024x1024", + aiohttp_session=session, + key_id=os.getenv("FAL_KEY_ID"), + key_secret=os.getenv("FAL_KEY_SECRET"), + ) + image_task = asyncio.create_task( + imagegen.run_to_queue( + transport.send_queue, [TextQueueFrame("a cat in the style of picasso")] + ) + ) + + async def run_tk(): + while not transport._stop_threads.is_set(): + tk_root.update() + tk_root.update_idletasks() + await asyncio.sleep(0.1) + + await asyncio.gather(transport.run(), image_task, run_tk()) + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/samples/foundational/04-utterance-and-speech.py b/src/samples/foundational/04-utterance-and-speech.py index 968d9d896..9ac8a8f81 100644 --- a/src/samples/foundational/04-utterance-and-speech.py +++ b/src/samples/foundational/04-utterance-and-speech.py @@ -17,12 +17,12 @@ async def main(room_url: str): transport = DailyTransportService( room_url, None, - "Say Two Things Bot", - 1, + "Static And Dynamic Speech", + duration_minutes=1, ) - transport.mic_enabled = True - transport.mic_sample_rate = 16000 - transport.camera_enabled = False + transport._mic_enabled = True + transport._mic_sample_rate = 16000 + transport._camera_enabled = False llm = AzureLLMService(api_key=os.getenv("AZURE_CHATGPT_API_KEY"), endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"), model=os.getenv("AZURE_CHATGPT_MODEL")) azure_tts = AzureTTSService(api_key=os.getenv("AZURE_SPEECH_API_KEY"), region=os.getenv("AZURE_SPEECH_REGION")) diff --git a/src/samples/foundational/05-sync-speech-and-image.py b/src/samples/foundational/05-sync-speech-and-image.py index 77e95fc7b..af4e2d71d 100644 --- a/src/samples/foundational/05-sync-speech-and-image.py +++ b/src/samples/foundational/05-sync-speech-and-image.py @@ -19,13 +19,13 @@ async def main(room_url): room_url, None, "Month Narration Bot", - meeting_duration_minutes, + duration_minutes=meeting_duration_minutes, ) - transport.mic_enabled = True - transport.camera_enabled = True - transport.mic_sample_rate = 16000 - transport.camera_width = 1024 - transport.camera_height = 1024 + transport._mic_enabled = True + transport._camera_enabled = True + transport._mic_sample_rate = 16000 + transport._camera_width = 1024 + transport._camera_height = 1024 llm = AzureLLMService(api_key=os.getenv("AZURE_CHATGPT_API_KEY"), endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"), model=os.getenv("AZURE_CHATGPT_MODEL")) tts = ElevenLabsTTSService(aiohttp_session=session, api_key=os.getenv("ELEVENLABS_API_KEY"), voice_id="ErXwobaYiN019PkySvjV") @@ -34,7 +34,7 @@ async def main(room_url): dalle = FalImageGenService(image_size="1024x1024", aiohttp_session=session, key_id=os.getenv("FAL_KEY_ID"), key_secret=os.getenv("FAL_KEY_SECRET")) # dalle = OpenAIImageGenService(aiohttp_session=session, api_key=os.getenv("OPENAI_DALLE_API_KEY"), image_size="1024x1024") # dalle = AzureImageGenServiceREST(image_size="1024x1024", aiohttp_session=session, api_key=os.getenv("AZURE_DALLE_API_KEY"), endpoint=os.getenv("AZURE_DALLE_ENDPOINT"), model=os.getenv("AZURE_DALLE_MODEL")) - + # Get a complete audio chunk from the given text. Splitting this into its own # coroutine lets us ensure proper ordering of the audio chunks on the send queue. async def get_all_audio(text): @@ -121,4 +121,4 @@ async def main(room_url): if __name__ == "__main__": (url, token) = configure() - asyncio.run(main(url)) \ No newline at end of file + asyncio.run(main(url)) diff --git a/src/samples/foundational/05a-local-sync-speech-and-text.py b/src/samples/foundational/05a-local-sync-speech-and-text.py new file mode 100644 index 000000000..fb9f419bb --- /dev/null +++ b/src/samples/foundational/05a-local-sync-speech-and-text.py @@ -0,0 +1,134 @@ +import aiohttp +import argparse +import asyncio +import tkinter as tk +import os + +from dailyai.queue_frame import AudioQueueFrame, ImageQueueFrame +from dailyai.services.azure_ai_services import AzureLLMService +from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService +from dailyai.services.fal_ai_services import FalImageGenService +from dailyai.services.local_transport_service import LocalTransportService + + +async def main(room_url): + async with aiohttp.ClientSession() as session: + meeting_duration_minutes = 5 + tk_root = tk.Tk() + tk_root.title("Calendar") + + transport = LocalTransportService( + mic_enabled=True, + camera_enabled=True, + camera_width=1024, + camera_height=1024, + duration_minutes=meeting_duration_minutes, + tk_root=tk_root, + ) + + llm = AzureLLMService( + api_key=os.getenv("AZURE_CHATGPT_API_KEY"), + endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"), + model=os.getenv("AZURE_CHATGPT_MODEL"), + ) + tts = ElevenLabsTTSService( + aiohttp_session=session, + api_key=os.getenv("ELEVENLABS_API_KEY"), + voice_id="ErXwobaYiN019PkySvjV", + ) + dalle = FalImageGenService( + image_size="1024x1024", + aiohttp_session=session, + key_id=os.getenv("FAL_KEY_ID"), + key_secret=os.getenv("FAL_KEY_SECRET"), + ) + + # Get a complete audio chunk from the given text. Splitting this into its own + # coroutine lets us ensure proper ordering of the audio chunks on the send queue. + async def get_all_audio(text): + all_audio = bytearray() + async for audio in tts.run_tts(text): + all_audio.extend(audio) + + return all_audio + + async def get_month_data(month): + messages = [ + { + "role": "system", + "content": f"Describe a nature photograph suitable for use in a calendar, for the month of {month}. Include only the image description with no preamble. Limit the description to one sentence, please.", + } + ] + + image_description = await llm.run_llm(messages) + if not image_description: + return + + to_speak = f"{month}: {image_description}" + audio_task = asyncio.create_task(get_all_audio(to_speak)) + image_task = asyncio.create_task(dalle.run_image_gen(image_description)) + (audio, image_data) = await asyncio.gather( + audio_task, image_task + ) + + return { + "month": month, + "text": image_description, + "image_url": image_data[0], + "image": image_data[1], + "audio": audio, + } + + months: list[str] = [ + "January", + "February", + "March", + "April", + "May", + "June", + "July", + "August", + "September", + "October", + "November", + "December", + ] + + async def show_images(): + # This will play the months in the order they're completed. The benefit + # is we'll have as little delay as possible before the first month, and + # likely no delay between months, but the months won't display in order. + for month_data_task in asyncio.as_completed(month_tasks): + data = await month_data_task + if data: + await transport.send_queue.put( + [ + ImageQueueFrame(data["image_url"], data["image"]), + AudioQueueFrame(data["audio"]), + ] + ) + + await asyncio.sleep(25) + + # wait for the output queue to be empty, then leave the meeting + await transport.stop_when_done() + + async def run_tk(): + while not transport._stop_threads.is_set(): + tk_root.update() + tk_root.update_idletasks() + await asyncio.sleep(0.1) + + month_tasks = [asyncio.create_task(get_month_data(month)) for month in months] + + await asyncio.gather(transport.run(), show_images(), run_tk()) + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="Simple Daily Bot Sample") + parser.add_argument( + "-u", "--url", type=str, required=True, help="URL of the Daily room to join" + ) + + args, unknown = parser.parse_known_args() + + asyncio.run(main(args.url)) diff --git a/src/samples/foundational/06-listen-and-respond.py b/src/samples/foundational/06-listen-and-respond.py index d10ff1e0c..582701428 100644 --- a/src/samples/foundational/06-listen-and-respond.py +++ b/src/samples/foundational/06-listen-and-respond.py @@ -15,11 +15,11 @@ async def main(room_url: str, token): room_url, token, "Respond bot", - 5, + duration_minutes=5, ) - transport.mic_enabled = True - transport.mic_sample_rate = 16000 - transport.camera_enabled = False + transport._mic_enabled = True + transport._mic_sample_rate = 16000 + transport._camera_enabled = False llm = AzureLLMService(api_key=os.getenv("AZURE_CHATGPT_API_KEY"), endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"), model=os.getenv("AZURE_CHATGPT_MODEL")) tts = AzureTTSService(api_key=os.getenv("AZURE_SPEECH_API_KEY"), region=os.getenv("AZURE_SPEECH_REGION")) @@ -30,11 +30,14 @@ async def main(room_url: str, token): async def handle_transcriptions(): messages = [ - {"role": "system", "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio. Respond to what the user said in a creative and helpful way."}, + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio. Respond to what the user said in a creative and helpful way.", + }, ] - tma_in = LLMUserContextAggregator(messages, transport.my_participant_id) - tma_out = LLMAssistantContextAggregator(messages, transport.my_participant_id) + tma_in = LLMUserContextAggregator(messages, transport._my_participant_id) + tma_out = LLMAssistantContextAggregator(messages, transport._my_participant_id) await tts.run_to_queue( transport.send_queue, tma_out.run( diff --git a/src/samples/foundational/06a-image-sync.py b/src/samples/foundational/06a-image-sync.py index 2c280044b..cff2fa48a 100644 --- a/src/samples/foundational/06a-image-sync.py +++ b/src/samples/foundational/06a-image-sync.py @@ -40,11 +40,11 @@ async def main(room_url: str, token): "Respond bot", 5, ) - transport.camera_enabled = True - transport.camera_width = 1024 - transport.camera_height = 1024 - transport.mic_enabled = True - transport.mic_sample_rate = 16000 + transport._camera_enabled = True + transport._camera_width = 1024 + transport._camera_height = 1024 + transport._mic_enabled = True + transport._mic_sample_rate = 16000 llm = AzureLLMService(api_key=os.getenv("AZURE_CHATGPT_API_KEY"), endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"), model=os.getenv("AZURE_CHATGPT_MODEL")) tts = AzureTTSService(api_key=os.getenv("AZURE_SPEECH_API_KEY"), region=os.getenv("AZURE_SPEECH_REGION")) @@ -74,10 +74,10 @@ async def main(room_url: str, token): ] tma_in = LLMUserContextAggregator( - messages, transport.my_participant_id + messages, transport._my_participant_id ) tma_out = LLMAssistantContextAggregator( - messages, transport.my_participant_id + messages, transport._my_participant_id ) image_sync_aggregator = ImageSyncAggregator( os.path.join(os.path.dirname(__file__), "assets", "speaking.png"), diff --git a/src/samples/foundational/07-interruptible.py b/src/samples/foundational/07-interruptible.py index 532cdf11d..b75d2e1ad 100644 --- a/src/samples/foundational/07-interruptible.py +++ b/src/samples/foundational/07-interruptible.py @@ -16,12 +16,12 @@ async def main(room_url: str, token): room_url, token, "Respond bot", - 5, + duration_minutes=5, ) - transport.mic_enabled = True - transport.mic_sample_rate = 16000 - transport.camera_enabled = False - transport.start_transcription = True + transport._mic_enabled = True + transport._mic_sample_rate = 16000 + transport._camera_enabled = False + transport._start_transcription = True llm = AzureLLMService(api_key=os.getenv("AZURE_CHATGPT_API_KEY"), endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"), model=os.getenv("AZURE_CHATGPT_MODEL")) tts = AzureTTSService(api_key=os.getenv("AZURE_SPEECH_API_KEY"), region=os.getenv("AZURE_SPEECH_REGION")) @@ -51,7 +51,7 @@ async def main(room_url: str, token): frame_generator=transport.get_receive_frames, runner=run_response, interrupt=transport.interrupt, - my_participant_id=transport.my_participant_id, + my_participant_id=transport._my_participant_id, llm_messages=messages, ) await conversation_wrapper.run_conversation() diff --git a/src/samples/foundational/08-bots-arguing.py b/src/samples/foundational/08-bots-arguing.py index 00f7bc64f..3fd040578 100644 --- a/src/samples/foundational/08-bots-arguing.py +++ b/src/samples/foundational/08-bots-arguing.py @@ -20,13 +20,13 @@ async def main(room_url:str): room_url, None, "Respond bot", - 600, + duration_minutes=10 ) - transport.mic_enabled = True - transport.mic_sample_rate = 16000 - transport.camera_enabled = True - transport.camera_width = 1024 - transport.camera_height = 1024 + transport._mic_enabled = True + transport._mic_sample_rate = 16000 + transport._camera_enabled = True + transport._camera_width = 1024 + transport._camera_height = 1024 llm = AzureLLMService(api_key=os.getenv("AZURE_CHATGPT_API_KEY"), endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"), model=os.getenv("AZURE_CHATGPT_MODEL")) tts1 = AzureTTSService(api_key=os.getenv("AZURE_SPEECH_API_KEY"), region=os.getenv("AZURE_SPEECH_REGION")) @@ -101,4 +101,4 @@ async def main(room_url:str): if __name__ == "__main__": (url, token) = configure() - asyncio.run(main(url)) \ No newline at end of file + asyncio.run(main(url)) diff --git a/src/samples/foundational/10-wake-word.py b/src/samples/foundational/10-wake-word.py index 17ae12216..897e0a221 100644 --- a/src/samples/foundational/10-wake-word.py +++ b/src/samples/foundational/10-wake-word.py @@ -113,13 +113,13 @@ async def main(room_url: str, token): room_url, token, "Santa Cat", - 180, + duration_minutes=3 ) - transport.mic_enabled = True - transport.mic_sample_rate = 16000 - transport.camera_enabled = True - transport.camera_width = 720 - transport.camera_height = 1280 + transport._mic_enabled = True + transport._mic_sample_rate = 16000 + transport._camera_enabled = True + transport._camera_width = 720 + transport._camera_height = 1280 llm = AzureLLMService(api_key=os.getenv("AZURE_CHATGPT_API_KEY"), endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"), model=os.getenv("AZURE_CHATGPT_MODEL")) tts = ElevenLabsTTSService(aiohttp_session=session, api_key=os.getenv("ELEVENLABS_API_KEY"), voice_id="jBpfuIE2acCO8z3wKNLl") @@ -135,12 +135,12 @@ async def main(room_url: str, token): ] tma_in = LLMUserContextAggregator( - messages, transport.my_participant_id + messages, transport._my_participant_id ) tma_out = LLMAssistantContextAggregator( - messages, transport.my_participant_id + messages, transport._my_participant_id ) - tf = TranscriptFilter(transport.my_participant_id) + tf = TranscriptFilter(transport._my_participant_id) ncf = NameCheckFilter(["Santa Cat", "Santa"]) await tts.run_to_queue( transport.send_queue, diff --git a/src/samples/foundational/11-sound-effects.py b/src/samples/foundational/11-sound-effects.py index e2d704a4b..a454780e2 100644 --- a/src/samples/foundational/11-sound-effects.py +++ b/src/samples/foundational/11-sound-effects.py @@ -74,11 +74,11 @@ async def main(room_url: str, token): room_url, token, "Respond bot", - 5, + duration_minutes=5, ) - transport.mic_enabled = True - transport.mic_sample_rate = 16000 - transport.camera_enabled = False + transport._mic_enabled = True + transport._mic_sample_rate = 16000 + transport._camera_enabled = False llm = AzureLLMService(api_key=os.getenv("AZURE_CHATGPT_API_KEY"), endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"), model=os.getenv("AZURE_CHATGPT_MODEL")) tts = ElevenLabsTTSService(aiohttp_session=session, api_key=os.getenv("ELEVENLABS_API_KEY"), voice_id="ErXwobaYiN019PkySvjV") @@ -94,10 +94,10 @@ async def main(room_url: str, token): ] tma_in = LLMUserContextAggregator( - messages, transport.my_participant_id + messages, transport._my_participant_id ) tma_out = LLMAssistantContextAggregator( - messages, transport.my_participant_id + messages, transport._my_participant_id ) out_sound = OutboundSoundEffectWrapper() in_sound = InboundSoundEffectWrapper() @@ -121,7 +121,7 @@ async def main(room_url: str, token): ) ) ) - + transport.transcription_settings["extra"]["punctuate"] = True await asyncio.gather(transport.run(), handle_transcriptions()) @@ -129,4 +129,4 @@ async def main(room_url: str, token): if __name__ == "__main__": (url, token) = configure() - asyncio.run(main(url, token)) \ No newline at end of file + asyncio.run(main(url, token)) diff --git a/src/samples/foundational/13-whisper-transcription.py b/src/samples/foundational/13-whisper-transcription.py index ff5dc2f28..e7d0320e6 100644 --- a/src/samples/foundational/13-whisper-transcription.py +++ b/src/samples/foundational/13-whisper-transcription.py @@ -14,9 +14,9 @@ async def main(room_url: str): None, "Transcription bot", ) - transport.mic_enabled = False - transport.camera_enabled = False - transport.speaker_enabled = True + transport._mic_enabled = False + transport._camera_enabled = False + transport._speaker_enabled = True stt = WhisperSTTService() transcription_output_queue = asyncio.Queue() @@ -36,4 +36,4 @@ async def main(room_url: str): if __name__ == "__main__": (url, token) = configure() - asyncio.run(main(url)) \ No newline at end of file + asyncio.run(main(url)) diff --git a/src/samples/foundational/13a-whisper-local.py b/src/samples/foundational/13a-whisper-local.py new file mode 100644 index 000000000..4656fa70d --- /dev/null +++ b/src/samples/foundational/13a-whisper-local.py @@ -0,0 +1,58 @@ +import argparse +import asyncio +import wave +from dailyai.queue_frame import EndStreamQueueFrame, TranscriptionQueueFrame + +from dailyai.services.local_transport_service import LocalTransportService +from dailyai.services.whisper_ai_services import WhisperSTTService + + +async def main(room_url: str): + global transport + global stt + + meeting_duration_minutes = 1 + transport = LocalTransportService( + mic_enabled=True, + camera_enabled=False, + speaker_enabled=True, + duration_minutes=meeting_duration_minutes, + ) + stt = WhisperSTTService() + transcription_output_queue = asyncio.Queue() + transport_done = asyncio.Event() + + async def handle_transcription(): + print("`````````TRANSCRIPTION`````````") + while not transport_done.is_set(): + item = await transcription_output_queue.get() + print("got item from queue", item) + if isinstance(item, TranscriptionQueueFrame): + print(item.text) + elif isinstance(item, EndStreamQueueFrame): + break + print("handle_transcription done") + + async def handle_speaker(): + await stt.run_to_queue( + transcription_output_queue, transport.get_receive_frames() + ) + await transcription_output_queue.put(EndStreamQueueFrame()) + print("handle speaker done.") + + async def run_until_done(): + await transport.run() + transport_done.set() + print("run_until_done done") + + await asyncio.gather(run_until_done(), handle_speaker(), handle_transcription()) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="Simple Daily Bot Sample") + parser.add_argument( + "-u", "--url", type=str, required=True, help="URL of the Daily room to join" + ) + + args, unknown = parser.parse_known_args() + asyncio.run(main(args.url)) diff --git a/src/samples/image-gen.py b/src/samples/image-gen.py index fc2ddbaaa..8b5ca5566 100644 --- a/src/samples/image-gen.py +++ b/src/samples/image-gen.py @@ -23,11 +23,11 @@ async def main(room_url: str, token): "Imagebot", 1, ) - transport.mic_enabled = True - transport.camera_enabled = True - transport.mic_sample_rate = 16000 - transport.camera_width = 1024 - transport.camera_height = 1024 + 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() @@ -39,7 +39,7 @@ async def main(room_url: str, token): sentence = "" async for message in transport.get_transcriptions(): print(f"transcription message: {message}") - if message["session_id"] == transport.my_participant_id: + if message["session_id"] == transport._my_participant_id: continue finder = message["text"].find("start over") print(f"finder: {finder}") diff --git a/src/samples/internal/11a-dial-out.py b/src/samples/internal/11a-dial-out.py index cd6b39a4d..3ea1519d6 100644 --- a/src/samples/internal/11a-dial-out.py +++ b/src/samples/internal/11a-dial-out.py @@ -70,9 +70,9 @@ async def main(room_url: str, token, phone): "Respond bot", 300, ) - transport.mic_enabled = True - transport.mic_sample_rate = 16000 - transport.camera_enabled = False + transport._mic_enabled = True + transport._mic_sample_rate = 16000 + transport._camera_enabled = False llm = AzureLLMService() tts = AzureTTSService() @@ -87,10 +87,10 @@ async def main(room_url: str, token, phone): ] tma_in = LLMContextAggregator( - messages, "user", transport.my_participant_id + messages, "user", transport._my_participant_id ) tma_out = LLMContextAggregator( - messages, "assistant", transport.my_participant_id + messages, "assistant", transport._my_participant_id ) out_sound = OutboundSoundEffectWrapper() in_sound = InboundSoundEffectWrapper() @@ -109,14 +109,14 @@ async def main(room_url: str, token, phone): ) ) ) - ) + ) ) ) @transport.event_handler("on_participant_joined") async def pax_joined(transport, pax): print(f"PARTICIPANT JOINED: {pax}") - + @transport.event_handler("on_call_state_updated") async def on_call_state_updated(transport, state): if (state == "joined"): @@ -132,4 +132,4 @@ async def main(room_url: str, token, phone): if __name__ == "__main__": (url, token) = configure() - asyncio.run(main(url, token)) \ No newline at end of file + asyncio.run(main(url, token))