小智终端 5种 场景模式 OK

wangzhibo

2026-09-11 659c470eba542383987aad227b7f918100f131fa

 xiaozhi-esp32-server/main/xiaozhi-server/core/providers/tts/base.py

@@ -117,8 +117,38 @@
        )

    def handle_opus(self, opus_data: bytes):
        self.last_stream_had_audio = True
        if not getattr(self, "_logged_first_opus", False):
            self._logged_first_opus = True
            logger.bind(tag=TAG).info(f"首帧opus已入队 bytes={len(opus_data)}")
        logger.bind(tag=TAG).debug(f"推送数据到队列里面帧数～～ {len(opus_data)}")
        self.tts_audio_queue.put((SentenceType.MIDDLE, opus_data, None, getattr(self, 'current_sentence_id', None)))

    def _run_text_to_speak(self, text, output_file=None):
        """Generate TTS with a hard timeout so LAST/stop is not blocked forever.

        Edge TTS (and some HTTP TTS APIs) can hang past asyncio.wait_for cancel.
        Run synthesis in a daemon thread and abandon it if the join times out.
        """
        result = {"data": None, "error": None}

        def worker():
            try:
                result["data"] = asyncio.run(self.text_to_speak(text, output_file))
            except Exception as e:
                result["error"] = e

        worker_thread = threading.Thread(target=worker, name="tts-synth", daemon=True)
        worker_thread.start()
        worker_thread.join(timeout=self.tts_timeout)
        if worker_thread.is_alive():
            logger.bind(tag=TAG).error(
                f"TTS timed out after {self.tts_timeout}s: {text}"
            )
            raise TimeoutError(f"TTS timed out after {self.tts_timeout}s")
        if result["error"] is not None:
            raise result["error"]
        return result["data"]

    def handle_audio_file(self, file_audio: bytes, text):
        self.before_stop_play_files.append((file_audio, text))
@@ -130,21 +160,42 @@
        # 使用正则一次性替换，避免重复遍历和部分匹配问题
        if self._correct_words_pattern:
            text = self._correct_words_pattern.sub(lambda m: self.correct_words[m.group(0)], text)
        max_repeat_time = 5
        # 先合成 opus，再下发 sentence_start。先发字幕再合成时，设备会进入说话中却收不到音频。
        max_repeat_time = 2
        opus_frames = {"n": 0}
        pending_opus = []
        self.last_stream_had_audio = False

        def _count_opus(data):
            opus_frames["n"] += 1
            pending_opus.append(bytes(data) if not isinstance(data, bytes) else data)

        def _flush_ready_audio():
            sid = getattr(self, "current_sentence_id", None)
            self.tts_audio_queue.put((SentenceType.FIRST, None, original_text, sid))
            for frame in pending_opus:
                if opus_handler:
                    opus_handler(frame)
                else:
                    self.tts_audio_queue.put((SentenceType.MIDDLE, frame, None, sid))
            self.last_stream_had_audio = True
            pending_opus.clear()

        counted_handler = _count_opus
        if self.delete_audio_file:
            # 需要删除文件的直接转为音频数据
            audio_bytes = None
            while max_repeat_time > 0:
                try:
                    audio_bytes = asyncio.run(self.text_to_speak(text, None))
                    audio_bytes = self._run_text_to_speak(text, None)
                    if audio_bytes:
                        # 使用原始文本用于显示/上报
                        self.tts_audio_queue.put((SentenceType.FIRST, None, original_text, getattr(self, 'current_sentence_id', None)))
                        encode_rate = getattr(self.opus_encoder, "sample_rate", None) or self.conn.sample_rate
                        audio_bytes_to_data_stream(
                            audio_bytes,
                            file_type=self.audio_file_type,
                            is_opus=True,
                            callback=opus_handler,
                            sample_rate=self.conn.sample_rate,
                            callback=counted_handler,
                            sample_rate=encode_rate,
                            opus_encoder=self.opus_encoder,
                        )
                        break
