fix: close would lock

audio_stream_processor.py

@@ -1,22 +1,28 @@
 import subprocess
 from threading import Thread
-from queue import Queue
+from queue import Queue, Empty
 from typing import Iterator
+import threading
+import time
 
 class AudioStreamProcessor:
     def __init__(self):
         self.queue = Queue()
+        self._is_running = threading.Event()
+        self._is_running.set()
         self.thread = Thread(target=self._process_audio_streams)
         self.thread.start()
 
     def add_audio_stream(self, audio_stream: Iterator[bytes]):
-        self.queue.put(audio_stream)
+        if self._is_running.is_set():
+            self.queue.put(audio_stream)
 
     def _process_audio_streams(self):
-        while True:
-            audio_stream = self.queue.get()
-            if audio_stream is None:  # We'll use None as a sentinel to mark the end
-                break
+        while self._is_running.is_set() or not self.queue.empty():
+            try:
+                audio_stream = self.queue.get(timeout=1)
+            except Empty:
+                continue
             self._stream(audio_stream)
 
     def _stream(self, audio_stream: Iterator[bytes]):
@@ -38,5 +44,7 @@ class AudioStreamProcessor:
         mpv_process.wait()
 
     def close(self):
-        self.queue.put(None)  # Signal the processing thread to terminate
+        self._is_running.clear()
+        while not self.queue.empty():  # Wait until the queue is empty
+            time.sleep(0.1)
         self.thread.join()

streaming_chat_service.py

@@ -53,19 +53,19 @@ class StreamingChatService:
             chunk_message = chunk['choices'][0]['delta']
             if 'content' in chunk_message:
                 chunk_text = chunk_message['content']
-                print(chunk_text)
+                # print(chunk_text)
                 current_sentence += chunk_text
                 agent_response += chunk_text
                 text_to_speak = self._should_we_send_to_voice(current_sentence)
                 if text_to_speak:
                     stream = self._speech_service.stream(text_to_speak)
                     self._audio_processor.add_audio_stream(stream)
-                    
-                    # current_sentence should be reset to the text after the last sentence termination character
+                    print(text_to_speak)
                     current_sentence = current_sentence[len(text_to_speak):]
 
         if len(current_sentence) > 0:
             stream = self._speech_service.stream(current_sentence)
             self._audio_processor.add_audio_stream(stream)
+            print(current_sentence)
         self._messages.append({"role": "assistant", "content": agent_response})
         return agent_response