Add thread readiness checks and is_ready property across services

src/VoiceDialogue/main.py

@@ -1,3 +1,4 @@
+import time
 import typing
 from multiprocessing import Queue
 from pathlib import Path
@@ -79,6 +80,10 @@ def launch_system(
     audio_playing_worker = AudioStreamPlayer(audio_playing_queue=tts_generated_audio_queue)
     audio_playing_worker.start()
     threads.append(audio_playing_worker)
+
+    while not all([thread.is_ready for thread in threads]):
+        time.sleep(0.1)
+
     # audio_frame_probe.start_record()
     print(f'{"=" * 80}\n服务启动成功\n{"=" * 80}')
     for thread in threads:

src/VoiceDialogue/services/audio/aec_audio_capture.py

@@ -32,6 +32,8 @@ class EchoCancellingAudioCapture(BaseThread):
         audio_recorder.freeAudioData.argtypes = [ctypes.POINTER(ctypes.c_ubyte)]
         audio_recorder.startRecord()
 
+        self.is_ready = True
+
         try:
             while not self.stopped():
                 size = ctypes.c_int(0)

src/VoiceDialogue/services/audio/audio_answer.py

@@ -23,15 +23,8 @@ class TTSAudioGenerator(BaseThread):
         self.processed_answer_queue: Queue = processed_answer_queue
         self.tts_generated_audio_queue: Queue = tts_generated_audio_queue
 
-        device = "cpu"  # mps slower 11.66(cpu) vs 39.42(mps)
-        tts_config = self.setup_tts_config(device, voice_role)
-
-        self.tts_module = TTSModule(tts_config)
-        self.tts_module.setup_inference_params(
-            ref_audio=voice_role.reference_audio_path,
-            parallel_infer=False,
-            **voice_role.inference_parameters
-        )
+        self._device = "cpu"  # mps slower 11.66(cpu) vs 39.42(mps)
+        self._voice_role = voice_role
 
     def setup_tts_config(self, device, voice_role: VoiceModel):
         config = {
@@ -60,8 +53,18 @@ class TTSAudioGenerator(BaseThread):
 
     def run(self):
 
+        tts_config = self.setup_tts_config(self._device, self._voice_role)
+
+        self.tts_module = TTSModule(tts_config)
+        self.tts_module.setup_inference_params(
+            ref_audio=self._voice_role.reference_audio_path,
+            parallel_infer=False,
+            **self._voice_role.inference_parameters
+        )
         self.warmup()
 
+        self.is_ready = True
+
         while not self.stopped():
             try:
                 voice_task: VoiceTask = self.processed_answer_queue.get(block=False, timeout=0.1)

src/VoiceDialogue/services/audio/audio_player.py

@@ -1,4 +1,5 @@
 import tempfile
+import time
 from collections import OrderedDict
 from multiprocessing import Queue
 from queue import Empty
@@ -23,6 +24,8 @@ class AudioStreamPlayer(BaseThread):
         self.audio_playing_queue: Queue = audio_playing_queue
 
     def run(self):
+        self.is_ready = True
+
         while not self.stopped():
 
             try:
@@ -54,14 +57,14 @@ class AudioStreamPlayer(BaseThread):
                     if answer_id not in voice_state_manager.waiting_second_answer_mapping:
                         continue
 
-                # now = time.time()
-                # print(
-                #     f'整体耗时: {(now - voice_task.send_time):.2f}\n'
-                #     f'Whisper 耗时: {(voice_task.whisper_end_time - voice_task.whisper_start_time):.2f}\n'
-                #     f'LLM 耗时: {(voice_task.llm_end_time - voice_task.llm_start_time):.2f}\n'
-                #     f'TTS generate sentence: {voice_task.answer_sentence}\n'
-                #     f'TTS 耗时: {(voice_task.tts_end_time - voice_task.tts_start_time):.2f}\n\n'
-                # )
+                now = time.time()
+                print(
+                    f'整体耗时: {(now - voice_task.send_time):.2f}\n'
+                    f'Whisper/FunASR 耗时: {(voice_task.whisper_end_time - voice_task.whisper_start_time):.2f}\n'
+                    f'LLM 耗时: {(voice_task.llm_end_time - voice_task.llm_start_time):.2f}\n'
+                    f'TTS generate sentence: {voice_task.answer_sentence}\n'
+                    f'TTS 耗时: {(voice_task.tts_end_time - voice_task.tts_start_time):.2f}\n\n'
+                )
 
                 self.update_chat_history(voice_task)
 

src/VoiceDialogue/services/core/base.py

@@ -6,9 +6,21 @@ class BaseThread(threading.Thread):
     def __init__(self, group=None, target=None, name=None, args=(), kwargs=None, *, daemon=None):
         super().__init__(group, target, name, args, kwargs, daemon=daemon)
         self._stop_event = threading.Event()
+        self._is_ready_event = threading.Event()
 
     def stop(self):
         self._stop_event.set()
 
     def stopped(self):
         return self._stop_event.is_set()
+
+    @property
+    def is_ready(self):
+        return self._is_ready_event.is_set()
+
+    @is_ready.setter
+    def is_ready(self, value: bool):
+        if value:
+            self._is_ready_event.set()
+        else:
+            self._is_ready_event.clear()

src/VoiceDialogue/services/speech/asr_service.py

@@ -182,6 +182,8 @@ class ASRWorker(BaseThread):
         self.client = UnifiedASRClient(self.language)
         self.client.warmup()
 
+        self.is_ready = True
+
         while not self.stopped():
             voice_task: VoiceTask = self.user_voice_queue.get()
             voice_task.language = self.language

src/VoiceDialogue/services/speech/speech_monitor.py

@@ -210,6 +210,8 @@ class SpeechStateMonitor(BaseThread):
         主运行循环 - 监控语音状态并处理音频帧
         """
 
+        self.is_ready = True
+
         # 初始化状态变量
         audio_frames = np.array([])
         is_audio_sent_for_processing = False

src/VoiceDialogue/services/text/text_generator.py

@@ -182,6 +182,8 @@ class LLMResponseGenerator(BaseThread):
         pipeline = create_langchain_pipeline(self.model_instance, CHINESE_SYSTEM_PROMPT, self.get_session_history)
         warmup_langchain_pipeline(pipeline)
 
+        self.is_ready = True
+
         """主运行循环"""
         while not self.stopped():
             try: