update:初始化上传

core/connection.py

@@ -0,0 +1,228 @@
+import os
+import json
+import uuid
+import time
+import queue
+import asyncio
+import logging
+import threading
+import websockets
+from typing import Dict, Any
+from collections import deque
+from core.utils.util import is_segment
+from core.utils.dialogue import Message, Dialogue
+from core.handle.textHandle import handleTextMessage
+from core.handle.helloHandle import handleHelloMessage
+from core.utils.util import get_string_no_punctuation_or_emoji
+from concurrent.futures import ThreadPoolExecutor, TimeoutError
+from core.handle.audioHandle import handleAudioMessage, sendAudioMessage
+
+
+class ConnectionHandler:
+    def __init__(self, config: Dict[str, Any], _vad, _asr, _llm, _tts):
+        self.config = config
+        self.logger = logging.getLogger(__name__)
+
+        self.websocket = None
+        self.headers = None
+        self.session_id = None
+        self.prompt = None
+        self.welcome_msg = None
+
+        # 线程任务相关
+        self.loop = asyncio.get_event_loop()
+        self.stop_event = threading.Event()
+        self.tts_queue = queue.Queue()
+        self.executor = ThreadPoolExecutor(max_workers=10)
+        self.scheduled_tasks = deque()
+
+        # 依赖的组件
+        self.vad = _vad
+        self.asr = _asr
+        self.llm = _llm
+        self.tts = _tts
+        self.dialogue = None
+
+        # vad相关变量
+        self.client_audio_buffer = bytes()
+        self.client_have_voice = False
+        self.client_have_voice_last_time = 0.0
+        self.client_voice_stop = False
+
+        # asr相关变量
+        self.asr_audio = []
+        self.asr_server_receive = True
+
+        # llm相关变量
+        self.llm_finish_task = False
+        self.dialogue = Dialogue()
+
+        # tts相关变量
+        self.tts_first_text = None
+        self.tts_last_text = None
+        self.tts_start_speak_time = None
+        self.tts_duration = 0
+
+    async def handle_connection(self, ws):
+        self.websocket = ws
+        """处理单个WebSocket连接"""
+        self.headers = dict(self.websocket.request.headers)
+        self.logger.info(f"连接建立，请求头：\n{self.headers}")
+
+        self.welcome_msg = self.config["xiaozhi"]
+        self.session_id = str(uuid.uuid4())
+        self.welcome_msg["session_id"] = self.session_id
+        await self.websocket.send(json.dumps(self.welcome_msg))
+
+        await self.loop.run_in_executor(None, self._initialize_components)
+
+        tts_priority = threading.Thread(target=self._priority_thread, daemon=True)
+        tts_priority.start()
+
+        try:
+            async for message in self.websocket:
+                await self._route_message(message)
+        except websockets.exceptions.ConnectionClosed:
+            self.logger.info("客户端断开连接")
+            await self.close()
+
+    async def _route_message(self, message):
+        """消息路由"""
+        if isinstance(message, str):
+            await self._handle_text(message)
+        elif isinstance(message, bytes):
+            await handleAudioMessage(self, message)
+
+    async def _handle_text(self, message):
+        """处理文本消息"""
+        self.logger.info(f"收到文本消息：{message}")
+        try:
+            msg_json = json.loads(message)
+            if msg_json["type"] == "hello":
+                await handleHelloMessage(self, "你好")
+        except json.JSONDecodeError:
+            await handleTextMessage(self, message)
+
+    def _initialize_components(self):
+        self.prompt = self.config["prompt"]
+        # 赋予LLM时间观念
+        if "{date_time}" in self.prompt:
+            date_time = time.strftime("%Y-%m-%d %H:%M", time.localtime())
+            self.prompt = self.prompt.replace("{date_time}", date_time)
+        self.dialogue.put(Message(role="user", content=self.prompt))
+
+    def chat(self, query):
+        self.dialogue.put(Message(role="user", content=query))
+        response_message = []
+        start = 0
+        # 提交 LLM 任务
+        try:
+            start_time = time.time()  # 记录开始时间
+            llm_responses = self.llm.response(self, self.dialogue.get_llm_dialogue())
+        except Exception as e:
+            self.logger.error(f"LLM 处理出错 {query}: {e}")
+            return None
+        # 提交 TTS 任务到线程池
+        self.llm_finish_task = False
+        for content in llm_responses:
+            response_message.append(content)
+            end_time = time.time()  # 记录结束时间
+            self.logger.debug(f"大模型返回时间时间: {end_time - start_time} 秒, 生成token={content}")
+            if is_segment(response_message):
+                segment_text = "".join(response_message[start:])
+                segment_text = get_string_no_punctuation_or_emoji(segment_text)
+                if len(segment_text) > 0:
+                    self.recode_first_last_text(segment_text)
+                    future = self.executor.submit(self.speak_and_play, segment_text)
+                    self.tts_queue.put(future)
+                    start = len(response_message)
+
+        # 处理剩余的响应
+        if start < len(response_message):
+            segment_text = "".join(response_message[start:])
+            self.recode_first_last_text(segment_text)
+            future = self.executor.submit(self.speak_and_play, segment_text)
+            self.tts_queue.put(future)
+
+        self.llm_finish_task = True
+        # 更新对话
+        self.dialogue.put(Message(role="assistant", content="".join(response_message)))
+        self.logger.debug(json.dumps(self.dialogue.get_llm_dialogue(), indent=4, ensure_ascii=False))
+        return True
+
+    def _priority_thread(self):
+        while not self.stop_event.is_set():
+            text = None
+            try:
+                future = self.tts_queue.get()
+                text = None
+                try:
+                    tts_file, text = future.result(timeout=10)
+                    if os.path.exists(tts_file):
+                        opus_datas, duration = self.tts.wav_to_opus_data(tts_file)
+                    else:
+                        opus_datas = []
+                        duration = 0
+                except TimeoutError:
+                    self.logger.error("TTS 任务超时")
+                    continue
+                except Exception as e:
+                    self.logger.error(f"TTS 任务出错: {e}")
+                    continue
+                asyncio.run_coroutine_threadsafe(
+                    sendAudioMessage(self, opus_datas, duration, text), self.loop
+                )
+                if self.tts.delete_audio_file and os.path.exists(tts_file):
+                    os.remove(tts_file)
+            except Exception as e:
+                self.clearSpeakStatus()
+                asyncio.run_coroutine_threadsafe(
+                    self.websocket.send(json.dumps({"type": "tts", "state": "stop", "session_id": self.session_id})),
+                    self.loop
+                )
+                self.logger.error(f"tts_priority priority_thread: {text}{e}")
+
+    def speak_and_play(self, text):
+        if text is None or len(text) <= 0:
+            self.logger.info(f"无需tts转换，query为空，{text}")
+            return None
+        tts_file = self.tts.to_tts(text)
+        if tts_file is None:
+            self.logger.error(f"tts转换失败，{text}")
+            return None
+        self.logger.debug(f"TTS 文件生成完毕")
+        return tts_file, text
+
+    def clearSpeakStatus(self):
+        self.logger.debug(f"清除服务端讲话状态")
+        self.asr_server_receive = True
+        self.tts_last_text = None
+        self.tts_first_text = None
+        self.tts_duration = 0
+        self.tts_start_speak_time = None
+
+    def recode_first_last_text(self, text):
+        if not self.tts_first_text:
+            self.tts_first_text = text
+        self.tts_last_text = text
+
+    async def close(self):
+        """资源清理方法"""
+        self.stop_event.set()
+        self.executor.shutdown(wait=False)
+        if self.websocket:
+            await self.websocket.close()
+        self.logger.info("连接资源已释放")
+
+    def reset_vad_states(self):
+        self.client_audio_buffer = bytes()
+        self.client_have_voice = False
+        self.client_have_voice_last_time = 0
+        self.client_voice_stop = False
+        self.logger.debug("VAD states reset.")
+
+    def stop_all_tasks(self):
+        while self.scheduled_tasks:
+            task = self.scheduled_tasks.popleft()
+            task.cancel()
+        self.scheduled_tasks.clear()