@@ -155,9 +206,14 @@
                        f"语音生成失败{5 - max_repeat_time + 1}次: {original_text}，错误: {e}"
                    )
                    max_repeat_time -= 1
            if max_repeat_time > 0:
            if max_repeat_time > 0 and opus_frames["n"] > 0:
                logger.bind(tag=TAG).info(
                    f"语音生成成功: {original_text}，重试{5 - max_repeat_time}次"
                    f"语音生成成功: {original_text}，opus帧={opus_frames['n']}，重试{5 - max_repeat_time}次"
                )
                _flush_ready_audio()
            elif max_repeat_time > 0:
                logger.bind(tag=TAG).error(
                    f"语音生成无opus帧: {original_text}，bytes={0 if not audio_bytes else len(audio_bytes)}"
                )
            else:
                logger.bind(tag=TAG).error(
@@ -169,7 +225,7 @@
            try:
                while not os.path.exists(tmp_file) and max_repeat_time > 0:
                    try:
                        asyncio.run(self.text_to_speak(text, tmp_file))
                        self._run_text_to_speak(text, tmp_file)
                    except Exception as e:
                        logger.bind(tag=TAG).warning(
                            f"语音生成失败{5 - max_repeat_time + 1}次: {original_text}，错误: {e}"
@@ -183,12 +239,17 @@
                    logger.bind(tag=TAG).info(
                        f"语音生成成功: {original_text}:{tmp_file}，重试{5 - max_repeat_time}次"
                    )
                    self._process_audio_file_stream(tmp_file, callback=counted_handler)
                    if opus_frames["n"] > 0:
                        _flush_ready_audio()
                    else:
                        logger.bind(tag=TAG).error(
                            f"语音生成无opus帧: {original_text}:{tmp_file}"
                        )
                else:
                    logger.bind(tag=TAG).error(
                        f"语音生成失败: {original_text}，请检查网络或服务是否正常"
                    )
                self.tts_audio_queue.put((SentenceType.FIRST, None, original_text, getattr(self, 'current_sentence_id', None)))
                self._process_audio_file_stream(tmp_file, callback=opus_handler)
            except Exception as e:
                logger.bind(tag=TAG).error(f"Failed to generate TTS file: {e}")
                return None
@@ -204,7 +265,7 @@
            # 需要删除文件的直接转为音频数据
            while max_repeat_time > 0:
                try:
                    audio_bytes = asyncio.run(self.text_to_speak(text, None))
                    audio_bytes = self._run_text_to_speak(text, None)
                    if audio_bytes:
                        audio_datas = []
                        audio_bytes_to_data_stream(
@@ -236,7 +297,7 @@
            try:
                while not os.path.exists(tmp_file) and max_repeat_time > 0:
                    try:
                        asyncio.run(self.text_to_speak(text, tmp_file))
                        self._run_text_to_speak(text, tmp_file)
                    except Exception as e:
                        logger.bind(tag=TAG).warning(
                            f"语音生成失败{5 - max_repeat_time + 1}次: {original_text}，错误: {e}"
@@ -306,24 +367,32 @@

    async def open_audio_channels(self, conn):
        self.conn = conn
        if not hasattr(self, "_open_lock"):
            self._open_lock = threading.Lock()

        # 根据conn的sample_rate创建编码器，如果子类已经创建则不覆盖（IndexTTS接口返回为24kHZ-待重采样处理）
        if not hasattr(self, 'opus_encoder') or self.opus_encoder is None:
            self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
                sample_rate=conn.sample_rate, channels=1, frame_size_ms=60
        with self._open_lock:
            encoder = getattr(self, "opus_encoder", None)
            encoder_rate = getattr(encoder, "sample_rate", None) if encoder is not None else None
            if encoder is None or encoder_rate != conn.sample_rate:
                self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
                    sample_rate=conn.sample_rate, channels=1, frame_size_ms=60
                )

            if getattr(self, "_channels_opened", False):
                play = getattr(self, "audio_play_priority_thread", None)
                text_th = getattr(self, "tts_priority_thread", None)
                if play and play.is_alive() and text_th and text_th.is_alive():
                    return

            self.tts_priority_thread = threading.Thread(
                target=self.tts_text_priority_thread, daemon=True
            )

        # tts 消化线程
        self.tts_priority_thread = threading.Thread(
            target=self.tts_text_priority_thread, daemon=True
        )
        self.tts_priority_thread.start()

        # 音频播放 消化线程
        self.audio_play_priority_thread = threading.Thread(
            target=self._audio_play_priority_thread, daemon=True
        )
        self.audio_play_priority_thread.start()
            self.tts_priority_thread.start()
            self.audio_play_priority_thread = threading.Thread(
                target=self._audio_play_priority_thread, daemon=True
            )
            self.audio_play_priority_thread.start()
            self._channels_opened = True

    def store_tts_text(self, sentence_id, text):
        """存储指定 sentence_id 对应的文本，用于流式TTS获取正确的字幕文本
@@ -385,11 +454,16 @@
                    self.tts_text_buff = []
                    self.is_first_sentence = True
                    self.tts_audio_first_sentence = True
                    self._logged_first_opus = False
                    self.last_stream_had_audio = False
                elif ContentType.TEXT == message.content_type:
                    self.tts_text_buff.append(message.content_detail)
                    segment_text = self._get_segment_text()
                    if segment_text:
                        self.to_tts_stream(segment_text, opus_handler=self.handle_opus)
                        try:
                            self.to_tts_stream(segment_text, opus_handler=self.handle_opus)
                        except Exception as e:
                            logger.bind(tag=TAG).error(f"TTS分段合成失败: {e}")
                elif ContentType.FILE == message.content_type:
                    self._process_remaining_text_stream(opus_handler=self.handle_opus)
                    tts_file = message.content_file
@@ -398,10 +472,18 @@
                            tts_file, callback=self.handle_opus
                        )
                if message.sentence_type == SentenceType.LAST:
                    self._process_remaining_text_stream(opus_handler=self.handle_opus)
                    self.tts_audio_queue.put(
                        (message.sentence_type, [], message.content_detail, message.sentence_id)
                    )
                    try:
                        self._process_remaining_text_stream(opus_handler=self.handle_opus)
                    except Exception as e:
                        logger.bind(tag=TAG).error(
                            f"处理剩余TTS文本失败: {e}",
                        )
                    if getattr(self, "last_stream_had_audio", False):
                        self.tts_audio_queue.put(
                            (message.sentence_type, [], message.content_detail, message.sentence_id)
                        )
                    else:
                        logger.bind(tag=TAG).error("跳过TTS stop：本句没有opus音频")

            except queue.Empty:
                continue
@@ -462,7 +544,7 @@
                    sendAudioMessage(self.conn, sentence_type, audio_datas, text, sentence_id),
                    self.conn.loop,
                )
                future.result()
                future.result(timeout=20)

                # 记录输出和报告
                if self.conn.max_output_size > 0 and text: