diff --git a/pyproject.toml b/pyproject.toml index 7603f8500..2f5d8dc95 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -13,10 +13,14 @@ dependencies = [ "fal", "faster_whisper", "google-cloud-texttospeech", + "numpy", "openai", "Pillow", "pyht", "python-dotenv", + "torch", + "torchaudio", + "pyaudio", "typing-extensions" ] diff --git a/src/dailyai/queue_frame.py b/src/dailyai/queue_frame.py index d43dbdf82..570c66447 100644 --- a/src/dailyai/queue_frame.py +++ b/src/dailyai/queue_frame.py @@ -23,6 +23,14 @@ class LLMResponseEndQueueFrame(QueueFrame): pass +class UserStartedSpeakingFrame(QueueFrame): + pass + + +class UserStoppedSpeakingFrame(QueueFrame): + pass + + @dataclass() class AudioQueueFrame(QueueFrame): data: bytes diff --git a/src/dailyai/services/ai_services.py b/src/dailyai/services/ai_services.py index 29b7d6488..e4b1fe598 100644 --- a/src/dailyai/services/ai_services.py +++ b/src/dailyai/services/ai_services.py @@ -2,6 +2,7 @@ import asyncio import io import logging import time +import datetime import wave from dailyai.queue_frame import ( @@ -200,8 +201,9 @@ class FrameLogger(AIService): async def process_frame(self, frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]: if isinstance(frame, (AudioQueueFrame, ImageQueueFrame)): - self.logger.info(f"{self.prefix}: {type(frame)}") + self.logger.info( + f"{datetime.datetime.utcnow().isoformat()} {self.prefix}: {type(frame)}") else: - print(f"{self.prefix}: {frame}") + print(f"{datetime.datetime.utcnow().isoformat()} {self.prefix}: {frame}") yield frame diff --git a/src/dailyai/services/base_transport_service.py b/src/dailyai/services/base_transport_service.py index 530990cfa..ff366c67e 100644 --- a/src/dailyai/services/base_transport_service.py +++ b/src/dailyai/services/base_transport_service.py @@ -6,6 +6,12 @@ import queue import threading import time from typing import AsyncGenerator +import numpy as np +import pyaudio +import torch +import torchaudio +from enum import Enum +import datetime from dailyai.queue_frame import ( AudioQueueFrame, @@ -14,8 +20,57 @@ from dailyai.queue_frame import ( QueueFrame, SpriteQueueFrame, StartStreamQueueFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame ) +torch.set_num_threads(1) + +model, utils = torch.hub.load(repo_or_dir='snakers4/silero-vad', + model='silero_vad', + force_reload=False) + +(get_speech_timestamps, + save_audio, + read_audio, + VADIterator, + collect_chunks) = utils + +# Taken from utils_vad.py + + +def validate(model, + inputs: torch.Tensor): + with torch.no_grad(): + outs = model(inputs) + return outs + +# Provided by Alexander Veysov + + +def int2float(sound): + abs_max = np.abs(sound).max() + sound = sound.astype('float32') + if abs_max > 0: + sound *= 1/32768 + sound = sound.squeeze() # depends on the use case + return sound + + +FORMAT = pyaudio.paInt16 +CHANNELS = 1 +SAMPLE_RATE = 16000 +CHUNK = int(SAMPLE_RATE / 10) + +audio = pyaudio.PyAudio() + + +class VADState(Enum): + QUIET = 1 + STARTING = 2 + SPEAKING = 3 + STOPPING = 4 + class BaseTransportService(): @@ -31,6 +86,16 @@ class BaseTransportService(): 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 + self._vad_start_s = kwargs.get("vad_start_s") or 0.2 + self._vad_stop_s = kwargs.get("vad_stop_s") or 1.2 + + self._vad_samples = 1536 + vad_frame_s = self._vad_samples / SAMPLE_RATE + self._vad_start_frames = round(self._vad_start_s / vad_frame_s) + self._vad_stop_frames = round(self._vad_stop_s / vad_frame_s) + self._vad_starting_count = 0 + self._vad_stopping_count = 0 + self._vad_state = VADState.QUIET duration_minutes = kwargs.get("duration_minutes") or 10 self._expiration = time.time() + duration_minutes * 60 @@ -41,6 +106,7 @@ class BaseTransportService(): self._threadsafe_send_queue = queue.Queue() self._images = None + self._user_is_speaking = False try: self._loop: asyncio.AbstractEventLoop | None = asyncio.get_running_loop() @@ -55,17 +121,25 @@ class BaseTransportService(): async def run(self): 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(target=self._frame_consumer, daemon=True) + 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() + # TODO-CB: This is interesting + # self._receive_audio_thread = threading.Thread( + # target=self._receive_audio, daemon=True) + # self._receive_audio_thread.start() + + self._vad_thread = threading.Thread(target=self._vad, daemon=True) + self._vad_thread.start() try: while ( @@ -122,6 +196,59 @@ class BaseTransportService(): def _prerun(self): pass + def _vad(self): + # CB: Starting silero VAD stuff + # TODO-CB: Probably need to force virtual speaker creation if we're + # going to build this in? + # TODO-CB: pyaudio installation + while not self._stop_threads.is_set(): + 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() + speaking = new_confidence > 0.5 + + if speaking: + match self._vad_state: + case VADState.QUIET: + self._vad_state = VADState.STARTING + self._vad_starting_count = 1 + case VADState.STARTING: + self._vad_starting_count += 1 + case VADState.STOPPING: + self._vad_state = VADState.SPEAKING + self._vad_stopping_count = 0 + else: + match self._vad_state: + case VADState.STARTING: + self._vad_state = VADState.QUIET + self._vad_starting_count = 0 + case VADState.SPEAKING: + self._vad_state = VADState.STOPPING + self._vad_stopping_count = 1 + case VADState.STOPPING: + self._vad_stopping_count += 1 + + if self._vad_state == VADState.STARTING and self._vad_starting_count >= self._vad_start_frames: + print( + f'{datetime.datetime.utcnow().isoformat()} queueing start frame') + asyncio.run_coroutine_threadsafe( + self.receive_queue.put( + UserStartedSpeakingFrame()), self._loop + ) + self._vad_state = VADState.SPEAKING + self._vad_starting_count = 0 + if self._vad_state == VADState.STOPPING and self._vad_stopping_count >= self._vad_stop_frames: + print( + f'{datetime.datetime.utcnow().isoformat()} queueing stop frame') + asyncio.run_coroutine_threadsafe( + self.receive_queue.put( + UserStoppedSpeakingFrame()), self._loop + ) + self._vad_state = VADState.QUIET + self._vad_stopping_count = 0 + async def _marshal_frames(self): while True: frame: QueueFrame | list = await self.send_queue.get() @@ -213,7 +340,8 @@ 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, ImageQueueFrame): self._set_image(frame.image) @@ -227,7 +355,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, StartStreamQueueFrame): @@ -240,5 +369,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 diff --git a/src/dailyai/services/daily_transport_service.py b/src/dailyai/services/daily_transport_service.py index 44c2f63fb..72a1c3fd9 100644 --- a/src/dailyai/services/daily_transport_service.py +++ b/src/dailyai/services/daily_transport_service.py @@ -1,18 +1,4 @@ -import asyncio -import inspect -import logging -import signal -import threading -import types - -from functools import partial - -from dailyai.queue_frame import ( - TranscriptionQueueFrame, -) - -from threading import Event - +from dailyai.services.base_transport_service import BaseTransportService from daily import ( EventHandler, CallClient, @@ -21,8 +7,61 @@ from daily import ( VirtualMicrophoneDevice, VirtualSpeakerDevice, ) +from threading import Event +from dailyai.queue_frame import ( + TranscriptionQueueFrame, +) +from functools import partial +import types +import pyaudio +import torchaudio +import asyncio +import inspect +import io +import logging +import numpy as np +import signal +import threading +import torch +torch.set_num_threads(1) -from dailyai.services.base_transport_service import BaseTransportService +model, utils = torch.hub.load(repo_or_dir='snakers4/silero-vad', + model='silero_vad', + force_reload=True) + +(get_speech_timestamps, + save_audio, + read_audio, + VADIterator, + collect_chunks) = utils + +# Taken from utils_vad.py + + +def validate(model, + inputs: torch.Tensor): + with torch.no_grad(): + outs = model(inputs) + return outs + +# Provided by Alexander Veysov + + +def int2float(sound): + abs_max = np.abs(sound).max() + sound = sound.astype('float32') + if abs_max > 0: + sound *= 1/32768 + sound = sound.squeeze() # depends on the use case + return sound + + +FORMAT = pyaudio.paInt16 +CHANNELS = 1 +SAMPLE_RATE = 16000 +CHUNK = int(SAMPLE_RATE / 10) + +audio = pyaudio.PyAudio() class DailyTransportService(BaseTransportService, EventHandler): @@ -45,7 +84,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 @@ -80,7 +120,8 @@ class DailyTransportService(BaseTransportService, EventHandler): for handler in self._event_handlers[event_name]: if inspect.iscoroutinefunction(handler): if self._loop: - asyncio.run_coroutine_threadsafe(handler(*args, **kwargs), self._loop) + asyncio.run_coroutine_threadsafe( + handler(*args, **kwargs), self._loop) else: raise Exception( "No event loop to run coroutine. In order to use async event handlers, you must run the DailyTransportService in an asyncio event loop.") @@ -92,7 +133,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]: @@ -105,7 +147,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): @@ -149,7 +192,8 @@ class DailyTransportService(BaseTransportService, EventHandler): 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.join(self._room_url, self._token, + completion=self.call_joined) self._my_participant_id = self.client.participants()["local"]["id"] self.client.update_inputs( @@ -242,8 +286,10 @@ class DailyTransportService(BaseTransportService, EventHandler): participantId = message["participantId"] elif "session_id" in message: participantId = message["session_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_stopped(self, stopped_by, stopped_by_error): pass diff --git a/src/examples/foundational/02-llm-say-one-thing.py b/src/examples/foundational/02-llm-say-one-thing.py index b15023380..f2c454314 100644 --- a/src/examples/foundational/02-llm-say-one-thing.py +++ b/src/examples/foundational/02-llm-say-one-thing.py @@ -20,7 +20,8 @@ async def main(room_url): None, "Say One Thing From an LLM", duration_minutes=meeting_duration_minutes, - mic_enabled=True + mic_enabled=True, + speaker_enabled=True ) tts = ElevenLabsTTSService( diff --git a/src/examples/foundational/06-listen-and-respond.py b/src/examples/foundational/06-listen-and-respond.py index fa5e077cc..c67f696c0 100644 --- a/src/examples/foundational/06-listen-and-respond.py +++ b/src/examples/foundational/06-listen-and-respond.py @@ -5,6 +5,7 @@ from dailyai.services.daily_transport_service import DailyTransportService from dailyai.services.azure_ai_services import AzureLLMService, AzureTTSService from dailyai.queue_aggregators import LLMAssistantContextAggregator, LLMContextAggregator, LLMUserContextAggregator from examples.foundational.support.runner import configure +from dailyai.services.ai_services import FrameLogger async def main(room_url: str, token): @@ -16,7 +17,8 @@ async def main(room_url: str, token): start_transcription=True, mic_enabled=True, mic_sample_rate=16000, - camera_enabled=False + camera_enabled=False, + speaker_enabled=True ) llm = AzureLLMService( @@ -26,6 +28,7 @@ async def main(room_url: str, token): tts = AzureTTSService( api_key=os.getenv("AZURE_SPEECH_API_KEY"), region=os.getenv("AZURE_SPEECH_REGION")) + fl = FrameLogger("transport") @transport.event_handler("on_first_other_participant_joined") async def on_first_other_participant_joined(transport): @@ -39,14 +42,18 @@ async def main(room_url: str, token): }, ] - 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( llm.run( tma_in.run( - transport.get_receive_frames() + fl.run( + transport.get_receive_frames() + ) ) ) )