core/handle/audioHandle.py

@@ -0,0 +1,112 @@
+import logging
+import json
+import asyncio
+import time
+from core.utils.util import remove_punctuation_and_length, get_string_no_punctuation_or_emoji
+
+logger = logging.getLogger(__name__)
+
+
+async def handleAudioMessage(conn, audio):
+    if not conn.asr_server_receive:
+        logger.debug(f"前期数据处理中，暂停接收")
+        return
+    have_voice = conn.vad.is_vad(conn, audio)
+
+    # 如果本次没有声音，本段也没声音，就把声音丢弃了
+    if have_voice == False and conn.client_have_voice == False:
+        conn.asr_audio.clear()
+        return
+    conn.asr_audio.append(audio)
+    # 如果本段有声音，且已经停止了
+    if conn.client_voice_stop:
+        conn.asr_server_receive = False
+        text, file_path = conn.asr.speech_to_text(conn.asr_audio, conn.session_id)
+        logger.info(f"识别文本: {text}")
+        text_len = remove_punctuation_and_length(text)
+        if text_len > 0:
+            stt_text = get_string_no_punctuation_or_emoji(text)
+            await conn.websocket.send(json.dumps({
+                "type": "stt",
+                "text": stt_text,
+                "session_id": conn.session_id}
+            ))
+            await conn.websocket.send(
+                json.dumps({
+                    "type": "llm",
+                    "text": "😊",
+                    "emotion": "happy",
+                    "session_id": conn.session_id}
+                ))
+            conn.executor.submit(conn.chat, text)
+        else:
+            conn.asr_server_receive = True
+        conn.asr_audio.clear()
+        conn.reset_vad_states()
+
+
+async def sendAudioMessage(conn, audios, duration, text):
+    base_delay = conn.tts_duration
+
+    if text == conn.tts_first_text:
+        conn.tts_start_speak_time = time.time()
+        await conn.websocket.send(json.dumps({
+            "type": "tts",
+            "state": "start",
+            "session_id": conn.session_id
+        }))
+
+    # 调度文字显示任务
+    text_task = asyncio.create_task(
+        schedule_with_interrupt(
+            base_delay - 0.5,
+            send_sentence_start(conn, text)
+        )
+    )
+    conn.scheduled_tasks.append(text_task)
+
+    conn.tts_duration = conn.tts_duration + duration
+
+    # 发送音频数据
+    for opus_packet in audios:
+        await conn.websocket.send(opus_packet)
+
+    if conn.llm_finish_task and text == conn.tts_last_text:
+        stop_duration = conn.tts_duration - (time.time() - conn.tts_start_speak_time)
+        stop_task = asyncio.create_task(
+            schedule_with_interrupt(stop_duration, send_tts_stop(conn, text))
+        )
+        conn.scheduled_tasks.append(stop_task)
+
+
+async def send_sentence_start(conn, text):
+    await conn.websocket.send(json.dumps({
+        "type": "tts",
+        "state": "sentence_start",
+        "text": text,
+        "session_id": conn.session_id
+    }))
+
+
+async def send_tts_stop(conn, text):
+    await conn.websocket.send(json.dumps({
+        "type": "tts",
+        "state": "sentence_end",
+        "text": text,
+        "session_id": conn.session_id
+    }))
+    await conn.websocket.send(json.dumps({
+        "type": "tts",
+        "state": "stop",
+        "session_id": conn.session_id
+    }))
+    conn.clearSpeakStatus()
+
+
+async def schedule_with_interrupt(delay, coro):
+    """可中断的延迟调度"""
+    try:
+        await asyncio.sleep(delay)
+        await coro
+    except asyncio.CancelledError:
+        pass

core/handle/helloHandle.py

@@ -0,0 +1,8 @@
+import json
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+async def handleHelloMessage(conn, text):
+    await conn.websocket.send(json.dumps(conn.welcome_msg))

core/handle/textHandle.py

@@ -0,0 +1,7 @@
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+async def handleTextMessage(conn, message):
+    await conn.websocket.send(message)

core/server.py

@@ -0,0 +1,54 @@
+import asyncio
+import websockets
+import logging
+from core.connection import ConnectionHandler
+from core.utils.util import get_local_ip
+from core.utils import asr, vad, llm, tts
+
+
+class WebSocketServer:
+    def __init__(self, config: dict):
+        self.config = config
+        self.logger = logging.getLogger(__name__)
+        self._vad, self._asr, self._llm, self._tts = self._create_processing_instances()
+
+    def _create_processing_instances(self):
+        """创建处理模块实例"""
+        return (
+            vad.create_instance(
+                self.config["selected_module"]["VAD"],
+                self.config["VAD"][self.config["selected_module"]["VAD"]]
+            ),
+            asr.create_instance(
+                self.config["selected_module"]["ASR"],
+                self.config["ASR"][self.config["selected_module"]["ASR"]],
+                self.config["delete_audio"]
+            ),
+            llm.create_instance(
+                self.config["selected_module"]["LLM"],
+                self.config["LLM"][self.config["selected_module"]["LLM"]],
+            ),
+            tts.create_instance(
+                self.config["selected_module"]["TTS"],
+                self.config["TTS"][self.config["selected_module"]["TTS"]],
+                self.config["delete_audio"]
+            )
+        )
+
+    async def start(self):
+        server_config = self.config["server"]
+        host = server_config["ip"]
+        port = server_config["port"]
+
+        self.logger.info("Server is running at ws://%s:%s", get_local_ip(), port)
+        async with websockets.serve(
+                self._handle_connection,
+                host,
+                port
+        ):
+            await asyncio.Future()
+
+    async def _handle_connection(self, websocket):
+        """处理新连接，每次创建独立的ConnectionHandler"""
+        handler = ConnectionHandler(self.config, self._vad, self._asr, self._llm, self._tts)
+        await handler.handle_connection(websocket)

