High risk async refactor

server.py

@@ -86,8 +86,8 @@ async def websocket_endpoint(websocket: WebSocket):
 
     try:
         # Start translation and audio processing tasks
-        print("[SERVER] Calling translator.start_translation()...")
-        translator.start_translation()
+        print("[SERVER] Awaiting translator.start_translation()...")
+        await translator.start_translation()
         print("[SERVER] translator.start_translation() returned. Creating tasks...")
         output_sender_task = asyncio.create_task(
             audio_output_sender(websocket, translator.output_queue)

translator.py

@@ -1,11 +1,8 @@
 import asyncio
 import base64
 import json
-import queue
-import threading
-import time
 from collections import deque
-from typing import Dict, Optional
+from typing import Optional
 
 import deepl
 import websockets
@@ -22,18 +19,11 @@ class VoiceTranslator:
         self.audio_rate = 16000
         self.audio_chunk = 1024
 
-        self.input_queue = asyncio.Queue()
-        self.output_queue = asyncio.Queue()
-
-        self.lang_queues: Dict[str, queue.Queue] = {
-            "en-US": queue.Queue(),
-            "fr-FR": queue.Queue(),
-        }
+        self.input_queue = asyncio.Queue()  # Audio from browser
+        self.output_queue = asyncio.Queue() # Audio to browser
         self.prebuffer = deque(maxlen=12)
 
-        self.is_recording = False
-        self.is_speaking = False
-        self.speaking_event = threading.Event()
+        self.is_running = False
 
         self.last_processed_transcript = ""
         self.last_tts_text_en = ""
@@ -41,48 +31,19 @@ class VoiceTranslator:
         self.min_confidence_threshold = 0.5
 
         self.async_loop = asyncio.new_event_loop()
-        self.async_thread = threading.Thread(target=self._run_async_loop, daemon=True)
-        self.async_thread.start()
 
         self._tts_queue: "asyncio.Queue[Optional[dict]]" = asyncio.Queue()
         self._tts_consumer_task: Optional[asyncio.Task] = None
         self._process_audio_task: Optional[asyncio.Task] = None
-
-        def _start_consumer():
-            self._tts_consumer_task = asyncio.create_task(self._tts_consumer())
-
-        self.async_loop.call_soon_threadsafe(_start_consumer)
-
-        self.stt_threads: Dict[str, threading.Thread] = {}
-        self.restart_events: Dict[str, threading.Event] = {
-            "en-US": threading.Event(),
-            "fr-FR": threading.Event(),
-        }
-        self.stream_cancel_events: Dict[str, threading.Event] = {
-            "en-US": threading.Event(),
-            "fr-FR": threading.Event(),
-        }
-        self._stream_started = {"en-US": False, "fr-FR": False}
+        self.stt_tasks: list[asyncio.Task] = []
         self._tts_job_counter = 0
 
-    def _run_async_loop(self):
-        asyncio.set_event_loop(self.async_loop)
-        try:
-            self.async_loop.run_forever()
-        except Exception as e:
-            print(f"[async_loop] stopped with error: {e}")
-
     async def _process_input_audio(self):
         print("🎤 Audio processing task started...")
-        while self.is_recording:
+        while self.is_running:
             try:
                 data = await self.input_queue.get()
-                if data is None:
-                    break
-                if not self.speaking_event.is_set():
-                    self.prebuffer.append(data)
-                    self.lang_queues["en-US"].put(data)
-                    self.lang_queues["fr-FR"].put(data)
+                self.prebuffer.append(data)
             except asyncio.CancelledError:
                 break
             except Exception as e:
@@ -95,12 +56,7 @@ class VoiceTranslator:
             f"/stream-input?model_id=eleven_flash_v2_5&output_format=pcm_16000"
         )
         try:
-            self.is_speaking = True
-            self.speaking_event.set()
             self.prebuffer.clear()
