zimonianhua
aafdf024d3
- 新增完整的 Home Assistant 集成示例，支持文本对话与上下文会话 - 实现小爱音箱事件解析与 TTS 播报，支持实时打断功能 - 添加连续会话管理，支持区域上下文注入和会话超时结束 - 提供完整的配置系统、类型定义和错误处理 - 包含 Rust 扩展模块用于小爱音箱通信，支持音频流处理 - 添加单元测试、集成测试脚本和 Docker 部署支持 - 提供详细的使用文档和配置说明
114 lines
3.0 KiB
Python
114 lines
3.0 KiB
Python
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 == []
|