core/utils/asr.py

@@ -0,0 +1,114 @@
+import time
+import wave
+import os
+from abc import ABC, abstractmethod
+import logging
+from typing import Optional, Tuple, List
+import uuid
+
+import opuslib
+from funasr import AutoModel
+from funasr.utils.postprocess_utils import rich_transcription_postprocess
+
+logger = logging.getLogger(__name__)
+
+
+class ASR(ABC):
+    @abstractmethod
+    def save_audio_to_file(self, opus_data: List[bytes], session_id: str) -> str:
+        """解码Opus数据并保存为WAV文件"""
+        pass
+
+    @abstractmethod
+    def speech_to_text(self, opus_data: List[bytes], session_id: str) -> Tuple[Optional[str], Optional[str]]:
+        """将语音数据转换为文本"""
+        pass
+
+
+class FunASR(ASR):
+    def __init__(self, config: dict, delete_audio_file: bool):
+        self.model_dir = config.get("model_dir")
+        self.output_dir = config.get("output_dir")  # 修正配置键名
+        self.delete_audio_file = delete_audio_file
+
+        # 确保输出目录存在
+        os.makedirs(self.output_dir, exist_ok=True)
+
+        self.model = AutoModel(
+            model=self.model_dir,
+            vad_kwargs={"max_single_segment_time": 30000},
+            disable_update=True,
+            hub="hf"
+            # device="cuda:0",  # 启用GPU加速
+        )
+
+    def save_audio_to_file(self, opus_data: List[bytes], session_id: str) -> str:
+        """将Opus音频数据解码并保存为WAV文件"""
+        file_name = f"asr_{session_id}_{uuid.uuid4()}.wav"
+        file_path = os.path.join(self.output_dir, file_name)
+
+        decoder = opuslib.Decoder(16000, 1)  # 16kHz, 单声道
+        pcm_data = []
+
+        for opus_packet in opus_data:
+            try:
+                pcm_frame = decoder.decode(opus_packet, 960)  # 960 samples = 60ms
+                pcm_data.append(pcm_frame)
+            except opuslib.OpusError as e:
+                logger.error(f"Opus解码错误: {e}", exc_info=True)
+
+        with wave.open(file_path, "wb") as wf:
+            wf.setnchannels(1)
+            wf.setsampwidth(2)  # 2 bytes = 16-bit
+            wf.setframerate(16000)
+            wf.writeframes(b"".join(pcm_data))
+
+        return file_path
+
+    def speech_to_text(self, opus_data: List[bytes], session_id: str) -> Tuple[Optional[str], Optional[str]]:
+        """语音转文本主处理逻辑"""
+        file_path = None
+        try:
+            # 保存音频文件
+            start_time = time.time()
+            file_path = self.save_audio_to_file(opus_data, session_id)
+            logger.debug(f"音频文件保存耗时: {time.time() - start_time:.3f}s | 路径: {file_path}")
+
+            # 语音识别
+            start_time = time.time()
+            result = self.model.generate(
+                input=file_path,
+                cache={},
+                language="auto",
+                use_itn=True,
+                batch_size_s=60,
+            )
+            text = rich_transcription_postprocess(result[0]["text"])
+            logger.debug(f"语音识别耗时: {time.time() - start_time:.3f}s | 结果: {text}")
+
+            return text, file_path
+
+        except Exception as e:
+            logger.error(f"语音识别失败: {e}", exc_info=True)
+            return None, None
+
+        finally:
+            # 文件清理逻辑
+            if self.delete_audio_file and file_path and os.path.exists(file_path):
+                try:
+                    os.remove(file_path)
+                    logger.debug(f"已删除临时音频文件: {file_path}")
+                except Exception as e:
+                    logger.error(f"文件删除失败: {file_path} | 错误: {e}")
+
+
+def create_instance(class_name: str, *args, **kwargs) -> ASR:
+    """工厂方法创建ASR实例"""
+    cls_map = {
+        "FunASR": FunASR,
+        # 可扩展其他ASR实现
+    }
+
+    if cls := cls_map.get(class_name):
+        return cls(*args, **kwargs)
+    raise ValueError(f"不支持的ASR类型: {class_name}")

