chore: 移除 Rust 客户端初始化脚本中的 ASCII 艺术和版本信息

简化初始化脚本，删除启动时显示的 ASCII 艺术标题和版本信息，使输出更简洁。

.gitignore

@@ -47,3 +47,4 @@ __pycache__
 .mi.json
 xiaozhi.json
 open-xiaoai.node
+examples/hass/config.json

examples/hass/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "open_xiaoai_server"
+version = "1.0.0"
+edition = "2021"
+
+[lib]
+name = "open_xiaoai_server"
+crate-type = ["cdylib"]
+
+[dependencies]
+pyo3 = { version = "0.24.0", features = ["extension-module", "abi3-py38"] }
+pyo3-async-runtimes = { version = "0.24", features = ["attributes", "tokio-runtime"] }
+open-xiaoai = { path = "../../packages/client-rust" }
+tokio = { version = "1.32.0", features = ["full"] }
+tokio-tungstenite = "0.26.2"
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
+

examples/hass/Dockerfile

@@ -0,0 +1,53 @@
+FROM python:3.12-slim AS builder
+
+SHELL ["/bin/bash", "-o", "pipefail", "-c"]
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    curl build-essential pkg-config libssl-dev \
+    && rm -rf /var/lib/apt/lists/*
+
+RUN curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y
+
+ENV BASH_ENV=/root/.bash_env
+RUN touch "$BASH_ENV"
+RUN echo '. "$BASH_ENV"' >> "$HOME/.bashrc"
+RUN echo '[ -s "$HOME/.cargo/env" ] && . "$HOME/.cargo/env"' >> "$BASH_ENV"
+
+WORKDIR /app
+
+COPY examples/hass/hass ./hass
+COPY examples/hass/src ./src
+COPY examples/hass/Cargo.toml ./Cargo.toml
+COPY examples/hass/pyproject.toml ./pyproject.toml
+COPY examples/hass/README.md ./README.md
+COPY examples/hass/requirements.txt ./requirements.txt
+COPY examples/hass/hass_assistant.py ./hass_assistant.py
+COPY examples/hass/config.example.json ./config.example.json
+COPY packages/client-rust ./client-rust
+
+RUN python -m venv /opt/venv
+ENV VIRTUAL_ENV=/opt/venv
+ENV PATH="/opt/venv/bin:${PATH}"
+
+RUN pip install --upgrade pip
+RUN pip install -r requirements.txt
+RUN pip install maturin
+
+RUN sed -i 's/\.\.\/\.\.\/packages\///g' Cargo.toml \
+    && maturin develop --release
+
+
+
+FROM python:3.12-slim
+
+ENV VIRTUAL_ENV=/opt/venv
+ENV PATH="/opt/venv/bin:${PATH}"
+
+WORKDIR /app
+
+COPY --from=builder /opt/venv /opt/venv
+COPY --from=builder /app/hass /app/hass
+COPY --from=builder /app/hass_assistant.py /app/hass_assistant.py
+COPY --from=builder /app/config.example.json /app/config.example.json
+
+CMD ["python", "hass_assistant.py"]

examples/hass/README.md

@@ -0,0 +1,125 @@
+# Open-XiaoAI x Home Assistant Assist
+
+该示例用于将小爱音箱接入 Home Assistant 的对话/语音助手能力，支持文本对话与上下文会话，并提供可选的语音输入管线对接能力。
+
+## 功能说明
+
+- 文本对话：对接 Home Assistant `conversation.process` API，支持自然语言控制设备/查询信息
+- 实时打断：接收到新用户消息时先中断当前播报，再处理新输入，避免并发播报
+- 连续对话会话管理：默认开启连续对话；当命中结束词或语音静默超时时自动结束并重建 `conversation_id`
+- 区域上下文注入：支持客户端 IP 到 Home Assistant 区域映射，命中后在每次请求前注入区域前置提示词
+- 小爱音箱 I/O：从小爱音箱事件中提取最终识别文本（ASR），将回复通过小爱本机 TTS 播放
+- 可选音频能力（扩展点）：支持接收小爱录音流 `record`，用于后续对接 Assist pipeline（STT→Intent→TTS）
+
+## 使用场景
+
+- 用中文/英文自然语言控制 Home Assistant：开灯、关灯、调亮度、问温度/天气、查询传感器状态
+- 把小爱音箱作为 Home Assistant 的“外置麦克风 + 扬声器”（局域网内运行）
+## 架构概览
+
+- 小爱音箱 →（WebSocket 4399）→ Rust 网关（`open_xiaoai_server`）→ Python 回调
+- 文本链路（默认，低延迟）：
+  - 设备事件 `instruction/NewLine` → 抽取 `SpeechRecognizer.RecognizeResult` 文本
+  - 调用 HA `/api/conversation/process`
+  - 把返回的 `response.speech.plain.speech` 交给小爱 TTS 播报
+
+## 目录结构
+
+本目录包含：
+
+- `hass_assistant.py`：主入口
+- `config.example.json`：配置模板
+- `requirements.txt`：Python 依赖
+- `src/`：Rust(PyO3) 扩展模块（小爱音箱 I/O 网关）
+- `hass/`：Python 实现（HA 客户端、对话编排、上下文记忆）
+- `tests/`：单元测试
+- `scripts/`：集成测试脚本
+
+## 安装与运行
+
+### 前置条件
+
+- 你的小爱音箱已刷机并运行 Rust 客户端补丁（否则服务端收不到事件/音频流）
+- Home Assistant 已启用 `conversation`（默认配置一般已启用）并生成 Long-Lived Access Token
+- 本机具备 Python（建议 3.10+）与 Rust 工具链（用于编译 PyO3 扩展）
+- 可选：如果你希望使用 webrtcvad（更好的语音端点检测），Windows 需要安装 MSVC Build Tools
+
+### 1) 准备配置
+
+复制配置模板：
+
+```bash
+cd examples/hass
+cp config.example.json config.json
+```
+
+编辑 `config.json`：
+
+- `homeassistant.url`：HA 地址（例如 `http://192.168.1.10:8123`）
+- `homeassistant.access_token`：HA Long-Lived Access Token
+- `homeassistant.assistant_entity_id`：对话助理 ID（对应 `agent_id`，默认 `home_assistant`）
+
+### 配置新增项说明
+
+`xiaoai` 段新增以下关键配置：
+
+- `interrupt_on_new_input`：是否在收到新消息时立即执行打断，默认 `true`
+- `interrupt_command`：打断命令，默认使用 `ubus call mibrain ai_service` 触发中断
+- `continuous_conversation_enabled`：是否开启连续会话，默认 `true`
+- `session_idle_timeout_seconds`：连续会话静默超时秒数，默认 `15`
+- `session_end_keywords`：结束词列表，默认包含 `再见` `拜拜` `滚` `退下`，按包含匹配
+- `client_ip_area_mapping`：客户端 IP 到区域名称映射，仅在命中时注入区域提示词
+
+区域提示词注入内容为：
+
+`当前用户对话所在区域：<区域名称>，如后续对话未明确指定区域，则默认为此区域`
+
+### 2) 安装依赖
+
+```bash
+pip install -r requirements.txt
+```
+
+### 3) 编译 Rust 扩展（open_xiaoai_server）
+
+推荐使用 maturin：
+
+```bash
+pip install maturin
+maturin develop --release
+```
+
+### 4) 运行
+
+```bash
+python hass_assistant.py
+```
+
+## API 调用示例
+
+### Home Assistant conversation.process（curl）
+
+```bash
+curl -X POST \
+  -H "Authorization: Bearer YOUR_LONG_LIVED_ACCESS_TOKEN" \
+  -H "Content-Type: application/json" \
+  http://HOME_ASSISTANT:8123/api/conversation/process \
+  -d '{"text":"打开客厅灯","language":"zh-cn","agent_id":"home_assistant"}'
+```
+
+### Python 调用（不连接小爱，仅验证 HA）
+
+可参考 `scripts/integration_text.py`。
+## 故障排除
+
+- 401/403：检查 `access_token` 是否正确、是否完整复制
+- 连接超时：检查 `homeassistant.url` 是否可达、是否被防火墙/反代阻断
+- 无语音输入：确认小爱已运行客户端补丁且能连到本机 4399 端口
+- 无语音输出：确认小爱系统 `ubus`/`mibrain` 可用（或改用 `play_url` 播放 URL）
+- 对话不连贯：确认 `state.json` 能写入（用于持久化 `conversation_id`），或删除它重置会话
+- 打断不生效：检查 `xiaoai.interrupt_command` 是否可在设备上执行，以及日志中是否出现 `xiaoai_interrupt_failed`
+- 区域提示词未注入：检查 `client_ip_area_mapping` 是否包含当前客户端 IP，并查看 `area_mapping_hit/area_mapping_miss` 日志
+## 部署说明
+
+- 建议仅在局域网内部署运行，避免将 4399 暴露到公网
+- 如果 HA 与运行本示例的机器不在同一网段，确保 HA 的 URL 在设备侧可访问（尤其是播放 URL 时）

examples/hass/config.example.json

@@ -0,0 +1,36 @@
+{
+  "homeassistant": {
+    "url": "http://homeassistant.local:8123",
+    "access_token": "YOUR_LONG_LIVED_ACCESS_TOKEN",
+    "assistant_entity_id": "home_assistant",
+    "language": "zh-cn",
+    "conversation_id": null,
+    "timeout_seconds": 2.5
+  },
+  "xiaoai": {
+    "tts_blocking": false,
+    "shell_timeout_ms": 600000,
+    "interrupt_on_new_input": true,
+    "interrupt_command": "ubus call mibrain ai_service '{\"nlp\":1,\"nlp_text\":\"\"}'",
+    "continuous_conversation_enabled": true,
+    "session_idle_timeout_seconds": 15,
+    "session_end_keywords": ["再见", "拜拜", "滚", "退下"],
+    "client_ip_area_mapping": {
+      "192.168.1.101": "客厅",
+      "192.168.1.102": "主卧"
+    }
+  },
+  "runtime": {
+    "concurrency": 4,
+    "log_level": "INFO",
+    "persist_state_path": "state.json"
+  },
+  "voice": {
+    "enabled": false,
+    "mode": "record_vad_pipeline",
+    "vad_aggressiveness": 2,
+    "min_silence_ms": 800,
+    "max_utterance_ms": 15000,
+    "sample_rate": 16000
+  }
+}

examples/hass/hass/__init__.py

@@ -0,0 +1,10 @@
+__all__ = [
+    "config_loader",
+    "errors",
+    "ha_client",
+    "memory",
+    "typing",
+    "vad",
+    "xiaoai_bridge",
+]
+

examples/hass/hass/config_loader.py

@@ -0,0 +1,10 @@
+import json
+from pathlib import Path
+
+from hass.typing import AppConfig
+
+
+def load_config(path: str) -> AppConfig:
+    data = json.loads(Path(path).read_text(encoding="utf-8"))
+    return AppConfig.model_validate(data)
+

examples/hass/hass/errors.py

@@ -0,0 +1,19 @@
+class HassAssistantError(Exception):
+    pass
+
+
+class HomeAssistantRequestError(HassAssistantError):
+    pass
+
+
+class HomeAssistantAuthError(HomeAssistantRequestError):
+    pass
+
+
+class HomeAssistantResponseError(HomeAssistantRequestError):
+    pass
+
+
+class XiaoAiBridgeError(HassAssistantError):
+    pass
+

examples/hass/hass/ha_client.py

@@ -0,0 +1,161 @@
+import asyncio
+from typing import Optional
+
+import aiohttp
+
+from hass.errors import (
+    HomeAssistantAuthError,
+    HomeAssistantRequestError,
+    HomeAssistantResponseError,
+)
+from hass.memory import ConversationMemory
+from hass.typing import AssistantReply, HomeAssistantConfig
+
+
+class HomeAssistantClient:
+    def __init__(
+        self, config: HomeAssistantConfig, memory: ConversationMemory, concurrency: int = 4
+    ) -> None:
+        self._config = config
+        self._memory = memory
+        self._session: Optional[aiohttp.ClientSession] = None
+        self._lock = asyncio.Lock()
+        self._semaphore = asyncio.Semaphore(max(1, int(concurrency)))
+
+    async def start(self) -> None:
+        if self._session is not None:
+            return
+        headers = {"Authorization": f"Bearer {self._config.access_token}"}
+        timeout = aiohttp.ClientTimeout(total=self._config.timeout_seconds)
+        self._session = aiohttp.ClientSession(headers=headers, timeout=timeout)
+
+    async def close(self) -> None:
+        if self._session is None:
+            return
+        await self._session.close()
+        self._session = None
+
+    async def reset_conversation(self) -> None:
+        self._memory.conversation_id = None
+        await self._memory.save()
+
+    async def process_text(self, text: str, prefix_prompt: Optional[str] = None) -> AssistantReply:
+        if self._session is None:
+            raise HomeAssistantRequestError("HomeAssistantClient is not started")
+
+        request_text = text
+        if prefix_prompt:
+            request_text = f"{prefix_prompt}\n用户请求：{text}"
+
+        payload = {"text": request_text}
+        if self._config.language:
+            payload["language"] = self._config.language
+        if self._config.agent_id:
+            payload["agent_id"] = self._config.agent_id
+
+        conversation_id = self._memory.conversation_id or self._config.conversation_id
+        if conversation_id:
+            payload["conversation_id"] = conversation_id
+
+        url = str(self._config.url).rstrip("/") + "/api/conversation/process"
+        async with self._semaphore:
+            async with (
+                self._lock
+                if (self._memory.conversation_id or self._config.conversation_id)
+                else _NullAsyncContext()
+            ):
+                data = None
+                reset_attempts = 2 if conversation_id else 1
+                for reset_index in range(reset_attempts):
+                    if reset_index == 1:
+                        payload.pop("conversation_id", None)
+                        if self._memory.conversation_id:
+                            self._memory.conversation_id = None
+                            await self._memory.save()
+                    attempts = 3
+                    delay = 0.5
+                    for i in range(attempts):
+                        try:
+                            timeout_seconds = (
+                                self._config.timeout_seconds
+                                if i == 0
+                                else max(self._config.timeout_seconds, 8.0)
+                            )
+                            request_timeout = aiohttp.ClientTimeout(total=timeout_seconds)
+                            async with self._session.post(
+                                url,
+                                json=payload,
+                                timeout=request_timeout,
+                            ) as resp:
+                                if resp.status in (401, 403):
+                                    raise HomeAssistantAuthError(
+                                        f"Home Assistant auth failed: {resp.status}"
+                                    )
+                                if resp.status >= 500:
+                                    raise HomeAssistantRequestError(
+                                        f"Home Assistant server error: {resp.status}"
+                                    )
+                                if resp.status >= 400:
+                                    raise HomeAssistantRequestError(
+                                        f"Home Assistant request failed: {resp.status}"
+                                    )
+                                data = await resp.json()
+                                break
+                        except asyncio.TimeoutError:
+                            if i < attempts - 1:
+                                await asyncio.sleep(delay)
+                                delay *= 2
+                                continue
+                            error = HomeAssistantRequestError(
+                                f"Home Assistant request timeout, timeout_seconds={max(self._config.timeout_seconds, 8.0)}"
+                            )
+                        except aiohttp.ClientError as e:
+                            if i < attempts - 1:
+                                await asyncio.sleep(delay)
+                                delay *= 2
+                                continue
+                            error = HomeAssistantRequestError(str(e))
+                        else:
+                            error = None
+                        if error is not None:
+                            if reset_index == reset_attempts - 1:
+                                raise error
+                            break
+                    if data is not None:
+                        break
+
+        if not isinstance(data, dict):
+            raise HomeAssistantResponseError("Unexpected response payload type")
+
+        new_conversation_id = data.get("conversation_id")
+        if new_conversation_id:
+            self._memory.conversation_id = new_conversation_id
+            await self._memory.save()
+
+        response = data.get("response", {}) if isinstance(data.get("response", {}), dict) else {}
+        speech = (
+            response.get("speech", {}).get("plain", {}).get("speech", "")
+            if isinstance(response.get("speech", {}), dict)
+            else ""
+        )
+        if not isinstance(speech, str):
+            raise HomeAssistantResponseError("Unexpected response speech type")
+        if not speech:
+            response_type = response.get("response_type")
+            speech = "好的" if response_type in ("action_done", "query_answer") else "出错了"
+
+        continue_conversation = bool(data.get("continue_conversation", False))
+        return AssistantReply(
+            speech=speech,
+            continue_conversation=continue_conversation,
+            conversation_id=new_conversation_id,
+            raw=data,
+        )
+
+
+class _NullAsyncContext:
+    async def __aenter__(self):
+        return None
+
+    async def __aexit__(self, exc_type, exc, tb):
+        return False

examples/hass/hass/memory.py

@@ -0,0 +1,22 @@
+import json
+from pathlib import Path
+from typing import Optional
+
+
+class ConversationMemory:
+    def __init__(self, state_path: str) -> None:
+        self._path = Path(state_path)
+        self.conversation_id: Optional[str] = None
+
+    async def load(self) -> None:
+        if not self._path.exists():
+            return
+        data = json.loads(self._path.read_text(encoding="utf-8"))
+        self.conversation_id = data.get("conversation_id")
+
+    async def save(self) -> None:
+        self._path.write_text(
+            json.dumps({"conversation_id": self.conversation_id}, ensure_ascii=False, indent=2),
+            encoding="utf-8",
+        )
+

examples/hass/hass/typing.py

@@ -0,0 +1,68 @@
+from __future__ import annotations
+
+from typing import Literal, Optional
+
+from pydantic import BaseModel, Field, HttpUrl, model_validator
+
+
+class HomeAssistantConfig(BaseModel):
+    url: HttpUrl
+    access_token: str
+    agent_id: str = "home_assistant"
+    assistant_entity_id: Optional[str] = None
+    language: Optional[str] = None
+    conversation_id: Optional[str] = None
+    timeout_seconds: float = 2.5
+
+    @model_validator(mode="after")
+    def _normalize_agent_id(self):
+        if self.assistant_entity_id:
+            self.agent_id = self.assistant_entity_id
+        return self
+
+
+class XiaoAiConfig(BaseModel):
+    tts_blocking: bool = False
+    shell_timeout_ms: int = 10 * 60 * 1000
+    interrupt_on_new_input: bool = True
+    interrupt_command: str = "ubus call mibrain ai_service '{\"nlp\":1,\"nlp_text\":\"\"}'"
+    continuous_conversation_enabled: bool = True
+    session_idle_timeout_seconds: int = 15
+    session_end_keywords: list[str] = Field(
+        default_factory=lambda: ["再见", "拜拜", "滚", "退下"]
+    )
+    client_ip_area_mapping: dict[str, str] = Field(default_factory=dict)
+
+
+class RuntimeConfig(BaseModel):
+    concurrency: int = 4
+    log_level: str = "INFO"
+    persist_state_path: str = "state.json"
+
+
+class VoiceConfig(BaseModel):
+    enabled: bool = False
+    mode: Literal["record_vad_pipeline"] = "record_vad_pipeline"
+    vad_aggressiveness: int = 2
+    min_silence_ms: int = 800
+    max_utterance_ms: int = 15000
+    sample_rate: int = 16000
+
+
+class AppConfig(BaseModel):
+    homeassistant: HomeAssistantConfig
+    xiaoai: XiaoAiConfig = Field(default_factory=XiaoAiConfig)
+    runtime: RuntimeConfig = Field(default_factory=RuntimeConfig)
+    voice: VoiceConfig = Field(default_factory=VoiceConfig)
+
+
+class AssistantReply(BaseModel):
+    speech: str
+    continue_conversation: bool = False
+    conversation_id: Optional[str] = None
+    raw: dict = Field(default_factory=dict)
+
+
+class RecognizedSpeechInput(BaseModel):
+    text: str
+    client_ip: Optional[str] = None

examples/hass/hass/vad.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+try:
+    import webrtcvad
+except Exception:
+    webrtcvad = None
+
+
+@dataclass(frozen=True)
+class VadConfig:
+    aggressiveness: int = 2
+    min_silence_ms: int = 800
+    max_utterance_ms: int = 15000
+    sample_rate: int = 16000
+    energy_threshold: int = 450
+
+
+class VadSegmenter:
+    def __init__(self, config: VadConfig) -> None:
+        self._config = config
+        self._vad = webrtcvad.Vad(int(config.aggressiveness)) if webrtcvad else None
+        self._frame_ms = 20
+        self._frame_bytes = int(config.sample_rate * self._frame_ms / 1000) * 2
+        self._buffer = bytearray()
+        self._in_speech = False
+        self._speech = bytearray()
+        self._silence_ms = 0
+        self._speech_ms = 0
+
+    def feed(self, pcm: bytes) -> list[bytes]:
+        if not pcm:
+            return []
+
+        self._buffer.extend(pcm)
+        utterances: list[bytes] = []
+
+        while len(self._buffer) >= self._frame_bytes:
+            frame = bytes(self._buffer[: self._frame_bytes])
+            del self._buffer[: self._frame_bytes]
+
+            if self._vad is not None:
+                is_speech = self._vad.is_speech(frame, self._config.sample_rate)
+            else:
+                samples = memoryview(frame).cast("h")
+                avg = sum(abs(x) for x in samples) / max(1, len(samples))
+                is_speech = avg >= self._config.energy_threshold
+            if is_speech:
+                if not self._in_speech:
+                    self._in_speech = True
+                    self._speech.clear()
+                    self._silence_ms = 0
+                    self._speech_ms = 0
+                self._speech.extend(frame)
+                self._speech_ms += self._frame_ms
+                self._silence_ms = 0
+            else:
+                if self._in_speech:
+                    self._speech.extend(frame)
+                    self._speech_ms += self._frame_ms
+                    self._silence_ms += self._frame_ms
+
+            if self._in_speech and (
+                self._silence_ms >= self._config.min_silence_ms
+                or self._speech_ms >= self._config.max_utterance_ms
+            ):
+                utterances.append(bytes(self._speech))
+                self._in_speech = False
+                self._speech.clear()
+                self._silence_ms = 0
+                self._speech_ms = 0
+
+        return utterances
+
+    def flush(self) -> list[bytes]:
+        if not self._in_speech or not self._speech:
+            return []
+        utterance = bytes(self._speech)
+        self._in_speech = False
+        self._speech.clear()
+        self._silence_ms = 0
+        self._speech_ms = 0
+        return [utterance]

examples/hass/hass/xiaoai_bridge.py

@@ -0,0 +1,243 @@
+import asyncio
+import json
+import logging
+import re
+from collections.abc import Awaitable, Callable
+from typing import Optional
+import time
+
+from hass.errors import XiaoAiBridgeError
+from hass.typing import RecognizedSpeechInput, XiaoAiConfig
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class XiaoAiBridge:
+    def __init__(self, config: XiaoAiConfig) -> None:
+        self._config = config
+        self._on_text: Optional[Callable[[RecognizedSpeechInput], Awaitable[None]]] = None
+        self._on_audio: Optional[Callable[[bytes], None]] = None
+        self._ready = asyncio.Event()
+        self._loop: Optional[asyncio.AbstractEventLoop] = None
+        self._last_text: Optional[str] = None
+        self._last_client_ip: Optional[str] = None
+        self._last_text_ts: float = 0.0
+
+    def set_on_text(self, fn: Callable[[RecognizedSpeechInput], Awaitable[None]]) -> None:
+        self._on_text = fn
+
+    def set_on_audio(self, fn: Callable[[bytes], None]) -> None:
+        self._on_audio = fn
+
+    async def start(self) -> None:
+        try:
+            import open_xiaoai_server
+        except Exception as e:
+            raise XiaoAiBridgeError(
+                "open_xiaoai_server module is not available (Rust extension not built)"
+            ) from e
+
+        self._open_xiaoai_server = open_xiaoai_server
+        self._loop = asyncio.get_running_loop()
+        self._register_callbacks()
+        asyncio.ensure_future(self._open_xiaoai_server.start_server())
+        self._ready.set()
+
+    def _register_callbacks(self) -> None:
+        def on_event(event_json: str) -> None:
+            # Rust 侧会把 Event 结构序列化成 JSON 字符串传入，这里做反序列化与容错解析
+            try:
+                event = json.loads(event_json)
+            except Exception:
+                return
+
+            if event.get("event") != "instruction":
+                return
+
+            data = event.get("data") or {}
+            line_text = data.get("NewLine")
+            if not line_text:
+                return
+
+            # NewLine 是设备端透传的 JSON 文本，包含语音识别结果等指令
+            try:
+                line = json.loads(line_text)
+            except Exception:
+                return
+
+            header = line.get("header") or {}
+            if header.get("namespace") != "SpeechRecognizer":
+                return
+            if header.get("name") != "RecognizeResult":
+                return
+
+            payload = line.get("payload") or {}
+            if not payload.get("is_final"):
+                return
+
+            results = payload.get("results") or []
+            if not results or not isinstance(results, list):
+                return
+
+            text = results[0].get("text") if isinstance(results[0], dict) else None
+            if not text or not isinstance(text, str):
+                return
+
+            text = text.strip()
+            if not text:
+                return
+
+            if self._on_text is None or self._loop is None:
+                return
+
+            recognized = RecognizedSpeechInput(
+                text=text,
+                client_ip=self._extract_client_ip(event=event, data=data, line=line, payload=payload),
+            )
+
+            now = time.monotonic()
+            if (
+                self._last_text == recognized.text
+                and self._last_client_ip == (recognized.client_ip or "")
+                and (now - self._last_text_ts) < 1.2
+            ):
+                _LOGGER.info(
+                    "xiaoai_duplicate_text_skipped text=%s client_ip=%s",
+                    recognized.text,
+                    recognized.client_ip or "",
+                )
+                return
+
+            self._last_text = recognized.text
+            self._last_client_ip = recognized.client_ip or ""
+            self._last_text_ts = now
+
+            async def _dispatch() -> None:
+                if self._config.interrupt_on_new_input:
+                    await self.interrupt_playback(reason="new_input")
+                await self._on_text(recognized)
+
+            if not recognized.client_ip:
+                _LOGGER.info("xiaoai_client_ip_missing")
+
+            # 回调来自 Rust 的 tokio 线程，需切回 Python 主事件循环执行异步对话逻辑
+            asyncio.run_coroutine_threadsafe(_dispatch(), self._loop)
+
+        def on_input_data(data: bytes) -> None:
+            if self._on_audio is None:
+                return
+            self._on_audio(data)
+
+        self._open_xiaoai_server.register_fn("on_event", on_event)
+        self._open_xiaoai_server.register_fn("on_input_data", on_input_data)
+
+    @staticmethod
+    def _extract_client_ip(
+        event: dict,
+        data: dict,
+        line: dict,
+        payload: dict,
+    ) -> Optional[str]:
+        def _normalize_ip(value: object) -> Optional[str]:
+            if not isinstance(value, str):
+                return None
+            text = value.strip()
+            if not text:
+                return None
+            if ":" in text:
+                host, _, maybe_port = text.rpartition(":")
+                if host and maybe_port.isdigit():
+                    text = host
+            match = re.search(r"\b(?:\d{1,3}\.){3}\d{1,3}\b", text)
+            if match:
+                return match.group(0)
+            return None
+
+        def _walk(obj: object) -> Optional[str]:
+            if isinstance(obj, dict):
+                for key, value in obj.items():
+                    key_text = str(key).lower()
+                    if any(token in key_text for token in ("ip", "addr", "remote")):
+                        normalized = _normalize_ip(value)
+                        if normalized:
+                            return normalized
+                    nested = _walk(value)
+                    if nested:
+                        return nested
+                return None
+            if isinstance(obj, list):
+                for item in obj:
+                    nested = _walk(item)
+                    if nested:
+                        return nested
+                return None
+            return _normalize_ip(obj)
+
+        candidates = [
+            data.get("ClientIP"),
+            data.get("client_ip"),
+            data.get("ip"),
+            data.get("remote_addr"),
+            line.get("client_ip"),
+            payload.get("client_ip"),
+            payload.get("remote_addr"),
+            event.get("client_ip"),
+            event.get("remote_addr"),
+            event.get("peer"),
+            event.get("addr"),
+            event,
+        ]
+        for candidate in candidates:
+            normalized = _walk(candidate)
+            if normalized:
+                return normalized
+        return None
+
+    @staticmethod
+    def _escape_single_quotes(text: str) -> str:
+        # run_shell 侧通常会交给 shell 执行，单引号中出现单引号需要转义
+        return text.replace("'", "'\\''")
+
+    async def interrupt_playback(self, reason: str) -> None:
+        await self._ready.wait()
+        command = self._config.interrupt_command.strip()
+        if not command:
+            _LOGGER.info("xiaoai_interrupt_skipped reason=%s detail=empty_command", reason)
+            return
+        _LOGGER.info("xiaoai_interrupt_triggered reason=%s", reason)
+        try:
+            await self._open_xiaoai_server.run_shell(
+                command, float(self._config.shell_timeout_ms)
+            )
+        except Exception as exc:
+            _LOGGER.warning("xiaoai_interrupt_failed reason=%s error=%s", reason, exc)
+
+    async def speak(self, text: str) -> None:
+        if not text:
+            return
+        await self._ready.wait()
+        if self._config.tts_blocking:
+            safe_text = self._escape_single_quotes(text)
+            script = f"/usr/sbin/tts_play.sh '{safe_text}'"
+            await self._open_xiaoai_server.run_shell(
+                script, float(self._config.shell_timeout_ms)
+            )
+            return
+
+        payload = json.dumps({"text": text, "save": 0}, ensure_ascii=False)
+        payload = self._escape_single_quotes(payload)
+        script = f"ubus call mibrain text_to_speech '{payload}'"
+        await self._open_xiaoai_server.run_shell(script, float(self._config.shell_timeout_ms))
+
+    async def play_url(self, url: str) -> None:
+        if not url:
+            return
+        await self._ready.wait()
+        payload = json.dumps({"url": url, "type": 1}, ensure_ascii=False)
+        payload = self._escape_single_quotes(payload)
+        script = f"ubus call mediaplayer player_play_url '{payload}'"
+        await self._open_xiaoai_server.run_shell(script, float(self._config.shell_timeout_ms))
+
+    async def wait_forever(self) -> None:
+        await self._ready.wait()
+        await asyncio.Event().wait()

examples/hass/hass_assistant.py

@@ -0,0 +1,164 @@
+import asyncio
+import logging
+import os
+from pathlib import Path
+from collections.abc import Awaitable, Callable
+from typing import Optional
+
+from hass.config_loader import load_config
+from hass.errors import HomeAssistantRequestError
+from hass.ha_client import HomeAssistantClient
+from hass.memory import ConversationMemory
+from hass.typing import RecognizedSpeechInput
+from hass.xiaoai_bridge import XiaoAiBridge
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class ConversationSession:
+    def __init__(
+        self,
+        *,
+        enabled: bool,
+        idle_timeout_seconds: int,
+        end_keywords: list[str],
+    ) -> None:
+        self._enabled = enabled
+        self._idle_timeout_seconds = max(1, int(idle_timeout_seconds))
+        self._end_keywords = [keyword.strip() for keyword in end_keywords if keyword and keyword.strip()]
+        self._idle_task: Optional[asyncio.Task] = None
+
+    def match_end_keyword(self, text: str) -> Optional[str]:
+        for keyword in self._end_keywords:
+            if keyword in text:
+                return keyword
+        return None
+
+    @staticmethod
+    def build_area_prefix(area: str) -> str:
+        return f"当前用户对话所在区域：{area}，如后续对话未明确指定区域，则默认为此区域"
+
+    def resolve_area(self, client_ip: Optional[str], mapping: dict[str, str]) -> Optional[str]:
+        if not client_ip:
+            return None
+        return mapping.get(client_ip)
+
+    def restart_idle_timer(self, on_timeout: Callable[[], Awaitable[None]]) -> None:
+        if not self._enabled:
+            return
+        self.cancel_idle_timer()
+
+        async def _timeout_worker() -> None:
+            try:
+                await asyncio.sleep(self._idle_timeout_seconds)
+                await on_timeout()
+            except asyncio.CancelledError:
+                return
+
+        self._idle_task = asyncio.create_task(_timeout_worker())
+
+    def cancel_idle_timer(self) -> None:
+        if self._idle_task and not self._idle_task.done():
+            self._idle_task.cancel()
+        self._idle_task = None
+
+    async def close(self) -> None:
+        self.cancel_idle_timer()
+
+
+async def run(config_path: str) -> None:
+    config = load_config(config_path)
+    logging.basicConfig(level=getattr(logging, config.runtime.log_level.upper(), logging.INFO))
+
+    memory = ConversationMemory(config.runtime.persist_state_path)
+    await memory.load()
+
+    ha = HomeAssistantClient(config.homeassistant, memory, concurrency=config.runtime.concurrency)
+    await ha.start()
+
+    bridge = XiaoAiBridge(config.xiaoai)
+    await bridge.start()
+
+    session = ConversationSession(
+        enabled=config.xiaoai.continuous_conversation_enabled,
+        idle_timeout_seconds=config.xiaoai.session_idle_timeout_seconds,
+        end_keywords=config.xiaoai.session_end_keywords,
+    )
+    session_end_lock = asyncio.Lock()
+
+    async def end_session(reason: str) -> None:
+        async with session_end_lock:
+            await ha.reset_conversation()
+            _LOGGER.info("session_ended reason=%s", reason)
+
+    async def on_idle_timeout() -> None:
+        await end_session("idle_timeout")
+
+    async def on_text(recognized: RecognizedSpeechInput) -> None:
+        text = recognized.text.strip()
+        if not text:
+            return
+
+        session.restart_idle_timer(on_idle_timeout)
+
+        matched_keyword = session.match_end_keyword(text)
+        if matched_keyword:
+            await end_session(f"keyword:{matched_keyword}")
+            await bridge.speak("好的，当前会话已结束")
+            return
+
+        area = session.resolve_area(recognized.client_ip, config.xiaoai.client_ip_area_mapping)
+        prefix_prompt = None
+        if area:
+            prefix_prompt = session.build_area_prefix(area)
+            _LOGGER.info(
+                "area_mapping_hit client_ip=%s area=%s",
+                recognized.client_ip,
+                area,
+            )
+        else:
+            _LOGGER.info("area_mapping_miss client_ip=%s", recognized.client_ip or "")
+
+        try:
+            if not config.xiaoai.continuous_conversation_enabled:
+                await end_session("continuous_disabled_new_round")
+            reply = await ha.process_text(text, prefix_prompt=prefix_prompt)
+            await bridge.speak(reply.speech)
+        except HomeAssistantRequestError as exc:
+            _LOGGER.warning(
+                "ha_process_failed text=%s client_ip=%s error=%s",
+                text,
+                recognized.client_ip or "",
+                exc,
+            )
+            if prefix_prompt:
+                try:
+                    _LOGGER.info("ha_process_retry_without_prefix client_ip=%s", recognized.client_ip)
+                    reply = await ha.process_text(text, prefix_prompt=None)
+                    await bridge.speak(reply.speech)
+                    return
+                except HomeAssistantRequestError as retry_exc:
+                    _LOGGER.warning(
+                        "ha_process_retry_failed client_ip=%s error=%s",
+                        recognized.client_ip or "",
+                        retry_exc,
+                    )
+            await bridge.speak("Home Assistant 连接异常")
+
+    bridge.set_on_text(on_text)
+    try:
+        await bridge.wait_forever()
+    finally:
+        await session.close()
+        await ha.close()
+
+
+def main() -> None:
+    config_path = os.environ.get("HASS_ASSISTANT_CONFIG", "config.json")
+    if not Path(config_path).exists():
+        config_path = "config.example.json"
+    asyncio.run(run(config_path))
+
+
+if __name__ == "__main__":
+    main()

examples/hass/pyproject.toml

@@ -0,0 +1,19 @@
+[project]
+name = "open-xiaoai-hass"
+version = "1.0.0"
+description = "小爱音箱接入 Home Assistant Assist"
+readme = "README.md"
+requires-python = ">=3.10"
+dependencies = [
+  "aiohttp>=3.9.0",
+  "websockets>=12.0",
+  "pydantic>=2.6.0",
+  "maturin>=1.8.3"
+]
+
+[build-system]
+requires = ["maturin>=1.0,<2.0"]
+build-backend = "maturin"
+
+[tool.maturin]
+features = ["pyo3/extension-module"]

examples/hass/requirements.txt

@@ -0,0 +1,8 @@
+aiohttp>=3.9.0
+websockets>=12.0
+webrtcvad>=2.0.10; platform_system != "Windows"
+pydantic>=2.6.0
+
+pytest>=8.0.0
+pytest-asyncio>=0.23.0
+aioresponses>=0.7.7

examples/hass/scripts/__init__.py

@@ -0,0 +1 @@
+

examples/hass/scripts/integration_device_speak.py

@@ -0,0 +1,21 @@
+import asyncio
+import sys
+
+from hass.config_loader import load_config
+from hass.xiaoai_bridge import XiaoAiBridge
+
+
+async def main() -> None:
+    config_path = sys.argv[1] if len(sys.argv) > 1 else "config.json"
+    text = sys.argv[2] if len(sys.argv) > 2 else "已连接"
+
+    config = load_config(config_path)
+    bridge = XiaoAiBridge(config.xiaoai)
+    await bridge.start()
+    await bridge.speak(text)
+    await asyncio.sleep(1)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+

examples/hass/scripts/integration_device_text.py

@@ -0,0 +1,10 @@
+import asyncio
+import sys
+
+from hass_assistant import run
+
+
+if __name__ == "__main__":
+    config_path = sys.argv[1] if len(sys.argv) > 1 else "config.json"
+    asyncio.run(run(config_path))
+

examples/hass/scripts/integration_text.py

@@ -0,0 +1,30 @@
+import asyncio
+import sys
+
+from hass.config_loader import load_config
+from hass.ha_client import HomeAssistantClient
+from hass.memory import ConversationMemory
+
+
+async def main() -> None:
+    config_path = sys.argv[1] if len(sys.argv) > 1 else "config.json"
+    text = sys.argv[2] if len(sys.argv) > 2 else "打开客厅灯"
+
+    config = load_config(config_path)
+    memory = ConversationMemory(config.runtime.persist_state_path)
+    await memory.load()
+
+    client = HomeAssistantClient(
+        config.homeassistant, memory, concurrency=config.runtime.concurrency
+    )
+    await client.start()
+    try:
+        reply = await client.process_text(text)
+        print(reply.speech)
+    finally:
+        await client.close()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+

examples/hass/src/lib.rs

@@ -0,0 +1,112 @@
+use open_xiaoai::services::connect::message::MessageManager;
+use open_xiaoai::services::connect::rpc::RPC;
+use open_xiaoai::services::audio::config::AudioConfig;
+use pyo3::prelude::*;
+use pyo3::types::PyBytes;
+use serde_json::json;
+use server::AppServer;
+
+pub mod macros;
+pub mod python;
+pub mod server;
+
+#[pyfunction]
+fn on_output_data(py: Python, data: Py<PyBytes>) -> PyResult<Bound<PyAny>> {
+    let bytes = data.as_bytes(py).to_vec();
+    pyo3_async_runtimes::tokio::future_into_py(py, async move {
+        let _ = MessageManager::instance()
+            .send_stream("play", bytes, None)
+            .await;
+        Ok(())
+    })
+}
+
+#[pyfunction]
+fn start_server(py: Python) -> PyResult<Bound<PyAny>> {
+    pyo3_async_runtimes::tokio::future_into_py(py, async {
+        AppServer::run().await;
+        Ok(())
+    })
+}
+
+#[pyfunction]
+fn run_shell(py: Python, script: String, timeout_millis: f64) -> PyResult<Bound<PyAny>> {
+    pyo3_async_runtimes::tokio::future_into_py(py, async move {
+        let res = RPC::instance()
+            .call_remote("run_shell", Some(json!(script)), Some(timeout_millis as u64))
+            .await;
+        let result = match res {
+            Err(e) => format!("run_shell error: {}", e),
+            Ok(res) => match res.data {
+                Some(data) => serde_json::to_string(&data).unwrap_or_default(),
+                None => String::new(),
+            },
+        };
+        Ok(result)
+    })
+}
+
+#[pyfunction]
+fn start_recording(py: Python, config_json: Option<String>) -> PyResult<Bound<PyAny>> {
+    pyo3_async_runtimes::tokio::future_into_py(py, async move {
+        let config = config_json
+            .and_then(|text| serde_json::from_str::<AudioConfig>(&text).ok());
+        let res = RPC::instance()
+            .call_remote("start_recording", config.map(|c| json!(c)), None)
+            .await;
+        Ok(match res {
+            Ok(_) => true,
+            Err(_) => false,
+        })
+    })
+}
+
+#[pyfunction]
+fn stop_recording(py: Python) -> PyResult<Bound<PyAny>> {
+    pyo3_async_runtimes::tokio::future_into_py(py, async move {
+        let res = RPC::instance().call_remote("stop_recording", None, None).await;
+        Ok(match res {
+            Ok(_) => true,
+            Err(_) => false,
+        })
+    })
+}
+
+#[pyfunction]
+fn start_play(py: Python, config_json: Option<String>) -> PyResult<Bound<PyAny>> {
+    pyo3_async_runtimes::tokio::future_into_py(py, async move {
+        let config = config_json
+            .and_then(|text| serde_json::from_str::<AudioConfig>(&text).ok());
+        let res = RPC::instance()
+            .call_remote("start_play", config.map(|c| json!(c)), None)
+            .await;
+        Ok(match res {
+            Ok(_) => true,
+            Err(_) => false,
+        })
+    })
+}
+
+#[pyfunction]
+fn stop_play(py: Python) -> PyResult<Bound<PyAny>> {
+    pyo3_async_runtimes::tokio::future_into_py(py, async move {
+        let res = RPC::instance().call_remote("stop_play", None, None).await;
+        Ok(match res {
+            Ok(_) => true,
+            Err(_) => false,
+        })
+    })
+}
+
+#[pymodule]
+fn open_xiaoai_server(_py: Python, m: Bound<'_, PyModule>) -> PyResult<()> {
+    m.add_function(wrap_pyfunction!(start_server, &m)?)?;
+    m.add_function(wrap_pyfunction!(on_output_data, &m)?)?;
+    m.add_function(wrap_pyfunction!(run_shell, &m)?)?;
+    m.add_function(wrap_pyfunction!(start_recording, &m)?)?;
+    m.add_function(wrap_pyfunction!(stop_recording, &m)?)?;
+    m.add_function(wrap_pyfunction!(start_play, &m)?)?;
+    m.add_function(wrap_pyfunction!(stop_play, &m)?)?;
+    crate::python::init_module(&m)?;
+    Ok(())
+}

examples/hass/src/macros.rs

@@ -0,0 +1,7 @@
+#[macro_export]
+macro_rules! pylog {
+    ($($arg:tt)*) => {{
+        println!($($arg)*);
+    }};
+}
+

examples/hass/src/python.rs

@@ -0,0 +1,64 @@
+use pyo3::prelude::*;
+use std::collections::HashMap;
+use std::sync::{Arc, LazyLock, RwLock};
+
+pub struct PythonManager {
+    functions: Arc<RwLock<HashMap<String, PyObject>>>,
+}
+
+static INSTANCE: LazyLock<PythonManager> = LazyLock::new(PythonManager::new);
+
+impl PythonManager {
+    pub fn new() -> Self {
+        Self {
+            functions: Arc::new(RwLock::new(HashMap::new())),
+        }
+    }
+
+    pub fn instance() -> &'static Self {
+        &INSTANCE
+    }
+
+    fn register_fn(&self, key: &str, function: PyObject) -> PyResult<()> {
+        let mut functions = self.functions.write().unwrap();
+        functions.insert(key.to_string(), function);
+        Ok(())
+    }
+
+    fn unregister_fn(&self, key: &str) -> PyResult<()> {
+        let mut functions = self.functions.write().unwrap();
+        functions.remove(key);
+        Ok(())
+    }
+
+    pub fn call_fn(&self, key: &str, arg: Option<PyObject>) -> PyResult<PyObject> {
+        let functions = self.functions.read().unwrap();
+        if let Some(function) = functions.get(key) {
+            Python::with_gil(|py| match arg {
+                None => function.call0(py),
+                Some(arg) => function.call1(py, (arg,)),
+            })
+        } else {
+            Err(pyo3::exceptions::PyKeyError::new_err(format!(
+                "未找到函数: {}",
+                key
+            )))
+        }
+    }
+}
+
+#[pyfunction]
+fn register_fn(key: &str, function: PyObject) -> PyResult<()> {
+    PythonManager::instance().register_fn(key, function)
+}
+
+#[pyfunction]
+fn unregister_fn(key: &str) -> PyResult<()> {
+    PythonManager::instance().unregister_fn(key)
+}
+
+pub fn init_module(m: &Bound<'_, PyModule>) -> PyResult<()> {
+    m.add_function(wrap_pyfunction!(register_fn, m)?)?;
+    m.add_function(wrap_pyfunction!(unregister_fn, m)?)?;
+    Ok(())
+}

examples/hass/src/server.rs

@@ -0,0 +1,104 @@
+use crate::python::PythonManager;
+use open_xiaoai::base::{AppError, VERSION};
+use open_xiaoai::services::connect::data::{Event, Request, Response, Stream};
+use open_xiaoai::services::connect::handler::MessageHandler;
+use open_xiaoai::services::connect::message::{MessageManager, WsStream};
+use open_xiaoai::services::connect::rpc::RPC;
+use pyo3::types::{PyBytes, PyString};
+use pyo3::Python;
+use serde_json::json;
+use std::sync::{LazyLock, RwLock};
+use tokio::net::{TcpListener, TcpStream};
+use tokio_tungstenite::accept_async;
+
+static REMOTE_ADDR: LazyLock<RwLock<Option<String>>> = LazyLock::new(|| RwLock::new(None));
+
+pub struct AppServer;
+
+impl AppServer {
+    pub async fn connect(stream: TcpStream) -> Result<WsStream, AppError> {
+        let ws_stream = accept_async(stream).await?;
+        Ok(WsStream::Server(ws_stream))
+    }
+
+    pub async fn run() {
+        let addr = "0.0.0.0:4399";
+        let listener = TcpListener::bind(&addr)
+            .await
+            .expect(format!("❌ 绑定地址失败: {}", &addr).as_str());
+        crate::pylog!("✅ 已启动: {:?}", addr);
+        while let Ok((stream, addr)) = listener.accept().await {
+            AppServer::handle_connection(stream, addr).await;
+        }
+    }
+
+    async fn handle_connection(stream: TcpStream, addr: std::net::SocketAddr) {
+        let Ok(ws_stream) = AppServer::connect(stream).await else {
+            crate::pylog!("❌ 连接异常: {}", addr);
+            return;
+        };
+        crate::pylog!("✅ 已连接: {:?}", addr);
+        {
+            let mut remote = REMOTE_ADDR.write().unwrap();
+            *remote = Some(addr.ip().to_string());
+        }
+        AppServer::init(ws_stream).await;
+        if let Err(e) = MessageManager::instance().process_messages().await {
+            crate::pylog!("❌ 消息处理异常: {}", e);
+        }
+        AppServer::dispose().await;
+        {
+            let mut remote = REMOTE_ADDR.write().unwrap();
+            *remote = None;
+        }
+        crate::pylog!("❌ 已断开连接");
+    }
+
+    async fn init(ws_stream: WsStream) {
+        MessageManager::instance().init(ws_stream).await;
+        MessageHandler::<Event>::instance()
+            .set_handler(on_event)
+            .await;
+        MessageHandler::<Stream>::instance()
+            .set_handler(on_stream)
+            .await;
+
+        let rpc = RPC::instance();
+        rpc.add_command("get_version", get_version).await;
+    }
+
+    async fn dispose() {
+        MessageManager::instance().dispose().await;
+    }
+}
+
+async fn get_version(_: Request) -> Result<Response, AppError> {
+    let data = json!(VERSION.to_string());
+    Ok(Response::from_data(data))
+}
+
+async fn on_stream(stream: Stream) -> Result<(), AppError> {
+    let Stream { tag, bytes, .. } = stream;
+    match tag.as_str() {
+        "record" => {
+            let data = Python::with_gil(|py| PyBytes::new(py, &bytes).into());
+            PythonManager::instance().call_fn("on_input_data", Some(data))?;
+        }
+        _ => {}
+    }
+    Ok(())
+}
+
+async fn on_event(event: Event) -> Result<(), AppError> {
+    let mut event_json_value = serde_json::to_value(&event)?;
+    if let Some(event_obj) = event_json_value.as_object_mut() {
+        let remote = REMOTE_ADDR.read().unwrap().clone();
+        if let Some(ip) = remote {
+            event_obj.insert("remote_addr".to_string(), serde_json::Value::String(ip));
+        }
+    }
+    let event_json = serde_json::to_string(&event_json_value)?;
+    let data = Python::with_gil(|py| PyString::new(py, &event_json).into());
+    PythonManager::instance().call_fn("on_event", Some(data))?;
+    Ok(())
+}

examples/hass/tests/__init__.py

@@ -0,0 +1 @@
+

examples/hass/tests/conftest.py

@@ -0,0 +1,5 @@
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+

examples/hass/tests/test_assistant_session.py

@@ -0,0 +1,53 @@
+import asyncio
+
+import pytest
+
+from hass_assistant import ConversationSession
+
+
+@pytest.mark.asyncio
+async def test_session_idle_timeout_triggers_end():
+    session = ConversationSession(
+        enabled=True,
+        idle_timeout_seconds=1,
+        end_keywords=["再见", "拜拜", "滚", "退下"],
+    )
+
+    triggered = {"value": False}
+
+    async def on_timeout():
+        triggered["value"] = True
+
+    session.restart_idle_timer(on_timeout)
+    await asyncio.sleep(1.2)
+
+    assert triggered["value"] is True
+    await session.close()
+
+
+def test_session_end_keyword_contains_match():
+    session = ConversationSession(
+        enabled=True,
+        idle_timeout_seconds=15,
+        end_keywords=["再见", "拜拜", "滚", "退下"],
+    )
+
+    assert session.match_end_keyword("那就先这样吧，再见哈") == "再见"
+    assert session.match_end_keyword("你先退下吧") == "退下"
+    assert session.match_end_keyword("继续聊聊") is None
+
+
+def test_session_area_mapping_hit_and_miss():
+    session = ConversationSession(
+        enabled=True,
+        idle_timeout_seconds=15,
+        end_keywords=["再见"],
+    )
+
+    mapping = {"192.168.1.101": "客厅"}
+    assert session.resolve_area("192.168.1.101", mapping) == "客厅"
+    assert session.resolve_area("192.168.1.102", mapping) is None
+    assert session.resolve_area(None, mapping) is None
+
+    prefix = session.build_area_prefix("客厅")
+    assert "当前用户对话所在区域：客厅" in prefix

examples/hass/tests/test_ha_client.py

@@ -0,0 +1,123 @@
+import json
+
+import pytest
+from aioresponses import aioresponses
+
+from hass.errors import HomeAssistantAuthError
+from hass.ha_client import HomeAssistantClient
+from hass.memory import ConversationMemory
+from hass.typing import HomeAssistantConfig
+
+
+@pytest.mark.asyncio
+async def test_process_text_updates_conversation_id(tmp_path):
+    state_path = tmp_path / "state.json"
+    memory = ConversationMemory(str(state_path))
+    await memory.load()
+
+    config = HomeAssistantConfig(
+        url="http://ha.local:8123",
+        access_token="token",
+        assistant_entity_id="home_assistant",
+        language="zh-cn",
+        timeout_seconds=2.5,
+    )
+    client = HomeAssistantClient(config, memory, concurrency=2)
+    await client.start()
+
+    url = "http://ha.local:8123/api/conversation/process"
+    with aioresponses() as mocked:
+        mocked.post(
+            url,
+            payload={
+                "continue_conversation": True,
+                "conversation_id": "cid-1",
+                "response": {
+                    "response_type": "action_done",
+                    "speech": {"plain": {"speech": "好的"}},
+                },
+            },
+            status=200,
+        )
+        reply = await client.process_text("打开客厅灯")
+
+    assert reply.speech == "好的"
+    assert reply.continue_conversation is True
+    assert memory.conversation_id == "cid-1"
+    await client.close()
+
+
+@pytest.mark.asyncio
+async def test_process_text_unauthorized(tmp_path):
+    memory = ConversationMemory(str(tmp_path / "state.json"))
+    await memory.load()
+
+    config = HomeAssistantConfig(url="http://ha.local:8123", access_token="bad-token")
+    client = HomeAssistantClient(config, memory)
+    await client.start()
+
+    url = "http://ha.local:8123/api/conversation/process"
+    with aioresponses() as mocked:
+        mocked.post(url, payload={"message": "unauthorized"}, status=401)
+        with pytest.raises(HomeAssistantAuthError):
+            await client.process_text("hi")
+
+    await client.close()
+
+
+@pytest.mark.asyncio
+async def test_process_text_with_prefix_prompt(tmp_path):
+    memory = ConversationMemory(str(tmp_path / "state.json"))
+    await memory.load()
+
+    config = HomeAssistantConfig(url="http://ha.local:8123", access_token="token")
+    client = HomeAssistantClient(config, memory)
+    await client.start()
+
+    url = "http://ha.local:8123/api/conversation/process"
+    with aioresponses() as mocked:
+        mocked.post(
+            url,
+            payload={
+                "continue_conversation": False,
+                "conversation_id": "cid-2",
+                "response": {
+                    "response_type": "query_answer",
+                    "speech": {"plain": {"speech": "在客厅"}},
+                },
+            },
+            status=200,
+        )
+        await client.process_text(
+            "灯开了吗",
+            prefix_prompt="当前用户对话所在区域：客厅，如后续对话未明确指定区域，则默认为此区域",
+        )
+
+        request = mocked.requests[("POST", url)][0]
+        payload = json.loads(request.kwargs["data"].decode("utf-8"))
+        assert payload["text"].startswith("当前用户对话所在区域：客厅")
+        assert "用户请求：灯开了吗" in payload["text"]
+
+    await client.close()
+
+
+@pytest.mark.asyncio
+async def test_reset_conversation(tmp_path):
+    state_path = tmp_path / "state.json"
+    memory = ConversationMemory(str(state_path))
+    await memory.load()
+    memory.conversation_id = "cid-old"
+    await memory.save()
+
+    config = HomeAssistantConfig(url="http://ha.local:8123", access_token="token")
+    client = HomeAssistantClient(config, memory)
+    await client.start()
+
+    await client.reset_conversation()
+
+    assert memory.conversation_id is None
+    saved = json.loads(state_path.read_text(encoding="utf-8"))
+    assert saved["conversation_id"] is None
+
+    await client.close()
+

examples/hass/tests/test_memory.py

@@ -0,0 +1,21 @@
+import json
+
+import pytest
+
+from hass.memory import ConversationMemory
+
+
+@pytest.mark.asyncio
+async def test_memory_persist_roundtrip(tmp_path):
+    path = tmp_path / "state.json"
+    m = ConversationMemory(str(path))
+    m.conversation_id = "abc"
+    await m.save()
+
+    raw = json.loads(path.read_text(encoding="utf-8"))
+    assert raw["conversation_id"] == "abc"
+
+    m2 = ConversationMemory(str(path))
+    await m2.load()
+    assert m2.conversation_id == "abc"
+

examples/hass/tests/test_typing.py

@@ -0,0 +1,38 @@
+from hass.typing import AppConfig
+
+
+def test_xiaoai_config_defaults():
+    cfg = AppConfig.model_validate(
+        {
+            "homeassistant": {
+                "url": "http://ha.local:8123",
+                "access_token": "token",
+            }
+        }
+    )
+
+    assert cfg.xiaoai.continuous_conversation_enabled is True
+    assert cfg.xiaoai.session_idle_timeout_seconds == 15
+    assert cfg.xiaoai.session_end_keywords == ["再见", "拜拜", "滚", "退下"]
+    assert cfg.xiaoai.client_ip_area_mapping == {}
+    assert cfg.xiaoai.interrupt_on_new_input is True
+
+
+def test_xiaoai_config_area_mapping_loaded():
+    cfg = AppConfig.model_validate(
+        {
+            "homeassistant": {
+                "url": "http://ha.local:8123",
+                "access_token": "token",
+            },
+            "xiaoai": {
+                "client_ip_area_mapping": {
+                    "192.168.1.10": "客厅",
+                    "192.168.1.11": "书房",
+                }
+            },
+        }
+    )
+
+    assert cfg.xiaoai.client_ip_area_mapping["192.168.1.10"] == "客厅"
+    assert cfg.xiaoai.client_ip_area_mapping["192.168.1.11"] == "书房"

examples/hass/tests/test_vad.py

@@ -0,0 +1,41 @@
+import math
+import struct
+
+import pytest
+
+from hass.vad import VadConfig, VadSegmenter
+
+
+def _sine_frame(sample_rate: int, freq_hz: float, frame_ms: int = 20) -> bytes:
+    samples = int(sample_rate * frame_ms / 1000)
+    values = [
+        int(10000 * math.sin(2 * math.pi * freq_hz * i / sample_rate))
+        for i in range(samples)
+    ]
+    return struct.pack("<" + "h" * samples, *values)
+
+
+def _silence_frame(sample_rate: int, frame_ms: int = 20) -> bytes:
+    samples = int(sample_rate * frame_ms / 1000)
+    return b"\x00\x00" * samples
+
+
+@pytest.mark.parametrize("min_silence_ms", [40, 60])
+def test_vad_emits_utterance(min_silence_ms: int):
+    config = VadConfig(
+        aggressiveness=1,
+        min_silence_ms=min_silence_ms,
+        max_utterance_ms=2000,
+        sample_rate=16000,
+    )
+    seg = VadSegmenter(config)
+
+    utterances = []
+    for _ in range(6):
+        utterances.extend(seg.feed(_sine_frame(16000, 440.0)))
+    for _ in range(6):
+        utterances.extend(seg.feed(_silence_frame(16000)))
+
+    assert len(utterances) >= 1
+    assert len(utterances[0]) > 0
+

examples/hass/tests/test_xiaoai_bridge.py

@@ -0,0 +1,113 @@
+import asyncio
+import json
+
+import pytest
+
+from hass.typing import XiaoAiConfig
+from hass.xiaoai_bridge import XiaoAiBridge
+
+
+class FakeServer:
+    def __init__(self):
+        self._fns = {}
+        self.shell_calls = []
+
+    def register_fn(self, name, fn):
+        self._fns[name] = fn
+
+    async def run_shell(self, script, timeout):
+        self.shell_calls.append((script, timeout))
+        return '{"stdout":"ok","stderr":"","exit_code":0}'
+
+    async def start_server(self):
+        return None
+
+
+@pytest.mark.asyncio
+async def test_bridge_interrupts_before_dispatch(monkeypatch):
+    fake_server = FakeServer()
+
+    import hass.xiaoai_bridge as module
+
+    monkeypatch.setattr(module, "_LOGGER", module._LOGGER)
+
+    cfg = XiaoAiConfig(interrupt_on_new_input=True, shell_timeout_ms=123)
+    bridge = XiaoAiBridge(cfg)
+    bridge._open_xiaoai_server = fake_server
+    bridge._loop = asyncio.get_running_loop()
+
+    called = []
+
+    async def on_text(input_data):
+        called.append(input_data)
+
+    bridge.set_on_text(on_text)
+    bridge._ready.set()
+    bridge._register_callbacks()
+
+    event = {
+        "event": "instruction",
+        "data": {
+            "ClientIP": "192.168.1.101",
+            "NewLine": json.dumps(
+                {
+                    "header": {"namespace": "SpeechRecognizer", "name": "RecognizeResult"},
+                    "payload": {
+                        "is_final": True,
+                        "results": [{"text": "打开客厅灯"}],
+                    },
+                },
+                ensure_ascii=False,
+            ),
+        },
+    }
+
+    fake_server._fns["on_event"](json.dumps(event, ensure_ascii=False))
+    await asyncio.sleep(0.05)
+
+    assert len(fake_server.shell_calls) == 1
+    assert "ubus call mibrain ai_service" in fake_server.shell_calls[0][0]
+    assert len(called) == 1
+    assert called[0].text == "打开客厅灯"
+    assert called[0].client_ip == "192.168.1.101"
+
+
+@pytest.mark.asyncio
+async def test_bridge_extracts_missing_ip_as_none():
+    fake_server = FakeServer()
+    cfg = XiaoAiConfig(interrupt_on_new_input=False)
+    bridge = XiaoAiBridge(cfg)
+    bridge._open_xiaoai_server = fake_server
+    bridge._loop = asyncio.get_running_loop()
+
+    captured = []
+
+    async def on_text(input_data):
+        captured.append(input_data)
+
+    bridge.set_on_text(on_text)
+    bridge._ready.set()
+    bridge._register_callbacks()
+
+    event = {
+        "event": "instruction",
+        "data": {
+            "NewLine": json.dumps(
+                {
+                    "header": {"namespace": "SpeechRecognizer", "name": "RecognizeResult"},
+                    "payload": {
+                        "is_final": True,
+                        "results": [{"text": "查询天气"}],
+                    },
+                },
+                ensure_ascii=False,
+            ),
+        },
+    }
+
+    fake_server._fns["on_event"](json.dumps(event, ensure_ascii=False))
+    await asyncio.sleep(0.05)
+
+    assert len(captured) == 1
+    assert captured[0].client_ip is None
+    assert fake_server.shell_calls == []

examples/stereo/src/app/master.rs

@@ -91,7 +91,8 @@ pub async fn run_master(master_role: ChannelRole) -> Result<()> {
     let mut opus_out = vec![0u8; 1500];
     let mut seq = 0u32;
 
-    let delay_us = 200_000;
+    // 播放延迟: 只需覆盖网络延迟 + 时钟偏移
+    let delay_us = 100_000; // 100ms 基础延迟
     let frame_duration_us =
         (config.frame_size as f64 / config.sample_rate as f64 * 1_000_000.0) as u128;
 
@@ -142,7 +143,8 @@ pub async fn run_master(master_role: ChannelRole) -> Result<()> {
                     stream_start_seq = seq;
                 }
 
-                // 计算该帧应当播放的基准时间（相对于流开始）
+                // 计算该帧应当播放的目标时间
+                // target_ts = 数据包发送时间 + 播放延迟
                 let target_ts = stream_start_ts
                     + ((seq - stream_start_seq) as u128 * frame_duration_us)
                     + delay_us;
@@ -215,6 +217,8 @@ pub async fn run_master(master_role: ChannelRole) -> Result<()> {
                         let wait = target_ts - now;
                         if wait > 1000 {
                             tokio::time::sleep(Duration::from_micros(wait as u64)).await;
+                        } else {
+                            // 小于 1ms，直接播放，让播放时机稍微早一点点
                         }
                     }
                 } else {

examples/stereo/src/app/slave.rs

@@ -6,7 +6,6 @@ use crate::audio::player::AudioPlayer;
 use crate::net::discovery::Discovery;
 use crate::net::network::SlaveNetwork;
 use crate::net::protocol::{AudioPacket, ChannelRole, ControlPacket};
-use crate::utils::jitter_buffer::JitterBuffer;
 use crate::utils::sync::{ClockSync, now_us};
 use anyhow::{Result, anyhow};
 use std::sync::Arc;
@@ -62,7 +61,6 @@ async fn handle_connection(role: ChannelRole) -> Result<()> {
     };
     let player = AudioPlayer::new(&config)?;
     let mut codec = OpusCodec::new(&config)?;
-    let mut jitter = JitterBuffer::new(50_000, 3);
     let clock = Arc::new(Mutex::new(ClockSync::new(100)));
 
     // 用于通知主循环 TCP 已断开的消息通道
@@ -85,7 +83,7 @@ async fn handle_connection(role: ChannelRole) -> Result<()> {
                 let _ = d_tx_ping.send(()).await; // 通知主线程 TCP 失败
                 break;
             }
-            tokio::time::sleep(Duration::from_secs(1)).await;
+            tokio::time::sleep(Duration::from_millis(200)).await;
             seq += 1;
         }
     });
@@ -130,7 +128,7 @@ async fn handle_connection(role: ChannelRole) -> Result<()> {
         }
     });
 
-    // 8. 播放主循环
+    // 8. 播放提示
     println!("✅ 主节点已连接，音频串流中...");
     let role_str = role.to_string();
     tokio::spawn(async move {
@@ -144,9 +142,9 @@ async fn handle_connection(role: ChannelRole) -> Result<()> {
             .await;
     });
 
+    // 9. 播放主循环
     let mut pcm_buf = vec![0i16; config.frame_size];
     let mut last_seq: Option<u32> = None;
-    let mut last_packet_time = now_us();
 
     loop {
         // 检查 TCP 是否已断开
@@ -154,48 +152,48 @@ async fn handle_connection(role: ChannelRole) -> Result<()> {
             return Err(anyhow!("主节点已断开: {}", master_tcp_addr));
         }
 
-        // 填充 Jitter Buffer
-        while let Ok(pkt) = audio_rx.try_recv() {
-            let now = now_us();
-            // 如果超过 500ms 没有收到包，认为是新流开始，重置状态
-            if now - last_packet_time > 500_000 {
-                jitter.clear();
-                last_seq = None;
-                codec = OpusCodec::new(&config)?;
-                let _ = player.prepare();
-            }
-            last_packet_time = now;
-            jitter.push(pkt);
-        }
-
+        // 接收数据包
+        if let Ok(pkt) = audio_rx.try_recv() {
             let now = now_us();
             let current_server_time = clock.lock().await.to_server_time(now);
 
-        if let Some((seq, data)) = jitter.pop_frame(current_server_time) {
-            if let Some(last) = last_seq {
-                let loss_count = seq.wrapping_sub(last) as i32 - 1;
-                if loss_count > 0 {
-                    // 1. 优先尝试 FEC 恢复最近丢失的那一帧
-                    // Opus 的 FEC 数据存储在当前包(data)中，用于恢复“前一帧”
-                    if let Ok(len) = codec.decode_fec(&data, &mut pcm_buf) {
-                        let _ = player.write(&pcm_buf[..len]);
+            // 检查包是否迟到（目标时间已过）
+            if current_server_time > pkt.timestamp {
+                let late_ms = (current_server_time - pkt.timestamp) / 1000;
+                if late_ms > 50 {
+                    // 迟到超过 50ms，直接丢弃
+                    continue;
+                }
+                // 轻微迟到（<50ms），尝试播放
             }
 
-                    // 2. 如果丢包超过 1 帧，剩下的帧只能靠丢包补偿(PLC)
-                    for _ in 0..(loss_count - 1) {
-                        if let Ok(len) = codec.decode_loss(&mut pcm_buf) {
-                            let _ = player.write(&pcm_buf[..len]);
-                        }
-                    }
-                }
-            }
-            last_seq = Some(seq);
+            last_seq = Some(pkt.seq);
 
-            // 3. 正常解码当前帧
-            let len = codec.decode(&data, &mut pcm_buf)?;
-            player.write(&pcm_buf[..len])?;
+            // 精确等待到播放时间
+            loop {
+                let now = now_us();
+                let current_server_time_precise = clock.lock().await.to_server_time(now);
+
+                if current_server_time_precise >= pkt.timestamp {
+                    break;
+                }
+
+                let wait_us = (pkt.timestamp - current_server_time_precise) as u64;
+
+                if wait_us > 1000 {
+                    tokio::time::sleep(Duration::from_micros(wait_us as u64)).await;
                 } else {
-            tokio::time::sleep(Duration::from_millis(10)).await;
+                    // 小于 1ms，直接播放，让播放时机稍微早一点点
+                    break;
+                }
+            }
+
+            // 解码并播放
+            if let Ok(len) = codec.decode(&pkt.data, &mut pcm_buf) {
+                let _ = player.write(&pcm_buf[..len]);
+            }
+        } else {
+            tokio::time::sleep(Duration::from_micros(100)).await;
         }
     }
 }

examples/stereo/src/audio/player.rs

@@ -52,18 +52,19 @@ fn setup_pcm(pcm: &PCM, sample_rate: u32, channels: u16) -> Result<()> {
     hwp.set_rate(sample_rate, alsa::ValueOr::Nearest)?;
     hwp.set_channels(channels as u32)?;
 
-    // // 设置较大的缓冲区以减少由于调度抖动导致的断音
-    // // 48000Hz * 0.2s = 9600 samples
-    // let buffer_size = (sample_rate as f64 * 0.2) as u32;
-    // let period_size = buffer_size / 4;
-    // hwp.set_buffer_size_near(buffer_size as alsa::pcm::Frames)?;
-    // hwp.set_period_size_near(period_size as alsa::pcm::Frames, alsa::ValueOr::Nearest)?;
+    // 设置较大的缓冲区以减少由于调度抖动和设备重初始化导致的断音/卡顿
+    // 使用 100ms 缓冲区，既能防止 underrun，又不会引入过大延迟
+    let buffer_size = (sample_rate as f64 * 0.1) as u32; // 100ms 缓冲
+    let period_size = buffer_size / 4; // 25ms 周期
+    hwp.set_buffer_size_near(buffer_size as alsa::pcm::Frames)?;
+    hwp.set_period_size_near(period_size as alsa::pcm::Frames, alsa::ValueOr::Nearest)?;
 
     pcm.hw_params(&hwp).context("Failed to set HwParams")?;
 
     let swp = pcm.sw_params_current()?;
-    // 设置 start_threshold 为 buffer_size 的一半，确保缓冲区有足够数据再开始播放
-    // swp.set_start_threshold(buffer_size as alsa::pcm::Frames / 2)?;
+    // 设置 start_threshold，当缓冲区有 1 个 period 数据时就开始播放
+    // 这样可以快速启动，同时保持足够的缓冲余量
+    swp.set_start_threshold(period_size as alsa::pcm::Frames)?;
     pcm.sw_params(&swp)?;
     pcm.prepare()?;
     Ok(())

examples/stereo/src/utils/jitter_buffer.rs

@@ -1,100 +0,0 @@
-use std::cmp::Ordering;
-use std::cmp::Reverse;
-use std::collections::BinaryHeap;
-
-use crate::net::protocol::AudioPacket;
-
-#[derive(Debug)]
-struct OrderedPacket {
-    seq: u32,
-    timestamp: u128,
-    data: Vec<u8>,
-}
-
-// 仅针对序列号进行比较，处理回绕逻辑
-impl PartialEq for OrderedPacket {
-    fn eq(&self, other: &Self) -> bool {
-        self.seq == other.seq
-    }
-}
-
-impl Eq for OrderedPacket {}
-
-impl PartialOrd for OrderedPacket {
-    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
-        Some(self.cmp(other))
-    }
-}
-
-impl Ord for OrderedPacket {
-    fn cmp(&self, other: &Self) -> Ordering {
-        // 使用 wrapping_sub 处理 u32 回绕 (Rollover)
-        let diff = self.seq.wrapping_sub(other.seq) as i32;
-        diff.cmp(&0)
-    }
-}
-
-pub struct JitterBuffer {
-    // 使用 Reverse 将 BinaryHeap 变为小顶堆，避免手动实现大量 Trait
-    buffer: BinaryHeap<Reverse<OrderedPacket>>,
-    last_played_seq: Option<u32>,
-    pub target_delay_us: u128,
-    min_packets: usize, // 最小缓冲包数，防止微小抖动
-}
-
-impl JitterBuffer {
-    pub fn new(target_delay_us: u128, min_packets: usize) -> Self {
-        Self {
-            buffer: BinaryHeap::with_capacity(32), // 预分配初始容量
-            last_played_seq: None,
-            target_delay_us,
-            min_packets,
-        }
-    }
-
-    pub fn push(&mut self, packet: AudioPacket) {
-        // 1. 处理序列号回绕的丢包逻辑
-        if let Some(last) = self.last_played_seq {
-            let diff = packet.seq.wrapping_sub(last) as i32;
-            if diff <= 0 {
-                return; // 这是一个延迟到达的旧包，直接丢弃
-            }
-        }
-
-        self.buffer.push(Reverse(OrderedPacket {
-            seq: packet.seq,
-            timestamp: packet.timestamp,
-            data: packet.data,
-        }));
-    }
-
-    pub fn pop_frame(&mut self, current_time: u128) -> Option<(u32, Vec<u8>)> {
-        // 2. 预缓冲逻辑：如果包量太少，先不播放，等待填充
-        if self.buffer.len() < self.min_packets && self.last_played_seq.is_none() {
-            return None;
-        }
-
-        // 3. 检查堆顶元素
-        if let Some(Reverse(pkt)) = self.buffer.peek() {
-            // 判断是否到达播放时间（考虑目标延迟）
-            if current_time >= pkt.timestamp + (self.target_delay_us / 1000) {
-                let Reverse(pkt) = self.buffer.pop().unwrap();
-                self.last_played_seq = Some(pkt.seq);
-                return Some((pkt.seq, pkt.data));
-            }
-        }
-        None
-    }
-
-    /// 如果缓冲区堆积过大，可以主动跳帧以降低延迟
-    pub fn shrink_to_fit_latency(&mut self, max_size: usize) {
-        while self.buffer.len() > max_size {
-            self.buffer.pop();
-        }
-    }
-
-    pub fn clear(&mut self) {
-        self.buffer.clear();
-        self.last_played_seq = None;
-    }
-}

examples/stereo/src/utils/mod.rs

@@ -1,3 +1,2 @@
 pub mod alsa;
-pub mod jitter_buffer;
 pub mod sync;

examples/stereo/src/utils/sync.rs

@@ -10,10 +10,38 @@ pub fn now_us() -> u128 {
 }
 
 /// 时钟同步管理器，用于计算主从节点间的时钟偏移
+/// 采用改进的 NTP 算法 + Kalman 滤波思想
 pub struct ClockSync {
-    offsets: VecDeque<i128>,
+    /// 偏移量样本窗口
+    offsets: VecDeque<OffsetSample>,
+    /// 当前估计的时钟偏移 (server_time - client_time)
     pub current_offset: i128,
+    /// RTT 样本窗口
+    rtts: VecDeque<i128>,
+    /// 当前估计的最小 RTT
+    min_rtt: i128,
+    /// 窗口大小
     window_size: usize,
+    /// 时钟漂移率 (ppm: parts per million)
+    /// 正值表示从节点时钟比主节点快
+    drift_rate: f64,
+    /// 上次更新时间
+    last_update_time: u128,
+    /// 漂移率估计窗口
+    drift_samples: VecDeque<DriftSample>,
+}
+
+#[derive(Clone, Copy)]
+struct OffsetSample {
+    offset: i128,
+    rtt: i128,
+    timestamp: u128,
+}
+
+#[derive(Clone, Copy)]
+struct DriftSample {
+    offset: i128,
+    timestamp: u128,
 }
 
 impl ClockSync {
@@ -21,43 +49,176 @@ impl ClockSync {
         Self {
             offsets: VecDeque::with_capacity(window_size),
             current_offset: 0,
+            rtts: VecDeque::with_capacity(window_size),
+            min_rtt: i128::MAX,
             window_size,
+            drift_rate: 0.0,
+            last_update_time: now_us(),
+            drift_samples: VecDeque::with_capacity(60), // 保留 60 秒的样本
         }
     }
 
-    /// 更新时钟偏移估计
+    /// 更新时钟偏移估计 (NTP 算法)
+    /// 
+    /// NTP 时间戳标记:
+    /// t1 = client_send_ts  : 客户端发送 Ping 的时间
+    /// t2 = server_ts       : 服务器接收 Ping 的时间  
+    /// t3 = server_ts       : 服务器发送 Pong 的时间 (假设处理时间忽略不计)
+    /// t4 = client_recv_ts  : 客户端接收 Pong 的时间
+    ///
+    /// RTT = (t4 - t1) - (t3 - t2) = (t4 - t1) (因为 t3 = t2)
+    /// Offset = ((t2 - t1) + (t3 - t4)) / 2 = ((t2 - t1) + (t2 - t4)) / 2
+    ///        = t2 - (t1 + t4) / 2
     pub fn update(&mut self, client_send_ts: u128, server_ts: u128, client_recv_ts: u128) {
-        let rtt = (client_recv_ts - client_send_ts) as i128;
-        // 基础过滤：如果 RTT 过大则忽略 (例如局域网内 > 100ms)
-        if rtt > 100_000 {
+        let t1 = client_send_ts as i128;
+        let t2 = server_ts as i128;
+        let t4 = client_recv_ts as i128;
+
+        let rtt = t4 - t1;
+
+        // 过滤异常 RTT (局域网内 > 100ms 视为异常)
+        if rtt < 0 || rtt > 100_000 {
             return;
         }
 
-        // 时钟偏移 = 主节点时间 - 从节点时间
-        // 假设主节点收到 Ping 的时间点在 (发送时间 + 接收时间) / 2
-        let estimated_server_time = server_ts as i128 + rtt / 2;
-        let offset = estimated_server_time - client_recv_ts as i128;
+        // 计算时钟偏移: offset = server_time - client_time
+        // offset = t2 - (t1 + t4) / 2
+        let offset = t2 - (t1 + t4) / 2;
 
-        self.offsets.push_back(offset);
+        // 更新 RTT 窗口
+        self.rtts.push_back(rtt);
+        if self.rtts.len() > self.window_size {
+            self.rtts.pop_front();
+        }
+        self.min_rtt = *self.rtts.iter().min().unwrap_or(&rtt);
+
+        // 更新偏移量窗口
+        let sample = OffsetSample {
+            offset,
+            rtt,
+            timestamp: client_recv_ts,
+        };
+        self.offsets.push_back(sample);
         if self.offsets.len() > self.window_size {
             self.offsets.pop_front();
         }
 
-        // 计算中位数偏移，以增强抗干扰能力
-        let mut sorted: Vec<i128> = self.offsets.iter().cloned().collect();
-        sorted.sort_unstable();
-        if !sorted.is_empty() {
-            self.current_offset = sorted[sorted.len() / 2];
+        // 偏移量估计: 使用低 RTT 样本的中位数
+        // 原理: RTT 较小的样本受网络抖动影响小，时间测量更准确
+        let mut low_rtt_offsets: Vec<i128> = self
+            .offsets
+            .iter()
+            .filter(|s| s.rtt <= self.min_rtt + 5000) // 5ms 容差
+            .map(|s| s.offset)
+            .collect();
+
+        if !low_rtt_offsets.is_empty() {
+            low_rtt_offsets.sort_unstable();
+            let new_offset = low_rtt_offsets[low_rtt_offsets.len() / 2];
+
+            // 漂移率估计
+            self.estimate_drift(new_offset, client_recv_ts);
+
+            // 平滑更新偏移量 (避免突变)
+            let alpha = 0.3; // 低通滤波系数
+            self.current_offset =
+                (alpha * new_offset as f64 + (1.0 - alpha) * self.current_offset as f64) as i128;
+        }
+
+        self.last_update_time = client_recv_ts;
+    }
+
+    /// 估计时钟漂移率
+    /// 时钟漂移率 = d(offset) / dt
+    fn estimate_drift(&mut self, offset: i128, timestamp: u128) {
+        self.drift_samples.push_back(DriftSample { offset, timestamp });
+        if self.drift_samples.len() > 60 {
+            self.drift_samples.pop_front();
+        }
+
+        // 至少需要 10 秒的数据才能估计漂移
+        if self.drift_samples.len() < 10 {
+            return;
+        }
+
+        // 使用线性回归估计漂移率
+        let first = self.drift_samples.front().unwrap();
+        let last = self.drift_samples.back().unwrap();
+
+        let dt = (last.timestamp - first.timestamp) as f64;
+        let d_offset = (last.offset - first.offset) as f64;
+
+        if dt > 10_000_000.0 {
+            // 超过 10 秒
+            // drift_rate 单位: 微秒/秒 = ppm
+            let new_drift = d_offset / (dt / 1_000_000.0);
+
+            // 平滑更新漂移率
+            let beta = 0.1;
+            self.drift_rate = beta * new_drift + (1.0 - beta) * self.drift_rate;
         }
     }
 
     /// 将本地时间转换为服务器(主节点)时间
+    /// 考虑时钟漂移补偿
     pub fn to_server_time(&self, client_time: u128) -> u128 {
-        (client_time as i128 + self.current_offset) as u128
+        let base_server_time = (client_time as i128 + self.current_offset) as u128;
+
+        // 漂移补偿: 根据距离上次同步的时间，补偿时钟漂移
+        let elapsed_since_update = client_time.saturating_sub(self.last_update_time) as f64;
+        let drift_correction = (self.drift_rate * elapsed_since_update / 1_000_000.0) as i128;
+
+        (base_server_time as i128 + drift_correction) as u128
     }
 
     /// 将服务器(主节点)时间转换为本地时间
     pub fn to_client_time(&self, server_time: u128) -> u128 {
+        // 简化版本，不考虑漂移补偿 (播放时主要用 to_server_time)
         (server_time as i128 - self.current_offset) as u128
     }
+
+    /// 获取当前估计的 RTT (微秒)
+    pub fn get_rtt(&self) -> i128 {
+        self.min_rtt
+    }
+
+    /// 获取当前时钟漂移率 (ppm)
+    pub fn get_drift_rate(&self) -> f64 {
+        self.drift_rate
+    }
+
+    /// 获取同步质量评估 (0-100, 越高越好)
+    pub fn get_sync_quality(&self) -> u8 {
+        if self.offsets.is_empty() {
+            return 0;
+        }
+
+        // 基于 RTT 稳定性和偏移量方差评估
+        let rtt_variance = self.calculate_variance(&self.rtts.iter().copied().collect::<Vec<_>>());
+        let offset_variance = self.calculate_variance(
+            &self.offsets.iter().map(|s| s.offset).collect::<Vec<_>>(),
+        );
+
+        // RTT 越稳定，方差越小，质量越高
+        let rtt_score = ((100_000.0 - rtt_variance.min(100_000.0)) / 100_000.0 * 50.0) as u8;
+        let offset_score = ((50_000.0 - offset_variance.min(50_000.0)) / 50_000.0 * 50.0) as u8;
+
+        rtt_score + offset_score
+    }
+
+    fn calculate_variance(&self, samples: &[i128]) -> f64 {
+        if samples.is_empty() {
+            return 0.0;
+        }
+        let mean = samples.iter().sum::<i128>() as f64 / samples.len() as f64;
+        let variance = samples
+            .iter()
+            .map(|&x| {
+                let diff = x as f64 - mean;
+                diff * diff
+            })
+            .sum::<f64>()
+            / samples.len() as f64;
+        variance.sqrt()
+    }
 }

examples/xiaozhi/README.md

@@ -136,15 +136,13 @@ APP_CONFIG = {
 
 ### Q：唤醒词一直没有反应？
 
-由于小爱音箱远场拾音音量较小，有时可能会识别不清，你可以调大 `config.py` 配置文件里的 `boost` 参数，然后重启应用 / Docker 试试看。
+如果唤醒词还是不敏感，可以先调低 `vad.threshold`，然后重启应用 / Docker 试试看。
 
 ```py
 APP_CONFIG = {
     "vad": {
-        # 小爱音箱录音音量较小，需要后期放大一下
-        "boost": 100,
-        # boost 调大后，语音检测阈值可能也需要一起调大些
-        "threshold": 0.50,
+        # 语音检测阈值（0-1，越小越灵敏）
+        "threshold": 0.05,
     },
     # ... 其他配置
 }

examples/xiaozhi/config.py

@@ -53,8 +53,6 @@ APP_CONFIG = {
         "after_wakeup": after_wakeup,
     },
     "vad": {
-        # 录音音量增强倍数（小爱音箱录音音量较小，需要后期放大一下）
-        "boost": 10,
         # 语音检测阈值（0-1，越小越灵敏）
         "threshold": 0.10,
         # 最小语音时长（ms）

examples/xiaozhi/xiaozhi/services/audio/kws/sherpa.py

@@ -9,10 +9,10 @@ class _SherpaOnnx:
         self.keyword_spotter = sherpa_onnx.KeywordSpotter(
             provider="cpu",
             num_threads=1,
-            max_active_paths=4,
+            max_active_paths=8,
             keywords_score=2.0,
             keywords_threshold=0.2,
-            num_trailing_blanks=1,
+            num_trailing_blanks=0,
             keywords_file=get_model_file_path("keywords.txt"),
             tokens=get_model_file_path("tokens.txt"),
             encoder=get_model_file_path("encoder.onnx"),

examples/xiaozhi/xiaozhi/services/audio/stream.py

@@ -1,9 +1,6 @@
 import uuid
 from typing import Any, Callable, ClassVar, Optional
 
-import numpy as np
-
-from config import APP_CONFIG
 from xiaozhi.ref import get_xiaoai
 
 
@@ -88,10 +85,7 @@ class MyStream:
             return
 
         if len(data) > 0:
-            samples = np.frombuffer(data, dtype=np.int16)
-            # 小爱音箱录音音量较小，需要后期放大一下
-            samples = samples * APP_CONFIG["vad"]["boost"]
-            self.input_bytes.extend(samples.tobytes())
+            self.input_bytes.extend(data)
 
     def read(self, num_frames=None, exception_on_overflow=False) -> bytes:
         if num_frames is None:

packages/client-patch/README.md

@@ -14,8 +14,8 @@
 
 你可以直接在 [Github Releases](https://github.com/idootop/open-xiaoai/releases) 页面下载打包好的固件：
 
-- [Xiaomi 智能音箱 Pro v1.58.1](https://github.com/idootop/open-xiaoai/releases/tag/OH2P_1.58.1)
-- [小爱音箱 Pro v1.88.221](https://github.com/idootop/open-xiaoai/releases/tag/LX06_1.88.221)
+- [Xiaomi 智能音箱 Pro v1.58.6](https://github.com/idootop/open-xiaoai/releases/tag/OH2P_1.58.6)
+- [小爱音箱 Pro v1.94.13](https://github.com/idootop/open-xiaoai/releases/tag/LX06_1.94.13)
 
 > [!TIP]
 > 里面有两个文件，下载 `patched` 那个：
@@ -33,8 +33,8 @@
 > [!CAUTION]
 > 当前支持的最新固件版本为：
 >
-> - Xiaomi 智能音箱 Pro 👉 [v1.58.1](https://github.com/idootop/open-xiaoai/releases/tag/OH2P_1.58.1)
-> - 小爱音箱 Pro 👉 [v1.88.221](https://github.com/idootop/open-xiaoai/releases/tag/LX06_1.88.221)
+> - Xiaomi 智能音箱 Pro 👉 [v1.58.6](https://github.com/idootop/open-xiaoai/releases/tag/OH2P_1.58.6)
+> - 小爱音箱 Pro 👉 [v1.94.13](https://github.com/idootop/open-xiaoai/releases/tag/LX06_1.94.13)
 >
 > 更新版本的固件可能存在变化，导致刷机失败，设备变砖，请自行评估风险。
 

packages/client-rust/init.sh

@@ -1,16 +1,5 @@
 #!/bin/sh
 
-cat << 'EOF'
-
-▄▖      ▖▖▘    ▄▖▄▖
-▌▌▛▌█▌▛▌▚▘▌▀▌▛▌▌▌▐ 
-▙▌▙▌▙▖▌▌▌▌▌█▌▙▌▛▌▟▖
-  ▌                 
-
-v1.0.0  by: https://del.wang
-
-EOF
-
 set -e
 
 

packages/client-rust/src/services/audio/record.rs

@@ -16,6 +16,8 @@ enum State {
     Recording,
 }
 
+const A113_CAPTURE_BITS_PER_SAMPLE: u16 = 32;
+
 pub struct AudioRecorder {
     state: Arc<Mutex<State>>,
     arecord_thread: Arc<Mutex<Option<Child>>>,
@@ -69,9 +71,97 @@ impl AudioRecorder {
             return Ok(());
         }
 
-        let config = config.unwrap_or_else(|| (*AUDIO_CONFIG).clone());
+        let requested_config = config.unwrap_or_else(|| (*AUDIO_CONFIG).clone());
+        let capture_config = capture_config_for_recording(&requested_config);
+        let mut arecord_thread = spawn_arecord(&capture_config)?;
 
-        let mut arecord_thread = Command::new("arecord")
+        let mut stdout = arecord_thread.stdout.take().unwrap();
+        let read_thread = tokio::spawn(async move {
+            let bytes_per_sample = (capture_config.bits_per_sample.max(8) / 8) as usize;
+            let bytes_per_frame = bytes_per_sample * capture_config.channels.max(1) as usize;
+            let target_frames = capture_config.buffer_size.max(1) as usize;
+            let read_frames = capture_config.period_size.max(1) as usize;
+            let target_size = target_frames * bytes_per_frame;
+            let read_size = read_frames * bytes_per_frame;
+
+            let mut accumulated_data = Vec::new();
+            let mut buffer = vec![0u8; read_size];
+
+            loop {
+                match stdout.read(&mut buffer).await {
+                    Ok(size) if size > 0 => {
+                        accumulated_data.extend_from_slice(&buffer[..size]);
+                        while accumulated_data.len() >= target_size {
+                            let data_to_send =
+                                accumulated_data.drain(..target_size).collect::<Vec<u8>>();
+                            let data_to_send =
+                                transform_stream_chunk(data_to_send, &requested_config, &capture_config);
+                            if !data_to_send.is_empty() {
+                                let _ = on_stream(data_to_send).await;
+                            }
+                        }
+                    }
+                    _ => break,
+                }
+            }
+
+            let _ = AudioRecorder::instance().stop_recording().await;
+        });
+
+        self.arecord_thread.lock().await.replace(arecord_thread);
+        self.read_thread.lock().await.replace(read_thread);
+
+        *state = State::Recording;
+        Ok(())
+    }
+}
+
+fn capture_config_for_recording(requested: &AudioConfig) -> AudioConfig {
+    let mut capture = requested.clone();
+    if requested.bits_per_sample == 16 {
+        capture.bits_per_sample = A113_CAPTURE_BITS_PER_SAMPLE;
+    }
+    capture
+}
+
+fn transform_stream_chunk(
+    chunk: Vec<u8>,
+    requested: &AudioConfig,
+    capture: &AudioConfig,
+) -> Vec<u8> {
+    if requested.bits_per_sample != 16 || capture.bits_per_sample != A113_CAPTURE_BITS_PER_SAMPLE {
+        return chunk;
+    }
+    convert_a113_s32_to_s16(&chunk)
+}
+
+fn convert_a113_s32_to_s16(chunk: &[u8]) -> Vec<u8> {
+    if chunk.len() % 4 != 0 {
+        return Vec::new();
+    }
+
+    let frame_count = chunk.len() / 4;
+    let mut out = vec![0u8; frame_count * 2];
+
+    for frame in 0..frame_count {
+        let base = frame * 4;
+        let sample = i32::from_le_bytes([
+            chunk[base],
+            chunk[base + 1],
+            chunk[base + 2],
+            chunk[base + 3],
+        ]);
+        // A113 PDM data lives in lower 24 bits of S32_LE: shift by 8 (not 16).
+        let mapped = (sample >> 8).clamp(i16::MIN as i32, i16::MAX as i32) as i16;
+        let out_base = frame * 2;
+        out[out_base..out_base + 2].copy_from_slice(&mapped.to_le_bytes());
+    }
+
+    out
+}
+
+fn spawn_arecord(config: &AudioConfig) -> Result<Child, AppError> {
+    let child = Command::new("arecord")
         .args([
             "--quiet",
             "-t",
@@ -91,36 +181,5 @@ impl AudioRecorder {
         ])
         .stdout(Stdio::piped())
         .spawn()?;
-
-        let mut stdout = arecord_thread.stdout.take().unwrap();
-        let read_thread = tokio::spawn(async move {
-            let size = (config.bits_per_sample / 8) as usize;
-            let target_size = config.buffer_size as usize * size;
-
-            let mut accumulated_data = Vec::new();
-            let mut buffer = vec![0u8; config.period_size as usize * size];
-
-            loop {
-                match stdout.read(&mut buffer).await {
-                    Ok(size) if size > 0 => {
-                        accumulated_data.extend_from_slice(&buffer[..size]);
-                        if accumulated_data.len() >= target_size {
-                            let data_to_send =
-                                accumulated_data.drain(..target_size).collect::<Vec<u8>>();
-                            let _ = on_stream(data_to_send).await;
-                        }
-                    }
-                    _ => break,
-                }
-            }
-
-            let _ = AudioRecorder::instance().stop_recording().await;
-        });
-
-        self.arecord_thread.lock().await.replace(arecord_thread);
-        self.read_thread.lock().await.replace(read_thread);
-
-        *state = State::Recording;
-        Ok(())
-    }
+    Ok(child)
 }