feat(xiaozhi): 支持在 python 端监听 event 和 runShell

examples/xiaozhi/.vscode/launch.json

@@ -0,0 +1,25 @@
+{
+  "version": "0.2.0",
+  "configurations": [
+    {
+      "name": "Python: App",
+      "type": "debugpy",
+      "request": "launch",
+      "program": "${workspaceFolder}/main.py",
+      "justMyCode": false,
+      "console": "internalConsole",
+      "internalConsoleOptions": "openOnSessionStart",
+      "envFile": "${workspaceFolder}/.env"
+    },
+    {
+      "name": "Python: Current",
+      "type": "debugpy",
+      "request": "launch",
+      "program": "${file}",
+      "justMyCode": false,
+      "console": "internalConsole",
+      "internalConsoleOptions": "openOnSessionStart",
+      "envFile": "${workspaceFolder}/.env"
+    }
+  ]
+}

examples/xiaozhi/boot.sh

@@ -0,0 +1,25 @@
+#! /bin/sh
+
+set -e
+
+DOWNLOAD_BASE_URL="https://gitee.com/idootop/artifacts/releases/download"
+
+WORK_DIR="/data/open-xiaoai/scripts"
+
+if [ ! -d "$WORK_DIR" ]; then
+    mkdir -p "$WORK_DIR"
+fi
+
+if [ ! -f "$WORK_DIR/client-boot.sh" ]; then
+    curl -L -# -o "$WORK_DIR/client-boot.sh" "$DOWNLOAD_BASE_URL/open-xiaoai-client/boot.sh"
+fi
+
+if [ ! -f "$WORK_DIR/kws-boot.sh" ]; then
+    curl -L -# -o "$WORK_DIR/kws-boot.sh" "$DOWNLOAD_BASE_URL/open-xiaoai-kws/boot.sh"
+fi
+
+kill -9 `ps|grep "open-xiaoai/kws/monitor"|grep -v grep|awk '{print $1}'` > /dev/null 2>&1 || true
+
+sh "$WORK_DIR/kws-boot.sh" --no-monitor > /dev/null 2>&1 &
+
+sh "$WORK_DIR/client-boot.sh"

examples/xiaozhi/init.sh

@@ -0,0 +1,25 @@
+#! /bin/sh
+
+set -e
+
+DOWNLOAD_BASE_URL="https://gitee.com/idootop/artifacts/releases/download"
+
+WORK_DIR="/data/open-xiaoai/scripts"
+
+if [ ! -d "$WORK_DIR" ]; then
+    mkdir -p "$WORK_DIR"
+fi
+
+if [ ! -f "$WORK_DIR/client.sh" ]; then
+    curl -L -# -o "$WORK_DIR/client.sh" "$DOWNLOAD_BASE_URL/open-xiaoai-client/init.sh"
+fi
+
+if [ ! -f "$WORK_DIR/kws.sh" ]; then
+    curl -L -# -o "$WORK_DIR/kws.sh" "$DOWNLOAD_BASE_URL/open-xiaoai-kws/init.sh"
+fi
+
+kill -9 `ps|grep "open-xiaoai/kws/monitor"|grep -v grep|awk '{print $1}'` > /dev/null 2>&1 || true
+
+sh "$WORK_DIR/kws.sh" --no-monitor > /dev/null 2>&1 &
+
+sh "$WORK_DIR/client.sh"

examples/xiaozhi/main.py

@@ -4,7 +4,7 @@ import sys
 
 from xiaozhi.services.audio.setup import setup_opus
 from xiaozhi.utils.logging_config import setup_logging
-from xiaozhi.xiaoai import XiaoAi
+from xiaozhi.xiaoai import XiaoAI
 from xiaozhi.xiaozhi import XiaoZhi
 
 logger = logging.getLogger("Main")
@@ -26,7 +26,7 @@ def setup_graceful_shutdown():
 
 
 if __name__ == "__main__":
-    XiaoAi.setup_mode()
+    XiaoAI.setup_mode()
     setup_logging()
     setup_graceful_shutdown()
     sys.exit(main())

examples/xiaozhi/src/lib.rs

@@ -1,6 +1,8 @@
 use open_xiaoai::services::connect::message::MessageManager;
+use open_xiaoai::services::connect::rpc::RPC;
 use pyo3::prelude::*;
 use pyo3::types::PyBytes;
+use serde_json::json;
 use server::AppServer;
 
 pub mod macros;
@@ -26,10 +28,29 @@ fn start_server(py: Python) -> PyResult<Bound<PyAny>> {
     })
 }
 
+#[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) => serde_json::to_string(&res.data.unwrap()).unwrap(),
+        };
+        Ok(result)
+    })
+}
+
 #[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)?)?;
     crate::python::init_module(&m)?;
     Ok(())
 }

examples/xiaozhi/src/server.rs

@@ -1,3 +1,4 @@
+use crate::python::PythonManager;
 use open_xiaoai::base::{AppError, VERSION};
 use open_xiaoai::services::audio::config::AudioConfig;
 use open_xiaoai::services::connect::data::{Event, Request, Response, Stream};
@@ -7,13 +8,12 @@ use open_xiaoai::services::connect::rpc::RPC;
 use open_xiaoai::services::speaker::SpeakerManager;
 use open_xiaoai::utils::task::TaskManager;
 use pyo3::types::PyBytes;
+use pyo3::types::PyString;
 use pyo3::Python;
 use serde_json::json;
 use tokio::net::{TcpListener, TcpStream};
 use tokio_tungstenite::accept_async;
 
-use crate::python::PythonManager;
-
 pub struct AppServer;
 
 async fn test() -> Result<(), AppError> {
@@ -127,6 +127,8 @@ async fn on_stream(stream: Stream) -> Result<(), AppError> {
 }
 
 async fn on_event(event: Event) -> Result<(), AppError> {
-    crate::pylog!("🔥 收到 Event: {:?}", event);
+    let event_json = serde_json::to_string(&event)?;
+    let data = Python::with_gil(|py| PyString::new(py, &event_json).into());
+    PythonManager::instance().call_fn("on_event", Some(data))?;
     Ok(())
 }

examples/xiaozhi/xiaozhi/services/audio/codec.py

@@ -1,14 +1,15 @@
 import logging
 import queue
-import numpy as np
-import pyaudio
-import opuslib
-from xiaozhi.services.protocols.typing import AudioConfig
-import time
 import sys
+import time
+
+import numpy as np
+import opuslib
+import pyaudio
 
 from xiaozhi.services.audio.stream import MyAudio
-from xiaozhi.xiaoai import XiaoAi
+from xiaozhi.services.protocols.typing import AudioConfig
+from xiaozhi.xiaoai import XiaoAI
 
 logger = logging.getLogger("AudioCodec")
 
@@ -24,14 +25,14 @@ class AudioCodec:
         self.opus_encoder = None
         self.opus_decoder = None
         self.audio_decode_queue = queue.Queue()
-        self._is_closing = False 
+        self._is_closing = False
 
         self._initialize_audio()
 
     def _initialize_audio(self):
         """初始化音频设备和编解码器"""
         try:
-            self.audio = MyAudio() if XiaoAi.mode == "xiaoai" else pyaudio.PyAudio()
+            self.audio = MyAudio() if XiaoAI.mode == "xiaoai" else pyaudio.PyAudio()
 
             # 初始化音频输入流
             self.input_stream = self.audio.open(
@@ -50,7 +51,7 @@ class AudioCodec:
                 output=True,
                 frames_per_buffer=AudioConfig.FRAME_SIZE,
             )
-            
+
             # 初始化Opus编码器
             self.opus_encoder = opuslib.Encoder(
                 fs=AudioConfig.SAMPLE_RATE,
@@ -193,7 +194,7 @@ class AudioCodec:
                     if self.output_stream.is_active():
                         self.output_stream.stop_stream()
                     self.output_stream.close()
-                except Exception as e:
+                except Exception:
                     # logger.warning(f"关闭旧输出流时出错: {e}")
                     pass
 

examples/xiaozhi/xiaozhi/utils/base.py

@@ -0,0 +1,21 @@
+import json
+
+
+def to_set(data):
+    if isinstance(data, list):
+        return list(set(data))
+    return data
+
+
+def json_encode(obj, pretty=False):
+    try:
+        return json.dumps(obj, ensure_ascii=False, indent=4 if pretty else None)
+    except Exception as _:
+        return None
+
+
+def json_decode(text):
+    try:
+        return json.loads(text)
+    except Exception:
+        return None

examples/xiaozhi/xiaozhi/xiaoai.py

@@ -1,20 +1,23 @@
 import argparse
 import asyncio
+import threading
 
 import numpy as np
 import open_xiaoai_server
 
 from xiaozhi.services.audio.stream import GlobalStream
+from xiaozhi.utils.base import json_decode
 
 
-class XiaoAi:
+class XiaoAI:
     mode = "xiaoai"
-    loop = asyncio.new_event_loop()
+    sync_loop: asyncio.AbstractEventLoop = None
+    async_loop: asyncio.AbstractEventLoop = None
 
     @classmethod
     def setup_mode(cls):
         parser = argparse.ArgumentParser(
-            description="小爱音箱接入小智 AI 演示 | by: https://del.wang"
+            description="小爱音箱接入小智 AI | by: https://del.wang"
         )
         parser.add_argument(
             "--mode",
@@ -38,10 +41,64 @@ class XiaoAi:
             return await open_xiaoai_server.on_output_data(data)
 
         future = on_output_data_async(data)
-        cls.loop.run_until_complete(future)
+        cls.sync_loop.run_until_complete(future)
+
+    @classmethod
+    async def run_shell(cls, script: str, timeout_millis: float = 10 * 1000):
+        return await open_xiaoai_server.run_shell(script, timeout_millis)
+
+    @classmethod
+    async def on_event(cls, event: str):
+        event_json = json_decode(event) or {}
+        event_data = event_json.get("data", {})
+        event_type = event_json.get("event")
+
+        if not event_json.get("event"):
+            print(f"❌ Event 解析失败: {event}")
+            return
+
+        if event_type == "kws":
+            if event_data == "Started":
+                print("🔥 自定义唤醒词已开启")
+                await cls.run_shell("/usr/sbin/tts_play.sh '自定义唤醒词已开启'")
+            else:
+                keyword = event_data.get("Keyword")
+                print(f"🔥 自定义唤醒词: {keyword}")
+        elif event_type == "instruction" and event_data.get("NewLine"):
+            line = json_decode(event_data.get("NewLine"))
+            if (
+                line
+                and line.get("header", {}).get("namespace") == "SpeechRecognizer"
+                and line.get("header", {}).get("name") == "RecognizeResult"
+            ):
+                text = line.get("payload", {}).get("results")[0].get("text")
+                if not text:
+                    print("🔥 唤醒小爱同学")
+                elif text and line.get("payload", {}).get("is_final"):
+                    print(f"🔥 收到用户指令: {text}")
+
+    @classmethod
+    def __init_background_event_loop(cls):
+        def run_event_loop():
+            cls.sync_loop = asyncio.new_event_loop()
+            cls.async_loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(cls.async_loop)
+            cls.async_loop.run_forever()
+
+        thread = threading.Thread(target=run_event_loop, daemon=True)
+        thread.start()
+
+    @classmethod
+    def __on_event(cls, event: str):
+        asyncio.run_coroutine_threadsafe(
+            cls.on_event(event),
+            cls.async_loop,
+        )
 
     @classmethod
     async def init_xiaoai(cls):
         GlobalStream().on_output_data = cls.on_output_data
         open_xiaoai_server.register_fn("on_input_data", cls.on_input_data)
+        open_xiaoai_server.register_fn("on_event", cls.__on_event)
+        cls.__init_background_event_loop()
         await open_xiaoai_server.start_server()

examples/xiaozhi/xiaozhi/xiaozhi.py

@@ -15,7 +15,7 @@ from xiaozhi.services.protocols.typing import (
 )
 from xiaozhi.services.protocols.websocket_protocol import WebsocketProtocol
 from xiaozhi.utils.config_manager import ConfigManager
-from xiaozhi.xiaoai import XiaoAi
+from xiaozhi.xiaoai import XiaoAI
 
 # 配置日志
 logger = logging.getLogger("XiaoZhi")
@@ -91,7 +91,7 @@ class XiaoZhi:
         time.sleep(0.1)
 
         # 初始化应用程序（移除自动连接）
-        asyncio.run_coroutine_threadsafe(XiaoAi.init_xiaoai(), self.loop)
+        asyncio.run_coroutine_threadsafe(XiaoAI.init_xiaoai(), self.loop)
         asyncio.run_coroutine_threadsafe(self._initialize_without_connect(), self.loop)
 
         # 启动主循环线程