core/utils/dialogue.py

@@ -0,0 +1,26 @@
+import uuid
+from typing import List, Dict
+from datetime import datetime
+
+
+class Message:
+    def __init__(self, role: str, content: str = None, uniq_id: str = None):
+        self.uniq_id = uniq_id if uniq_id is not None else str(uuid.uuid4())
+        self.role = role
+        self.content = content
+
+
+class Dialogue:
+    def __init__(self):
+        self.dialogue: List[Message] = []
+        # 获取当前时间
+        self.current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+
+    def put(self, message: Message):
+        self.dialogue.append(message)
+
+    def get_llm_dialogue(self) -> List[Dict[str, str]]:
+        dialogue = []
+        for m in self.dialogue:
+            dialogue.append({"role": m.role, "content": m.content})
+        return dialogue

core/utils/llm.py

@@ -0,0 +1,111 @@
+import json
+import logging
+import openai
+import requests
+from abc import ABC, abstractmethod
+
+logger = logging.getLogger(__name__)
+
+
+class LLM(ABC):
+    @abstractmethod
+    def response(self, conn, dialogue):
+        """LLM response generator"""
+        pass
+
+
+class DeepSeekLLM(LLM):
+    def __init__(self, config):
+        self.model_name = config.get("model_name")
+        self.api_key = config.get("api_key")
+        self.base_url = config.get("url")
+        self.client = openai.OpenAI(api_key=self.api_key, base_url=self.base_url)
+
+    def response(self, conn, dialogue):
+        logger.info(f"Generating response using {dialogue}")
+        try:
+            responses = self.client.chat.completions.create(
+                model=self.model_name,
+                messages=dialogue,
+                stream=True
+            )
+            for chunk in responses:
+                # 检查是否存在有效的choice且content不为空
+                if chunk.choices and len(chunk.choices) > 0:
+                    delta = chunk.choices[0].delta
+                    content = getattr(delta, 'content', '')
+                    if content:  # 仅在content非空时生成
+                        yield content
+        except Exception as e:
+            logger.error(f"Error in response generation: {e}")
+
+
+class ChatGLMLLM(LLM):
+    def __init__(self, config):
+        self.model_name = config.get("model_name")
+        self.api_key = config.get("api_key")
+        self.base_url = config.get("url")
+        self.client = openai.OpenAI(api_key=self.api_key, base_url=self.base_url)
+
+    def response(self, conn, dialogue):
+        try:
+            responses = self.client.chat.completions.create(
+                model=self.model_name,
+                messages=dialogue,
+                stream=True
+            )
+            for chunk in responses:
+                # 检查是否存在有效的choice且content不为空
+                if chunk.choices and len(chunk.choices) > 0:
+                    delta = chunk.choices[0].delta
+                    content = getattr(delta, 'content', '')
+                    if content:  # 仅在content非空时生成
+                        yield content
+        except Exception as e:
+            logger.error(f"Error in response generation: {e}")
+
+
+class DifyLLM(LLM):
+    def __init__(self, config):
+        self.api_key = config["api_key"]
+        self.base_url = config.get("base_url", "https://api.dify.ai/v1").rstrip('/')
+
+    def response(self, conn,dialogue):
+        try:
+            # 取最后一条用户消息
+            last_msg = next(m for m in reversed(dialogue) if m["role"] == "user")
+
+            # 发起流式请求
+            with requests.post(
+                    f"{self.base_url}/chat-messages",
+                    headers={"Authorization": f"Bearer {self.api_key}"},
+                    json={
+                        "query": last_msg["content"],
+                        "response_mode": "streaming",
+                        "user": conn.session_id,
+                        "inputs": {}
+                    },
+                    stream=True
+            ) as r:
+                for line in r.iter_lines():
+                    if line.startswith(b'data: '):
+                        event = json.loads(line[6:])
+                        if event.get('answer'):
+                            yield event['answer']
+
+        except Exception:
+            yield "【服务响应异常】"
+
+
+def create_instance(class_name, *args, **kwargs):
+    # 获取类对象
+    cls_map = {
+        "DeepSeekLLM": DeepSeekLLM,
+        "ChatGLMLLM": ChatGLMLLM,
+        "DifyLLM": DifyLLM,
+        # 可扩展其他LLM实现
+    }
+
+    if cls := cls_map.get(class_name):
+        return cls(*args, **kwargs)
+    raise ValueError(f"不支持的LLM类型: {class_name}")

