feat: 增加asr,tts文件上报功能

main/xiaozhi-server/core/connection.py

@@ -30,6 +30,7 @@ from core.mcp.manager import MCPManager
 from config.config_loader import get_private_config_from_api
 from config.manage_api_client import DeviceNotFoundException, DeviceBindException
 from core.utils.output_counter import add_device_output
+from core.handle.ttsReportHandle import enqueue_tts_report
 
 TAG = __name__
 
@@ -54,6 +55,7 @@ class ConnectionHandler:
 
         self.websocket = None
         self.headers = None
+        self.device_id = None
         self.client_ip = None
         self.client_ip_info = {}
         self.session_id = None
@@ -72,6 +74,13 @@ class ConnectionHandler:
         self.audio_play_queue = queue.Queue()
         self.executor = ThreadPoolExecutor(max_workers=10)
 
+        # 上报线程标志
+        self.session_open_time = time.time()
+        self.tts_report_queue = queue.Queue()
+        self.asr_report_queue = queue.Queue()
+        self.asr_report_thread = None
+        self.tts_report_thread = None
+
         # 依赖的组件
         self.vad = _vad
         self.asr = _asr
@@ -153,6 +162,7 @@ class ConnectionHandler:
 
             # 认证通过,继续处理
             self.websocket = ws
+            self.device_id = self.headers.get("device-id", None)
             self.session_id = str(uuid.uuid4())
 
             # 启动超时检查任务
@@ -301,6 +311,26 @@ class ConnectionHandler:
         self._initialize_memory()
         """加载意图识别"""
         self._initialize_intent()
+        """初始化上报线程"""
+        self._init_report_threads()
+
+    def _init_report_threads(self):
+        """初始化ASR和TTS上报线程"""
+        if self.asr_report_thread is None or not self.asr_report_thread.is_alive():
+            self.asr_report_thread = threading.Thread(
+                target=self._asr_report_worker,
+                daemon=True
+            )
+            self.asr_report_thread.start()
+            self.logger.bind(tag=TAG).info("ASR上报线程已启动")
+
+        if self.tts_report_thread is None or not self.tts_report_thread.is_alive():
+            self.tts_report_thread = threading.Thread(
+                target=self._tts_report_worker,
+                daemon=True
+            )
+            self.tts_report_thread.start()
+            self.logger.bind(tag=TAG).info("TTS上报线程已启动")
 
     def _initialize_private_config(self):
         read_config_from_api = self.config.get("read_config_from_api", False)
@@ -427,8 +457,7 @@ class ConnectionHandler:
 
     def _initialize_memory(self):
         """初始化记忆模块"""
-        device_id = self.headers.get("device-id", None)
-        self.memory.init_memory(device_id, self.llm)
+        self.memory.init_memory(self.device_id, self.llm)
 
     def _initialize_intent(self):
         if (
@@ -862,6 +891,9 @@ class ConnectionHandler:
                             f"TTS生成：文件路径: {tts_file}"
                         )
                         if os.path.exists(tts_file):
+                            # 在这里上报TTS数据（使用文件路径）
+                            enqueue_tts_report(self, text, tts_file)
+
                             opus_datas, duration = self.tts.audio_to_opus_data(tts_file)
                         else:
                             self.logger.bind(tag=TAG).error(
@@ -918,6 +950,72 @@ class ConnectionHandler:
                     f"audio_play_priority priority_thread: {text} {e}"
                 )
 
+    def _asr_report_worker(self):
+        """ASR上报工作线程"""
+        # 提前导入避免循环引用问题
+        from core.handle.asrReportHandle import report_asr
+
+        while not self.stop_event.is_set():
+            try:
+                # 从队列获取数据，设置超时以便定期检查停止事件
+                item = self.asr_report_queue.get(timeout=1)
+                if item is None:  # 检测毒丸对象
+                    break
+
+                text, file_path = item
+
+                try:
+                    # 执行上报（传入文件路径）
+                    await_result = report_asr(self, text, file_path)
+
+                    # 使用asyncio.run_coroutine_threadsafe执行异步操作
+                    future = asyncio.run_coroutine_threadsafe(await_result, self.loop)
+                    future.result()
+                except Exception as e:
+                    self.logger.bind(tag=TAG).error(f"ASR上报线程异常: {e}")
+                finally:
+                    # 标记任务完成
+                    self.asr_report_queue.task_done()
+            except queue.Empty:
+                continue
+            except Exception as e:
+                self.logger.bind(tag=TAG).error(f"ASR上报工作线程异常: {e}")
+
+        self.logger.bind(tag=TAG).info("ASR上报线程已退出")
+
+    def _tts_report_worker(self):
+        """TTS上报工作线程"""
+        # 提前导入避免循环引用问题
+        from core.handle.ttsReportHandle import report_tts
+
+        while not self.stop_event.is_set():
+            try:
+                # 从队列获取数据，设置超时以便定期检查停止事件
+                item = self.tts_report_queue.get(timeout=1)
+                if item is None:  # 检测毒丸对象
+                    break
+
+                text, audio_data = item
+
+                try:
+                    # 执行上报（传入二进制数据）
+                    await_result = report_tts(self, text, audio_data)
+
+                    # 使用asyncio.run_coroutine_threadsafe执行异步操作
+                    future = asyncio.run_coroutine_threadsafe(await_result, self.loop)
+                    future.result()
+                except Exception as e:
+                    self.logger.bind(tag=TAG).error(f"TTS上报线程异常: {e}")
+                finally:
+                    # 标记任务完成
+                    self.tts_report_queue.task_done()
+            except queue.Empty:
+                continue
+            except Exception as e:
+                self.logger.bind(tag=TAG).error(f"TTS上报工作线程异常: {e}")
+
+        self.logger.bind(tag=TAG).info("TTS上报线程已退出")
+
     def speak_and_play(self, text, text_index=0):
         if text is None or len(text) <= 0:
             self.logger.bind(tag=TAG).info(f"无需tts转换，query为空，{text}")
@@ -963,6 +1061,10 @@ class ConnectionHandler:
             self.executor.shutdown(wait=False, cancel_futures=True)
             self.executor = None
 
+        # 添加毒丸对象到上报队列确保线程退出
+        self.asr_report_queue.put(None)
+        self.tts_report_queue.put(None)
+
         # 清空任务队列
         self.clear_queues()