#!/usr/bin/env python3 """ STT 组件 - 语音转文字 支持 FunASR HTTP API、本地模型两种方式 """ import asyncio import logging import os import io import wave import numpy as np from typing import Optional, Dict, Any from event_bus import EventBus, EventType from components.base import BaseComponent, ComponentConfig, STTComponent, ComponentState logger = logging.getLogger(__name__) class FunASRSTT(STTComponent): """FunASR STT 组件 - 接入真实 HTTP API""" def __init__(self, config: Dict, event_bus: EventBus): super().__init__( config=ComponentConfig( name="stt_funasr", enabled=config.get("stt", {}).get("enabled", True), priority=1, dependencies=[] ), event_bus=event_bus ) self.stt_config = config.get("stt", {}) self.endpoint = self.stt_config.get("endpoint", "http://localhost:8080") self.model = self.stt_config.get("model", "paraformer-zh") self._client = None self._is_ready = False async def _do_initialize(self): """初始化 FunASR 客户端""" logger.info(f"🎤 初始化 FunASR STT: {self.model} @ {self.endpoint}") try: import httpx self._client = httpx.AsyncClient( base_url=self.endpoint.rstrip('/'), timeout=15.0 ) # 预热检查 await self._warmup() self._is_ready = True logger.info("✅ STT 就绪") except ImportError: logger.error("❌ httpx 未安装，请运行: pip install httpx") self._is_ready = False except Exception as e: logger.error(f"❌ STT 初始化失败: {e}") self._is_ready = False async def _warmup(self): """预热请求 - 触发模型加载""" try: # 发送短静音做预热 buf = io.BytesIO() with wave.open(buf, "w") as wf: wf.setnchannels(1) wf.setsampwidth(2) wf.setframerate(16000) wf.writeframes(b"\x00\x00" * 160) # 10ms 静音 resp = await self._client.post( "/stream", files={"file": ("warmup.wav", buf.getvalue(), "audio/wav")}, timeout=30.0 ) if resp.status_code == 200: logger.info("✅ STT 预热完成") else: logger.warning(f"STT 预热响应异常: {resp.status_code}") except Exception as e: logger.warning(f"STT 预热失败（非致命）: {e}") async def _run_loop(self): """STT 主循环 - 监听音频事件""" logger.info("🔄 STT 主循环启动") while self.state == ComponentState.RUNNING: try: await asyncio.sleep(1) except asyncio.CancelledError: break except Exception as e: logger.error(f"❌ STT 主循环错误: {e}") await asyncio.sleep(1) async def transcribe(self, audio_data: bytes) -> str: """ 转录音频 Args: audio_data: 音频数据 (bytes) Returns: str: 转录文本 """ if not self._is_ready: logger.warning("⚠️ STT 未就绪") return "" if len(audio_data) < 3200: return "" try: # 构造 WAV 文件 buf = io.BytesIO() with wave.open(buf, "w") as wf: wf.setnchannels(1) wf.setsampwidth(2) wf.setframerate(16000) wf.writeframes(audio_data) # 调用 FunASR HTTP API resp = await self._client.post( "/stream", files={"file": ("audio.wav", buf.getvalue(), "audio/wav")}, timeout=30.0 ) if resp.status_code == 200: result = resp.json() text = result.get("text", "").strip() logger.info(f"🎤 转录: {text[:50]}..." if text else "🎤 转录: (静音)") return text logger.warning(f"STT API 错误: {resp.status_code} {resp.text}") return "" except Exception as e: logger.error(f"❌ STT 转录失败: {e}") self.event_bus.publish(EventType.SYSTEM_ERROR, { "component": "stt", "error": str(e) }, source="stt_funasr") return "" def is_ready(self) -> bool: return self._is_ready def health_check(self) -> Dict[str, Any]: return { "status": "healthy" if self._is_ready else "unhealthy", "model": self.model, "endpoint": self.endpoint, "ready": self._is_ready } class LocalFunASRSTT(STTComponent): """本地 FunASR STT - 直接调用模型（不依赖 HTTP）""" def __init__(self, config: Dict, event_bus: EventBus): super().__init__( config=ComponentConfig( name="stt_local_funasr", enabled=config.get("stt", {}).get("local", False), priority=1, dependencies=[] ), event_bus=event_bus ) self.stt_config = config.get("stt", {}) self.model_name = self.stt_config.get("model", "SenseVoiceSmall") self._model = None self._is_ready = False async def _do_initialize(self): """初始化本地 FunASR 模型""" logger.info(f"🎤 初始化本地 FunASR: {self.model_name}") try: import sys # 重定向 stdout/stderr 避免 FunASR 噪音 _devnull = open(os.devnull, 'w') old_stdout, old_stderr = sys.stdout, sys.stderr sys.stdout, sys.stderr = _devnull, _devnull from funasr import AutoModel self._model = AutoModel( model=self.model_name, device="cpu", ncpu=4, batch_size_s=2 # 流式场景用小 batch ) sys.stdout, sys.stderr = old_stdout, old_stderr _devnull.close() self._is_ready = True logger.info("✅ 本地 FunASR STT 就绪") except ImportError as e: logger.error(f"❌ 缺少依赖: {e}") logger.error("请运行: pip install funasr torch") self._is_ready = False except Exception as e: sys.stdout, sys.stderr = old_stdout, old_stderr logger.error(f"❌ 本地 FunASR 初始化失败: {e}") self._is_ready = False async def _run_loop(self): pass async def transcribe(self, audio_data: bytes) -> str: """转录本地音频""" if not self._is_ready or self._model is None: return "" if len(audio_data) < 3200: return "" try: # 保存到临时文件 import tempfile with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f: temp_path = f.name with wave.open(f, "w") as wf: wf.setnchannels(1) wf.setsampwidth(2) wf.setframerate(16000) wf.writeframes(audio_data) # 转录 res = self._model.generate(input=temp_path) text = res[0].get("text", "").strip() if res else "" # 清理 os.unlink(temp_path) logger.info(f"🎤 本地转录: {text[:50]}..." if text else "🎤 本地转录: (静音)") return text except Exception as e: logger.error(f"❌ 本地转录失败: {e}") return "" def is_ready(self) -> bool: return self._is_ready def health_check(self) -> Dict[str, Any]: return { "status": "healthy" if self._is_ready else "unhealthy", "model": self.model_name, "mode": "local" } # ========== 工厂函数 ========== def create_stt_component(config: Dict, event_bus: EventBus) -> STTComponent: """创建 STT 组件""" mode = config.get("stt", {}).get("mode", "http") if mode == "local": return LocalFunASRSTT(config, event_bus) else: return FunASRSTT(config, event_bus)