core/utils/tts.py

@@ -0,0 +1,176 @@
+import asyncio
+import logging
+import os
+import json
+import uuid
+import base64
+from datetime import datetime
+import edge_tts
+import numpy as np
+import opuslib
+import requests
+from core.utils.util import read_config, get_project_dir
+from pydub import AudioSegment
+from abc import ABC, abstractmethod
+
+logger = logging.getLogger(__name__)
+
+
+class TTS(ABC):
+    def __init__(self, config, delete_audio_file):
+        self.delete_audio_file = delete_audio_file
+        self.output_file = config.get("output_file")
+        self.delete_audio_file = delete_audio_file
+
+    @abstractmethod
+    def generate_filename(self):
+        pass
+
+    def to_tts(self, text):
+        tmp_file = self.generate_filename()
+        try:
+            max_repeat_time = 5
+            while not os.path.exists(tmp_file) and max_repeat_time > 0:
+                asyncio.run(self.text_to_speak(text, tmp_file))
+                if not os.path.exists(tmp_file):
+                    max_repeat_time = max_repeat_time - 1
+                    logger.error(f"语音生成失败: {text}:{tmp_file}，再试{max_repeat_time}次")
+
+            return tmp_file
+        except Exception as e:
+            logger.info(f"Failed to generate TTS file: {e}")
+            return None
+
+    @abstractmethod
+    async def text_to_speak(self, text, output_file):
+        pass
+
+    def wav_to_opus_data(self, wav_file_path):
+        # 使用pydub加载PCM文件
+        # 获取文件后缀名
+        file_type = os.path.splitext(wav_file_path)[1]
+        if file_type:
+            file_type = file_type.lstrip('.')
+        audio = AudioSegment.from_file(wav_file_path, format=file_type)
+
+        duration = len(audio) / 1000.0
+
+        # 转换为单声道和16kHz采样率（确保与编码器匹配）
+        audio = audio.set_channels(1).set_frame_rate(16000)
+
+        # 获取原始PCM数据（16位小端）
+        raw_data = audio.raw_data
+
+        # 初始化Opus编码器
+        encoder = opuslib.Encoder(16000, 1, opuslib.APPLICATION_AUDIO)
+
+        # 编码参数
+        frame_duration = 60  # 60ms per frame
+        frame_size = int(16000 * frame_duration / 1000)  # 960 samples/frame
+
+        opus_datas = []
+        # 按帧处理所有音频数据（包括最后一帧可能补零）
+        for i in range(0, len(raw_data), frame_size * 2):  # 16bit=2bytes/sample
+            # 获取当前帧的二进制数据
+            chunk = raw_data[i:i + frame_size * 2]
+
+            # 如果最后一帧不足，补零
+            if len(chunk) < frame_size * 2:
+                chunk += b'\x00' * (frame_size * 2 - len(chunk))
+
+            # 转换为numpy数组处理
+            np_frame = np.frombuffer(chunk, dtype=np.int16)
+
+            # 编码Opus数据
+            opus_data = encoder.encode(np_frame.tobytes(), frame_size)
+            opus_datas.append(opus_data)
+
+        return opus_datas, duration
+
+
+class EdgeTTS(TTS):
+    def __init__(self, config, delete_audio_file):
+        super().__init__(config, delete_audio_file)
+        self.voice = config.get("voice")
+
+    def generate_filename(self, extension=".mp3"):
+        return os.path.join(self.output_file, f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}")
+
+    async def text_to_speak(self, text, output_file):
+        communicate = edge_tts.Communicate(text, voice=self.voice)  # Use your preferred voice
+        await communicate.save(output_file)
+
+
+class DoubaoTTS(TTS):
+    def __init__(self, config, delete_audio_file):
+        super().__init__(config, delete_audio_file)
+        self.appid = config.get("appid")
+        self.access_token = config.get("access_token")
+        self.cluster = config.get("cluster")
+        self.voice = config.get("voice")
+
+        self.host = "openspeech.bytedance.com"
+        self.api_url = f"https://{self.host}/api/v1/tts"
+        self.header = {"Authorization": f"Bearer;{self.access_token}"}
+
+    def generate_filename(self, extension=".wav"):
+        return os.path.join(self.output_file, f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}")
+
+    async def text_to_speak(self, text, output_file):
+        request_json = {
+            "app": {
+                "appid": self.appid,
+                "token": "access_token",
+                "cluster": self.cluster
+            },
+            "user": {
+                "uid": "1"
+            },
+            "audio": {
+                "voice_type": self.voice,
+                "encoding": "wav",
+                "speed_ratio": 1.0,
+                "volume_ratio": 1.0,
+                "pitch_ratio": 1.0,
+            },
+            "request": {
+                "reqid": str(uuid.uuid4()),
+                "text": text,
+                "text_type": "plain",
+                "operation": "query",
+                "with_frontend": 1,
+                "frontend_type": "unitTson"
+            }
+        }
+
+        resp = requests.post(self.api_url, json.dumps(request_json), headers=self.header)
+        if "data" in resp.json():
+            data = resp.json()["data"]
+            file_to_save = open(output_file, "wb")
+            file_to_save.write(base64.b64decode(data))
+
+
+def create_instance(class_name, *args, **kwargs):
+    # 获取类对象
+    cls_map = {
+        "DoubaoTTS": DoubaoTTS,
+        "EdgeTTS": EdgeTTS,
+        # 可扩展其他TTS实现
+    }
+
+    if cls := cls_map.get(class_name):
+        return cls(*args, **kwargs)
+    raise ValueError(f"不支持的TTS类型: {class_name}")
+
+
+if __name__ == "__main__":
+    config = read_config(get_project_dir() + "config.yaml")
+    tts = create_instance(
+        config["selected_module"]["TTS"],
+        config["TTS"][config["selected_module"]["TTS"]],
+        config["delete_audio"]
+    )
+    tts.output_file = get_project_dir() + tts.output_file
+    file_path = tts.to_tts("你好，测试")
+    print(file_path)
+    print(tts.wav_to_opus_data(file_path))

