wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

prefix suspected private members (#13)

Browse Source
This commit is contained in:
Liza
2024-01-26 18:28:54 +01:00
committed by GitHub
parent fcceb32bd7
commit 8baf137511
1 changed files with 60 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

121
src/dailyai/services/daily_transport_service.py
View File

@@ -64,9 +64,8 @@ class DailyTransportService(EventHandler):
# handler to send frames, to ensure that sending isn't subject to pauses in the async thread. # handler to send frames, to ensure that sending isn't subject to pauses in the async thread.
self.threadsafe_send_queue = Queue() self.threadsafe_send_queue = Queue()
self.is_interrupted = Event() self._is_interrupted = Event()
self.stop_threads = Event() self._stop_threads = Event()
self.story_started = False
self.mic_enabled = False self.mic_enabled = False
self.mic_sample_rate = 16000 self.mic_sample_rate = 16000
self.camera_width = 1024 self.camera_width = 1024
@@ -78,11 +77,11 @@ class DailyTransportService(EventHandler):
self.send_queue = asyncio.Queue() self.send_queue = asyncio.Queue()
self.receive_queue = asyncio.Queue() self.receive_queue = asyncio.Queue()
self.other_participant_has_joined = False self._other_participant_has_joined = False
self.my_participant_id = None self.my_participant_id = None
self.camera_thread = None self._camera_thread = None
self.frame_consumer_thread = None self._frame_consumer_thread = None
self.transcription_settings = { self.transcription_settings = {
"language": "en", "language": "en",
@@ -96,28 +95,28 @@ class DailyTransportService(EventHandler):
}, },
} }
self.logger: logging.Logger = logging.getLogger("dailyai") self._logger: logging.Logger = logging.getLogger("dailyai")
self.event_handlers = {} self._event_handlers = {}
try: try:
self.loop = asyncio.get_running_loop() self._loop = asyncio.get_running_loop()
except RuntimeError: except RuntimeError:
self.loop = None self._loop = None
def patch_method(self, event_name, *args, **kwargs): def _patch_method(self, event_name, *args, **kwargs):
try: try:
for handler in self.event_handlers[event_name]: for handler in self._event_handlers[event_name]:
if inspect.iscoroutinefunction(handler): if inspect.iscoroutinefunction(handler):
if self.loop: if self._loop:
asyncio.run_coroutine_threadsafe(handler(*args, **kwargs), self.loop) asyncio.run_coroutine_threadsafe(handler(*args, **kwargs), self._loop)
else: else:
raise Exception( 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.") "No event loop to run coroutine. In order to use async event handlers, you must run the DailyTransportService in an asyncio event loop.")
else: else:
handler(*args, **kwargs) handler(*args, **kwargs)
except Exception as e: except Exception as e:
self.logger.error(f"Exception in event handler {event_name}: {e}") self._logger.error(f"Exception in event handler {event_name}: {e}")
raise e raise e
def add_event_handler(self, event_name: str, handler): def add_event_handler(self, event_name: str, handler):
@@ -133,9 +132,9 @@ class DailyTransportService(EventHandler):
getattr( getattr(
self, event_name), types.MethodType( self, event_name), types.MethodType(
handler, self)] handler, self)]
setattr(self, event_name, partial(self.patch_method, event_name)) setattr(self, event_name, partial(self._patch_method, event_name))
else: 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 event_handler(self, event_name: str):
def decorator(handler): def decorator(handler):
@@ -144,7 +143,7 @@ class DailyTransportService(EventHandler):
return decorator return decorator
def configure_daily(self): def _configure_daily(self):
# Only initialize Daily once # Only initialize Daily once
if not DailyTransportService._daily_initialized: if not DailyTransportService._daily_initialized:
with DailyTransportService._lock: with DailyTransportService._lock:
@@ -168,13 +167,13 @@ class DailyTransportService(EventHandler):
) )
Daily.select_speaker_device("speaker") Daily.select_speaker_device("speaker")
self.image: bytes | None = None self._image: bytes | None = None
self.camera_thread = Thread(target=self.run_camera, daemon=True) self._camera_thread = Thread(target=self._run_camera, daemon=True)
self.camera_thread.start() self._camera_thread.start()
self.logger.info("Starting frame consumer thread") self._logger.info("Starting frame consumer thread")
self.frame_consumer_thread = Thread(target=self.frame_consumer, daemon=True) self._frame_consumer_thread = Thread(target=self._frame_consumer, daemon=True)
self.frame_consumer_thread.start() self._frame_consumer_thread.start()
self.client.set_user_name(self.bot_name) 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)
@@ -230,8 +229,8 @@ class DailyTransportService(EventHandler):
buffer = self.speaker.read_frames(desired_frame_count) buffer = self.speaker.read_frames(desired_frame_count)
if len(buffer) > 0: if len(buffer) > 0:
frame = AudioQueueFrame(buffer) frame = AudioQueueFrame(buffer)
if self.loop: if self._loop:
asyncio.run_coroutine_threadsafe(self.receive_queue.put(frame), self.loop) asyncio.run_coroutine_threadsafe(self.receive_queue.put(frame), self._loop)
def interrupt(self): def interrupt(self):
self.is_interrupted.set() self.is_interrupted.set()
@@ -246,7 +245,7 @@ class DailyTransportService(EventHandler):
def get_async_send_queue(self): def get_async_send_queue(self):
return self.send_queue return self.send_queue
async def marshal_frames(self): async def _marshal_frames(self):
while True: while True:
frame: QueueFrame | list = await self.send_queue.get() frame: QueueFrame | list = await self.send_queue.get()
self.threadsafe_send_queue.put(frame) self.threadsafe_send_queue.put(frame)
@@ -254,83 +253,83 @@ class DailyTransportService(EventHandler):
if isinstance(frame, EndStreamQueueFrame): if isinstance(frame, EndStreamQueueFrame):
break break
async def wait_for_send_queue_to_empty(self): async def _wait_for_send_queue_to_empty(self):
await self.send_queue.join() await self.send_queue.join()
self.threadsafe_send_queue.join() self.threadsafe_send_queue.join()
async def stop_when_done(self): async def stop_when_done(self):
await self.wait_for_send_queue_to_empty() await self._wait_for_send_queue_to_empty()
self.stop() self.stop()
async def run(self) -> None: async def run(self) -> None:
self.configure_daily() self._configure_daily()
self.do_shutdown = False self._do_shutdown = False
async_output_queue_marshal_task = asyncio.create_task(self.marshal_frames()) async_output_queue_marshal_task = asyncio.create_task(self._marshal_frames())
try: try:
participant_count: int = len(self.client.participants()) participant_count: int = len(self.client.participants())
self.logger.info(f"{participant_count} participants in room") 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(): while time.time() < self.expiration and not self._do_shutdown and not self._stop_threads.is_set():
await asyncio.sleep(1) await asyncio.sleep(1)
except Exception as e: except Exception as e:
self.logger.error(f"Exception {e}") self._logger.error(f"Exception {e}")
raise e raise e
finally: finally:
self.client.leave() self.client.leave()
self.stop_threads.set() self._stop_threads.set()
await self.receive_queue.put(EndStreamQueueFrame()) await self.receive_queue.put(EndStreamQueueFrame())
await self.send_queue.put(EndStreamQueueFrame()) await self.send_queue.put(EndStreamQueueFrame())
await async_output_queue_marshal_task await async_output_queue_marshal_task
if self.camera_thread and self.camera_thread.is_alive(): if self._camera_thread and self._camera_thread.is_alive():
self.camera_thread.join() self._camera_thread.join()
if self.frame_consumer_thread and self.frame_consumer_thread.is_alive(): if self._frame_consumer_thread and self._frame_consumer_thread.is_alive():
self.frame_consumer_thread.join() self._frame_consumer_thread.join()
def stop(self): def stop(self):
self.stop_threads.set() self._stop_threads.set()
def on_first_other_participant_joined(self): def _on_first_other_participant_joined(self):
pass pass
def call_joined(self, join_data, client_error): def call_joined(self, join_data, client_error):
self.logger.info(f"Call_joined: {join_data}, {client_error}") self._logger.info(f"Call_joined: {join_data}, {client_error}")
if self.speaker_enabled: if self.speaker_enabled:
t = Thread(target=self._receive_audio, daemon=True) t = Thread(target=self._receive_audio, daemon=True)
t.start() t.start()
def on_error(self, error): def on_error(self, error):
self.logger.error(f"on_error: {error}") self._logger.error(f"on_error: {error}")
def on_call_state_updated(self, state): def on_call_state_updated(self, state):
pass pass
def on_participant_joined(self, participant): 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._other_participant_has_joined = True
self.on_first_other_participant_joined() self._on_first_other_participant_joined()
def on_participant_left(self, participant, reason): def on_participant_left(self, participant, reason):
if len(self.client.participants()) < self.min_others_count + 1: if len(self.client.participants()) < self.min_others_count + 1:
self.do_shutdown = True self._do_shutdown = True
pass pass
def on_app_message(self, message, sender): def on_app_message(self, message, sender):
pass pass
def on_transcription_message(self, message: dict): def on_transcription_message(self, message: dict):
if self.loop: if self._loop:
participantId = "" participantId = ""
if "participantId" in message: if "participantId" in message:
participantId = message["participantId"] participantId = message["participantId"]
elif "session_id" in message: elif "session_id" in message:
participantId = message["session_id"] participantId = message["session_id"]
frame = TranscriptionQueueFrame(message["text"], participantId, message["timestamp"]) frame = TranscriptionQueueFrame(message["text"], participantId, message["timestamp"])
asyncio.run_coroutine_threadsafe(self.receive_queue.put(frame), self.loop) asyncio.run_coroutine_threadsafe(self.receive_queue.put(frame), self._loop)
def on_transcription_stopped(self, stopped_by, stopped_by_error): def on_transcription_stopped(self, stopped_by, stopped_by_error):
pass pass
@@ -342,21 +341,21 @@ class DailyTransportService(EventHandler):
pass pass
def set_image(self, image: bytes): def set_image(self, image: bytes):
self.image: bytes | None = image self._image: bytes | None = image
def run_camera(self): def _run_camera(self):
try: try:
while not self.stop_threads.is_set(): while not self._stop_threads.is_set():
if self.image: if self._image:
self.camera.write_frame(self.image) self.camera.write_frame(self._image)
time.sleep(1.0 / 8) # 8 fps time.sleep(1.0 / 8) # 8 fps
except Exception as e: except Exception as e:
self.logger.error(f"Exception {e} in camera thread.") self._logger.error(f"Exception {e} in camera thread.")
raise e raise e
def frame_consumer(self): def _frame_consumer(self):
self.logger.info("🎬 Starting frame consumer thread") self._logger.info("🎬 Starting frame consumer thread")
b = bytearray() b = bytearray()
smallest_write_size = 3200 smallest_write_size = 3200
all_audio_frames = bytearray() all_audio_frames = bytearray()
@@ -372,12 +371,12 @@ class DailyTransportService(EventHandler):
for frame in frames: for frame in frames:
if isinstance(frame, EndStreamQueueFrame): if isinstance(frame, EndStreamQueueFrame):
self.logger.info("Stopping frame consumer thread") self._logger.info("Stopping frame consumer thread")
self.threadsafe_send_queue.task_done() self.threadsafe_send_queue.task_done()
return return
# if interrupted, we just pull frames off the queue and discard them # if interrupted, we just pull frames off the queue and discard them
if not self.is_interrupted.is_set(): if not self._is_interrupted.is_set():
if frame: if frame:
if isinstance(frame, AudioQueueFrame): if isinstance(frame, AudioQueueFrame):
chunk = frame.data chunk = frame.data
@@ -402,7 +401,7 @@ class DailyTransportService(EventHandler):
b = bytearray() b = bytearray()
if isinstance(frame, StartStreamQueueFrame): if isinstance(frame, StartStreamQueueFrame):
self.is_interrupted.clear() self._is_interrupted.clear()
self.threadsafe_send_queue.task_done() self.threadsafe_send_queue.task_done()
except Empty: except Empty: