adding silero VAD

pyproject.toml

@@ -13,10 +13,14 @@ dependencies = [
     "fal",
     "faster_whisper",
     "google-cloud-texttospeech",
+    "numpy",
     "openai",
     "Pillow",
     "pyht",
     "python-dotenv",
+    "torch",
+    "torchaudio",
+    "pyaudio",
     "typing-extensions"
 ]
 

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

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

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)
+            # TODO-CB: This is interesting
-            self._receive_audio_thread.start()
+            # 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

src/dailyai/services/daily_transport_service.py

@@ -1,18 +1,4 @@
-import asyncio
+from dailyai.services.base_transport_service import BaseTransportService
-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 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"])
+            frame = TranscriptionQueueFrame(
-            asyncio.run_coroutine_threadsafe(self.receive_queue.put(frame), self._loop)
+                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

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(

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_in = LLMUserContextAggregator(
-        tma_out = LLMAssistantContextAggregator(messages, transport._my_participant_id)
+            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()
+                        )
                     )
                 )
             )