core/utils/util.py

@@ -0,0 +1,92 @@
+import yaml
+import unicodedata
+import socket
+import os
+import json
+
+
+def get_project_dir():
+    projectName = 'xiaozhi-esp32-server'
+    filePath = os.path.abspath(__file__)
+    return filePath[:filePath.rfind('/' + projectName + '/') + len(projectName) + 2]
+
+
+def get_local_ip():
+    try:
+        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        # Connect to Google's DNS servers
+        s.connect(("8.8.8.8", 80))
+        local_ip = s.getsockname()[0]
+        s.close()
+        return local_ip
+    except Exception as e:
+        return "127.0.0.1"
+
+
+def read_config(config_path):
+    with open(config_path, "r", encoding="utf-8") as file:
+        config = yaml.safe_load(file)
+    return config
+
+
+def write_json_file(file_path, data):
+    """将数据写入 JSON 文件"""
+    with open(file_path, 'w', encoding='utf-8') as file:
+        json.dump(data, file, ensure_ascii=False, indent=4)
+
+
+def is_segment(tokens):
+    if tokens[-1] in (",", ".", "?", "，", "。", "？", "！", "!", ";", "；", ":", "："):
+        return True
+    else:
+        return False
+
+def is_punctuation_or_emoji(char):
+    """检查字符是否为空格、指定标点或表情符号"""
+    # 定义需要去除的中英文标点（包括全角/半角）
+    punctuation_set = {
+        '，', ',',  # 中文逗号 + 英文逗号
+        '。', '.',  # 中文句号 + 英文句号
+        '！', '!',  # 中文感叹号 + 英文感叹号
+        '-', '－',  # 英文连字符 + 中文全角横线
+        '、'         # 中文顿号
+    }
+    if char.isspace() or char in punctuation_set:
+        return True
+    # 检查表情符号（保留原有逻辑）
+    code_point = ord(char)
+    emoji_ranges = [
+        (0x1F600, 0x1F64F), (0x1F300, 0x1F5FF),
+        (0x1F680, 0x1F6FF), (0x1F900, 0x1F9FF),
+        (0x1FA70, 0x1FAFF), (0x2600, 0x26FF),
+        (0x2700, 0x27BF)
+    ]
+    return any(start <= code_point <= end for start, end in emoji_ranges)
+
+def get_string_no_punctuation_or_emoji(s):
+    """去除字符串首尾的空格、标点符号和表情符号"""
+    chars = list(s)
+    # 处理开头的字符
+    start = 0
+    while start < len(chars) and is_punctuation_or_emoji(chars[start]):
+        start += 1
+    # 处理结尾的字符
+    end = len(chars) - 1
+    while end >= start and is_punctuation_or_emoji(chars[end]):
+        end -= 1
+    return ''.join(chars[start:end+1])
+
+def remove_punctuation_and_length(text):
+    # 全角符号和半角符号的Unicode范围
+    full_width_punctuations = '！＂＃＄％＆＇（）＊＋，－。／：；＜＝＞？＠［＼］＾＿｀｛｜｝～'
+    half_width_punctuations = '!"#$%&\'()*+,-./:;<=>?@[\]^_`{|}~'
+    space = ' '  # 半角空格
+    full_width_space = '　'  # 全角空格
+
+    # 去除全角和半角符号以及空格
+    result = ''.join([char for char in text if
+                      char not in full_width_punctuations and char not in half_width_punctuations and char not in space and char not in full_width_space])
+
+    if result == "Yeah":
+        return 0
+    return len(result)

