Starting refactor of transports into their own directory
This commit is contained in:
Moishe Lettvin
@@ -1,495 +0,0 @@
|
||||
from abc import abstractmethod
|
||||
import asyncio
|
||||
import itertools
|
||||
import logging
|
||||
import numpy as np
|
||||
import pyaudio
|
||||
import torch
|
||||
import queue
|
||||
import threading
|
||||
import time
|
||||
from typing import Any, AsyncGenerator
|
||||
from enum import Enum
|
||||
from dailyai.pipeline.frame_processor import FrameProcessor
|
||||
|
||||
from dailyai.pipeline.frames import (
|
||||
SendAppMessageFrame,
|
||||
AudioFrame,
|
||||
EndFrame,
|
||||
ImageFrame,
|
||||
Frame,
|
||||
PipelineStartedFrame,
|
||||
SpriteFrame,
|
||||
StartFrame,
|
||||
TextFrame,
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
)
|
||||
from dailyai.pipeline.pipeline import Pipeline
|
||||
from dailyai.services.ai_services import TTSService
|
||||
|
||||
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:
|
||||
|
||||
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
|
||||
self._vad_start_s = kwargs.get("vad_start_s") or 0.2
|
||||
self._vad_stop_s = kwargs.get("vad_stop_s") or 0.8
|
||||
self._context = kwargs.get("context") or []
|
||||
self._vad_enabled = kwargs.get("vad_enabled") or False
|
||||
|
||||
if self._vad_enabled and self._speaker_enabled:
|
||||
raise Exception(
|
||||
"Sorry, you can't use speaker_enabled and vad_enabled at the same time. Please set one to False."
|
||||
)
|
||||
|
||||
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
|
||||
self._user_is_speaking = False
|
||||
|
||||
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.completed_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, pipeline: Pipeline | None = None, override_pipeline_source_queue=True):
|
||||
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()
|
||||
|
||||
if self._vad_enabled:
|
||||
self._vad_thread = threading.Thread(target=self._vad, daemon=True)
|
||||
self._vad_thread.start()
|
||||
|
||||
pipeline_task = None
|
||||
if pipeline:
|
||||
pipeline_task = asyncio.create_task(
|
||||
self.run_pipeline(pipeline, override_pipeline_source_queue)
|
||||
)
|
||||
|
||||
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
|
||||
finally:
|
||||
# Do anything that must be done to clean up
|
||||
self._post_run()
|
||||
|
||||
self._stop_threads.set()
|
||||
|
||||
if pipeline_task:
|
||||
pipeline_task.cancel()
|
||||
|
||||
await self.send_queue.put(EndFrame())
|
||||
|
||||
await async_output_queue_marshal_task
|
||||
self._frame_consumer_thread.join()
|
||||
|
||||
if self._speaker_enabled:
|
||||
self._receive_audio_thread.join()
|
||||
|
||||
if self._vad_enabled:
|
||||
self._vad_thread.join()
|
||||
|
||||
async def run_pipeline(self, pipeline: Pipeline, override_pipeline_source_queue=True):
|
||||
pipeline.set_sink(self.send_queue)
|
||||
if override_pipeline_source_queue:
|
||||
pipeline.set_source(self.receive_queue)
|
||||
await pipeline.run_pipeline()
|
||||
|
||||
async def run_interruptible_pipeline(
|
||||
self,
|
||||
pipeline: Pipeline,
|
||||
allow_interruptions=True,
|
||||
pre_processor=None,
|
||||
post_processor: FrameProcessor | None = None,
|
||||
):
|
||||
pipeline.set_sink(self.send_queue)
|
||||
source_queue = asyncio.Queue()
|
||||
pipeline.set_source(source_queue)
|
||||
pipeline.set_sink(self.send_queue)
|
||||
pipeline_task = asyncio.create_task(pipeline.run_pipeline())
|
||||
|
||||
async def yield_frame(frame: Frame) -> AsyncGenerator[Frame, None]:
|
||||
yield frame
|
||||
|
||||
async def post_process(post_processor: FrameProcessor):
|
||||
while True:
|
||||
frame = await self.completed_queue.get()
|
||||
|
||||
# We ignore the output of the post_processor's process frame;
|
||||
# this is called to update the post-processor's state.
|
||||
async for frame in post_processor.process_frame(frame):
|
||||
pass
|
||||
|
||||
if isinstance(frame, EndFrame):
|
||||
break
|
||||
|
||||
if post_processor:
|
||||
post_process_task = asyncio.create_task(
|
||||
post_process(post_processor))
|
||||
|
||||
started = False
|
||||
|
||||
async for frame in self.get_receive_frames():
|
||||
if isinstance(frame, UserStartedSpeakingFrame):
|
||||
pipeline_task.cancel()
|
||||
self.interrupt()
|
||||
pipeline_task = asyncio.create_task(pipeline.run_pipeline())
|
||||
started = False
|
||||
|
||||
if not started:
|
||||
await self.send_queue.put(StartFrame())
|
||||
|
||||
if pre_processor:
|
||||
frame_generator = pre_processor.process_frame(frame)
|
||||
else:
|
||||
frame_generator = yield_frame(frame)
|
||||
|
||||
async for frame in frame_generator:
|
||||
await source_queue.put(frame)
|
||||
|
||||
if isinstance(frame, EndFrame):
|
||||
break
|
||||
|
||||
await asyncio.gather(pipeline_task, post_process_task)
|
||||
|
||||
async def say(self, text: str, tts: TTSService):
|
||||
"""Say a phrase. Use with caution; this bypasses any running pipelines."""
|
||||
async for frame in tts.process_frame(TextFrame(text)):
|
||||
await self.send_queue.put(frame)
|
||||
|
||||
def _post_run(self):
|
||||
# Note that this function must be idempotent! It can be called multiple times
|
||||
# if, for example, a keyboard interrupt occurs.
|
||||
pass
|
||||
|
||||
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
|
||||
|
||||
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
|
||||
):
|
||||
if self._loop:
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
self.receive_queue.put(
|
||||
UserStartedSpeakingFrame()), self._loop)
|
||||
# self.interrupt()
|
||||
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
|
||||
):
|
||||
if self._loop:
|
||||
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: Frame | list = await self.send_queue.get()
|
||||
self._threadsafe_send_queue.put(frame)
|
||||
self.send_queue.task_done()
|
||||
if isinstance(frame, EndFrame):
|
||||
break
|
||||
|
||||
def interrupt(self):
|
||||
self._logger.debug("### Interrupting")
|
||||
self._is_interrupted.set()
|
||||
|
||||
async def get_receive_frames(self) -> AsyncGenerator[Frame, None]:
|
||||
while True:
|
||||
frame = await self.receive_queue.get()
|
||||
yield frame
|
||||
if isinstance(frame, EndFrame):
|
||||
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 = AudioFrame(buffer)
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
self.receive_queue.put(frame), self._loop
|
||||
)
|
||||
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
self.receive_queue.put(
|
||||
EndFrame()), 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 send_app_message(self, message: Any, participantId: str | None):
|
||||
""" Child classes should override this to send a custom message to the room. """
|
||||
pass
|
||||
|
||||
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
|
||||
largest_write_size = 8000
|
||||
while True:
|
||||
try:
|
||||
frames_or_frame: Frame | list[Frame] = self._threadsafe_send_queue.get(
|
||||
)
|
||||
if (
|
||||
isinstance(frames_or_frame, AudioFrame)
|
||||
and len(frames_or_frame.data) > largest_write_size
|
||||
):
|
||||
# subdivide large audio frames to enable interruption
|
||||
frames = []
|
||||
for i in range(0, len(frames_or_frame.data),
|
||||
largest_write_size):
|
||||
frames.append(AudioFrame(
|
||||
frames_or_frame.data[i: i + largest_write_size]))
|
||||
elif isinstance(frames_or_frame, Frame):
|
||||
frames: list[Frame] = [frames_or_frame]
|
||||
elif isinstance(frames_or_frame, list):
|
||||
frames: list[Frame] = frames_or_frame
|
||||
else:
|
||||
raise Exception("Unknown type in output queue")
|
||||
|
||||
for frame in frames:
|
||||
if isinstance(frame, EndFrame):
|
||||
self._logger.info("Stopping frame consumer thread")
|
||||
self._stop_threads.set()
|
||||
self._threadsafe_send_queue.task_done()
|
||||
if self._loop:
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
self.completed_queue.put(frame), self._loop
|
||||
)
|
||||
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, AudioFrame):
|
||||
chunk = frame.data
|
||||
|
||||
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, ImageFrame):
|
||||
self._set_image(frame.image)
|
||||
elif isinstance(frame, SpriteFrame):
|
||||
self._set_images(frame.images)
|
||||
elif isinstance(frame, SendAppMessageFrame):
|
||||
self.send_app_message(
|
||||
frame.message, frame.participantId)
|
||||
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, StartFrame):
|
||||
self._is_interrupted.clear()
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
self.receive_queue.put(PipelineStartedFrame()),
|
||||
self._loop,
|
||||
)
|
||||
|
||||
if self._loop:
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
self.completed_queue.put(frame), self._loop
|
||||
)
|
||||
|
||||
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
|
||||
@@ -1,284 +0,0 @@
|
||||
import asyncio
|
||||
import inspect
|
||||
import logging
|
||||
import signal
|
||||
import threading
|
||||
import types
|
||||
|
||||
from functools import partial
|
||||
from typing import Any
|
||||
|
||||
from dailyai.pipeline.frames import (
|
||||
ReceivedAppMessageFrame,
|
||||
TranscriptionQueueFrame,
|
||||
)
|
||||
|
||||
from threading import Event
|
||||
|
||||
from daily import (
|
||||
EventHandler,
|
||||
CallClient,
|
||||
Daily,
|
||||
VirtualCameraDevice,
|
||||
VirtualMicrophoneDevice,
|
||||
VirtualSpeakerDevice,
|
||||
)
|
||||
|
||||
from dailyai.services.base_transport_service import BaseTransportService
|
||||
|
||||
|
||||
class DailyTransportService(BaseTransportService, EventHandler):
|
||||
_daily_initialized = False
|
||||
_lock = threading.Lock()
|
||||
|
||||
_speaker_enabled: bool
|
||||
_speaker_sample_rate: int
|
||||
_vad_enabled: bool
|
||||
|
||||
# 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,
|
||||
min_others_count: int = 1,
|
||||
start_transcription: bool = False,
|
||||
**kwargs,
|
||||
):
|
||||
# This will call BaseTransportService.__init__ method, not EventHandler
|
||||
super().__init__(**kwargs)
|
||||
|
||||
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._other_participant_has_joined = False
|
||||
self._my_participant_id = None
|
||||
|
||||
self.transcription_settings = {
|
||||
"language": "en",
|
||||
"tier": "nova",
|
||||
"model": "2-conversationalai",
|
||||
"profanity_filter": True,
|
||||
"redact": False,
|
||||
"extra": {
|
||||
"endpointing": True,
|
||||
"punctuate": False,
|
||||
},
|
||||
}
|
||||
|
||||
self._logger: logging.Logger = logging.getLogger("dailyai")
|
||||
|
||||
self._event_handlers = {}
|
||||
|
||||
def _patch_method(self, event_name, *args, **kwargs):
|
||||
try:
|
||||
for handler in self._event_handlers[event_name]:
|
||||
if inspect.iscoroutinefunction(handler):
|
||||
if self._loop:
|
||||
future = asyncio.run_coroutine_threadsafe(
|
||||
handler(*args, **kwargs), self._loop)
|
||||
|
||||
# wait for the coroutine to finish. This will also
|
||||
# raise any exceptions raised by the coroutine.
|
||||
future.result()
|
||||
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.")
|
||||
else:
|
||||
handler(*args, **kwargs)
|
||||
except Exception as e:
|
||||
self._logger.error(f"Exception in event handler {event_name}: {e}")
|
||||
raise e
|
||||
|
||||
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 event_name not in self._event_handlers:
|
||||
self._event_handlers[event_name] = [
|
||||
getattr(
|
||||
self, event_name), types.MethodType(
|
||||
handler, self)]
|
||||
setattr(self, event_name, partial(self._patch_method, event_name))
|
||||
else:
|
||||
self._event_handlers[event_name].append(
|
||||
types.MethodType(handler, self))
|
||||
|
||||
def event_handler(self, event_name: str):
|
||||
def decorator(handler):
|
||||
self.add_event_handler(event_name, handler)
|
||||
return handler
|
||||
|
||||
return decorator
|
||||
|
||||
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 send_app_message(self, message: Any, participantId: str | None):
|
||||
self.client.send_app_message(message, participantId)
|
||||
|
||||
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:
|
||||
Daily.init()
|
||||
DailyTransportService._daily_initialized = True
|
||||
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")
|
||||
|
||||
if self._speaker_enabled or self._vad_enabled:
|
||||
self._speaker: VirtualSpeakerDevice = Daily.create_speaker_device(
|
||||
"speaker", sample_rate=self._speaker_sample_rate, 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,
|
||||
client_settings={
|
||||
"inputs": {
|
||||
"camera": {
|
||||
"isEnabled": True,
|
||||
"settings": {
|
||||
"deviceId": "camera",
|
||||
},
|
||||
},
|
||||
"microphone": {
|
||||
"isEnabled": True,
|
||||
"settings": {
|
||||
"deviceId": "mic",
|
||||
"customConstraints": {
|
||||
"autoGainControl": {"exact": False},
|
||||
"echoCancellation": {"exact": False},
|
||||
"noiseSuppression": {"exact": False},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
"publishing": {
|
||||
"camera": {
|
||||
"sendSettings": {
|
||||
"maxQuality": "low",
|
||||
"encodings": {
|
||||
"low": {
|
||||
"maxBitrate": 250000,
|
||||
"scaleResolutionDownBy": 1.333,
|
||||
"maxFramerate": 8,
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
},
|
||||
},
|
||||
)
|
||||
self._my_participant_id = self.client.participants()["local"]["id"]
|
||||
|
||||
self.client.update_subscription_profiles({
|
||||
"base": {
|
||||
"camera": "unsubscribed",
|
||||
}
|
||||
})
|
||||
|
||||
if self._token and self._start_transcription:
|
||||
self.client.start_transcription(self.transcription_settings)
|
||||
|
||||
self.original_sigint_handler = signal.getsignal(signal.SIGINT)
|
||||
signal.signal(signal.SIGINT, self.process_interrupt_handler)
|
||||
|
||||
def process_interrupt_handler(self, signum, frame):
|
||||
self._post_run()
|
||||
if callable(self.original_sigint_handler):
|
||||
self.original_sigint_handler(signum, frame)
|
||||
|
||||
def _post_run(self):
|
||||
self.client.leave()
|
||||
self.client.release()
|
||||
|
||||
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}")
|
||||
pass
|
||||
|
||||
def dialout(self, number):
|
||||
self.client.start_dialout({"phoneNumber": number})
|
||||
|
||||
def start_recording(self):
|
||||
self.client.start_recording()
|
||||
|
||||
def on_error(self, error):
|
||||
self._logger.error(f"on_error: {error}")
|
||||
|
||||
def on_call_state_updated(self, state):
|
||||
pass
|
||||
|
||||
def on_participant_joined(self, participant):
|
||||
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._stop_threads.set()
|
||||
|
||||
def on_app_message(self, message: Any, sender: str):
|
||||
if self._loop:
|
||||
frame = ReceivedAppMessageFrame(message, sender)
|
||||
print(frame)
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
self.receive_queue.put(frame), self._loop
|
||||
)
|
||||
|
||||
def on_transcription_message(self, message: dict):
|
||||
if self._loop:
|
||||
participantId = ""
|
||||
if "participantId" in message:
|
||||
participantId = message["participantId"]
|
||||
elif "session_id" in message:
|
||||
participantId = message["session_id"]
|
||||
if self._my_participant_id and participantId != self._my_participant_id:
|
||||
frame = TranscriptionQueueFrame(
|
||||
message["text"], participantId, message["timestamp"])
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
self.receive_queue.put(frame), self._loop)
|
||||
|
||||
def on_transcription_error(self, message):
|
||||
self._logger.error(f"Transcription error: {message}")
|
||||
|
||||
def on_transcription_started(self, status):
|
||||
pass
|
||||
|
||||
def on_transcription_stopped(self, stopped_by, stopped_by_error):
|
||||
pass
|
||||
@@ -1,83 +0,0 @@
|
||||
import asyncio
|
||||
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
|
||||
)
|
||||
@@ -1,117 +0,0 @@
|
||||
import asyncio
|
||||
import time
|
||||
from typing import AsyncGenerator, List
|
||||
import websockets
|
||||
|
||||
from dailyai.pipeline.frame_processor import FrameProcessor
|
||||
from dailyai.pipeline.frames import AudioFrame, ControlFrame, EndFrame, Frame, TTSEndFrame, TTSStartFrame, TextFrame
|
||||
from dailyai.pipeline.pipeline import Pipeline
|
||||
from dailyai.serializers.protobuf_serializer import ProtobufFrameSerializer
|
||||
from dailyai.services.base_transport_service import BaseTransportService
|
||||
|
||||
|
||||
class WebSocketFrameProcessor(FrameProcessor):
|
||||
"""This FrameProcessor filters and mutates frames before they're sent over the websocket.
|
||||
This is necessary to aggregate audio frames into sizes that are cleanly playable by the client"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
audio_frame_size: int | None = None,
|
||||
sendable_frames: List[Frame] | None = None):
|
||||
super().__init__()
|
||||
if not audio_frame_size:
|
||||
raise ValueError("audio_frame_size must be provided")
|
||||
|
||||
self._audio_frame_size = audio_frame_size
|
||||
self._sendable_frames = sendable_frames or [TextFrame, AudioFrame]
|
||||
self._audio_buffer = bytes()
|
||||
self._in_tts_audio = False
|
||||
|
||||
async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]:
|
||||
if isinstance(frame, TTSStartFrame):
|
||||
self._in_tts_audio = True
|
||||
elif isinstance(frame, AudioFrame):
|
||||
if self._in_tts_audio:
|
||||
self._audio_buffer += frame.data
|
||||
while len(self._audio_buffer) >= self._audio_frame_size:
|
||||
yield AudioFrame(self._audio_buffer[:self._audio_frame_size])
|
||||
self._audio_buffer = self._audio_buffer[self._audio_frame_size:]
|
||||
elif isinstance(frame, TTSEndFrame):
|
||||
self._in_tts_audio = False
|
||||
if self._audio_buffer:
|
||||
yield AudioFrame(self._audio_buffer)
|
||||
self._audio_buffer = bytes()
|
||||
elif type(frame) in self._sendable_frames or isinstance(frame, ControlFrame):
|
||||
yield frame
|
||||
|
||||
|
||||
class WebsocketTransport(BaseTransportService):
|
||||
def __init__(self, **kwargs):
|
||||
super().__init__(**kwargs)
|
||||
self._sample_width = kwargs.get("sample_width", 2)
|
||||
self._n_channels = kwargs.get("n_channels", 1)
|
||||
self._port = kwargs.get("port", 8765)
|
||||
self._host = kwargs.get("host", "localhost")
|
||||
self._audio_frame_size = kwargs.get("audio_frame_size", 16000)
|
||||
self._sendable_frames = kwargs.get(
|
||||
"sendable_frames", [
|
||||
TextFrame, AudioFrame, TTSEndFrame, TTSStartFrame])
|
||||
self._serializer = kwargs.get("serializer", ProtobufFrameSerializer())
|
||||
|
||||
self._server: websockets.WebSocketServer | None = None
|
||||
self._websocket: websockets.WebSocketServerProtocol | None = None
|
||||
|
||||
self._connection_handlers = []
|
||||
|
||||
async def run(self, pipeline: Pipeline, override_pipeline_source_queue=True):
|
||||
self._stop_server_event = asyncio.Event()
|
||||
pipeline.set_sink(self.send_queue)
|
||||
if override_pipeline_source_queue:
|
||||
pipeline.set_source(self.receive_queue)
|
||||
|
||||
pipeline.add_processor(WebSocketFrameProcessor(
|
||||
audio_frame_size=self._audio_frame_size,
|
||||
sendable_frames=self._sendable_frames))
|
||||
|
||||
async def timeout():
|
||||
sleep_time = self._expiration - time.time()
|
||||
await asyncio.sleep(sleep_time)
|
||||
self._stop_server_event.set()
|
||||
|
||||
async def send_task():
|
||||
while not self._stop_server_event.is_set():
|
||||
frame = await self.send_queue.get()
|
||||
if isinstance(frame, EndFrame):
|
||||
self._stop_server_event.set()
|
||||
break
|
||||
if self._websocket and frame:
|
||||
proto = self._serializer.serialize(frame)
|
||||
await self._websocket.send(proto)
|
||||
|
||||
async def start_server():
|
||||
async with websockets.serve(self._websocket_handler, self._host, self._port) as server:
|
||||
self._logger.debug("Websocket server started.")
|
||||
await self._stop_server_event.wait()
|
||||
self._logger.debug("Websocket server stopped.")
|
||||
await self.receive_queue.put(EndFrame())
|
||||
|
||||
timeout_task = asyncio.create_task(timeout())
|
||||
await asyncio.gather(start_server(), send_task(), pipeline.run_pipeline())
|
||||
timeout_task.cancel()
|
||||
|
||||
def on_connection(self, handler):
|
||||
self._connection_handlers.append(handler)
|
||||
|
||||
async def _websocket_handler(self, websocket: websockets.WebSocketServerProtocol, path):
|
||||
if self._websocket:
|
||||
await self._websocket.close()
|
||||
self._logger.warning(
|
||||
"Got another websocket connection; closing first.")
|
||||
|
||||
for handler in self._connection_handlers:
|
||||
await handler()
|
||||
|
||||
self._websocket = websocket
|
||||
async for message in websocket:
|
||||
frame = self._serializer.deserialize(message)
|
||||
await self.receive_queue.put(frame)
|
||||
Reference in New Issue
Block a user