-            for q in self.lang_queues.values():
-                with q.mutex:
-                    q.queue.clear()
 
             async with websockets.connect(uri) as websocket:
                 await websocket.send(json.dumps({
@@ -183,94 +139,60 @@ class VoiceTranslator:
         except Exception as e:
             print(f"Translation error: {e}")
 
-    def _run_stt_stream(self, language: str):
-        print(f"[stt:{language}] Thread starting...")
-        while self.is_recording:
-            try:
-                self.restart_events[language].wait()
-                self.restart_events[language].clear()
-
-                config = speech.RecognitionConfig(
-                    encoding=speech.RecognitionConfig.AudioEncoding.LINEAR16,
-                    sample_rate_hertz=self.audio_rate,
-                    language_code=language,
-                    enable_automatic_punctuation=True,
-                    model="latest_short",
-                )
-                streaming_config = speech.StreamingRecognitionConfig(
-                    config=config, interim_results=True, single_utterance=False
-                )
-
-                def request_generator():
-                    while self.is_recording:
-                        if self.speaking_event.is_set():
-                            time.sleep(0.01)
-                            continue
-                        if self.stream_cancel_events[language].is_set():
-                            self.stream_cancel_events[language].clear()
-                            break
-                        try:
-                            chunk = self.lang_queues[language].get(timeout=0.1)
-                            yield speech.StreamingRecognizeRequest(audio_content=chunk)
-                        except queue.Empty:
-                            continue
-
-                responses = self.stt_client.streaming_recognize(streaming_config, request_generator())
+    async def _run_stt_stream(self, language: str):
+        print(f"[stt:{language}] Task starting...")
+        config = speech.RecognitionConfig(
+            encoding=speech.RecognitionConfig.AudioEncoding.LINEAR16,
+            sample_rate_hertz=self.audio_rate,
+            language_code=language,
+            enable_automatic_punctuation=True,
+            model="latest_short",
+        )
+        streaming_config = speech.StreamingRecognitionConfig(
+            config=config, interim_results=True
+        )
 
-                for response in responses:
-                    if not self.is_recording:
-                        break
+        async def request_generator():
+            last_chunk_time = self.async_loop.time()
+            while self.is_running:
+                if self.prebuffer:
+                    chunk = self.prebuffer.popleft()
+                    last_chunk_time = self.async_loop.time()
+                    yield speech.StreamingRecognizeRequest(audio_content=chunk)
+                else:
+                    # If no audio for a while, send empty to keep stream alive
+                    if self.async_loop.time() - last_chunk_time > 0.5:
+                         yield speech.StreamingRecognizeRequest()
+                    await asyncio.sleep(0.1)
+
+        while self.is_running:
+            responses = self.stt_client.streaming_recognize(streaming_config, request_generator())
+            try:
+                async for response in responses:
+                    if not self.is_running: break
                     for result in response.results:
                         if result.is_final and result.alternatives:
                             alt = result.alternatives[0]
-                            transcript = alt.transcript.strip()
-                            conf = getattr(alt, "confidence", 0.0)
-                            if transcript and conf >= self.min_confidence_threshold:
-                                asyncio.run_coroutine_threadsafe(
-                                    self._process_result(transcript, conf, language), self.async_loop
-                                )
+                            await self._process_result(alt.transcript, alt.confidence, language)
             except Exception as e:
-                print(f"[stt:{language}] Error: {e}")
-                time.sleep(1)
-        print(f"[stt:{language}] Thread exiting")
+                print(f"[stt:{language}] Error: {e}. Restarting stream...")
+                await asyncio.sleep(1)
+        print(f"[stt:{language}] Task exiting.")
 
-    def start_translation(self):
-        if self.is_recording:
+    async def start_translation(self):
+        if self.is_running:
             return
-        self.is_recording = True
-
-        for ev in self.restart_events.values():
-            ev.clear()
-        self.speaking_event.clear()
-
-        def _start_tasks():
-            self._process_audio_task = asyncio.create_task(self._process_input_audio())
-
-        self.async_loop.call_soon_threadsafe(_start_tasks)
-
-        for lang in ("en-US", "fr-FR"):
-            thread = threading.Thread(target=self._run_stt_stream, args=(lang,), daemon=True)
-            self.stt_threads[lang] = thread
-            thread.start()
-            self.restart_events[lang].set()
+        self.is_running = True
+        self._tts_consumer_task = asyncio.create_task(self._tts_consumer())
+        self.stt_tasks.append(asyncio.create_task(self._run_stt_stream("en-US")))
+        self.stt_tasks.append(asyncio.create_task(self._run_stt_stream("fr-FR")))
 
     def stop_translation(self):
-        if not self.is_recording:
+        if not self.is_running:
             return
-        self.is_recording = False
-        for ev in self.restart_events.values():
-            ev.set()
-
-        def _cancel_tasks():
-            if self._process_audio_task:
-                self._process_audio_task.cancel()
-            if self._tts_queue:
-                self._tts_queue.put_nowait(None)
-
-        self.async_loop.call_soon_threadsafe(_cancel_tasks)
-
-    def cleanup(self):
-        self.stop_translation()
-        time.sleep(0.5)
-        if self.async_loop.is_running():
-            self.async_loop.call_soon_threadsafe(self.async_loop.stop)
+        self.is_running = False
+        for task in self.stt_tasks:
+            if task: task.cancel()
+        if self._tts_consumer_task:
+            self._tts_consumer_task.cancel()
+        self.stt_tasks = []