core/utils/vad.py

@@ -0,0 +1,77 @@
+from abc import ABC, abstractmethod
+import logging
+import opuslib
+import time
+import numpy as np
+import torch
+
+logger = logging.getLogger(__name__)
+
+
+class VAD(ABC):
+    @abstractmethod
+    def is_vad(self, conn, data):
+        """检测音频数据中的语音活动"""
+        pass
+
+
+class SileroVAD(VAD):
+    def __init__(self, config):
+        logger.info("SileroVAD", config)
+        self.model, self.utils = torch.hub.load(repo_or_dir=config["model_dir"],
+                                                source='local',
+                                                model='silero_vad',
+                                                force_reload=False)
+        (get_speech_timestamps, _, _, _, _) = self.utils
+
+        self.decoder = opuslib.Decoder(16000, 1)
+        self.vad_threshold = config.get("threshold")
+        self.silence_threshold_ms = config.get("min_silence_duration_ms")
+
+    def is_vad(self, conn, opus_packet):
+        try:
+            pcm_frame = self.decoder.decode(opus_packet, 960)
+            conn.client_audio_buffer += pcm_frame  # 将新数据加入缓冲区
+
+            # 处理缓冲区中的完整帧（每次处理512采样点）
+            client_have_voice = False
+            while len(conn.client_audio_buffer) >= 512 * 2:
+                # 提取前512个采样点（1024字节）
+                chunk = conn.client_audio_buffer[:512 * 2]
+                conn.client_audio_buffer = conn.client_audio_buffer[512 * 2:]
+
+                # 转换为模型需要的张量格式
+                audio_int16 = np.frombuffer(chunk, dtype=np.int16)
+                audio_float32 = audio_int16.astype(np.float32) / 32768.0
+                audio_tensor = torch.from_numpy(audio_float32)
+
+                # 检测语音活动
+                speech_prob = self.model(audio_tensor, 16000).item()
+                client_have_voice = speech_prob >= self.vad_threshold
+
+                # 如果之前有声音，但本次没有声音，且与上次有声音的时间查已经超过了静默阈值，则认为已经说完一句话
+                if conn.client_have_voice and not client_have_voice:
+                    stop_duration = time.time() * 1000 - conn.client_have_voice_last_time
+                    if stop_duration >= self.silence_threshold_ms:
+                        conn.client_voice_stop = True
+                if client_have_voice:
+                    conn.client_have_voice = True
+                    conn.client_have_voice_last_time = time.time() * 1000
+
+            return client_have_voice
+        except opuslib.OpusError as e:
+            logger.info(f"解码错误: {e}")
+        except Exception as e:
+            logger.error(f"Error processing audio packet: {e}")
+
+
+def create_instance(class_name, *args, **kwargs) -> VAD:
+    # 获取类对象
+    cls_map = {
+        "SileroVAD": SileroVAD,
+        # 可扩展其他SileroVAD实现
+    }
+
+    if cls := cls_map.get(class_name):
+        return cls(*args, **kwargs)
+    raise ValueError(f"不支持的SileroVAD类型: {class_name}")