Metadata-Version: 2.4
Name: xiaozhuiot-robot-sdk
Version: 0.1.1
Summary: Host-driven Xiaozhuiot robot director SDK with MQTT audio and playback acknowledgments
License: MIT License
        
        Copyright (c) 2026 Xiaozhuiot Robot SDK contributors
        
        Permission is hereby granted, free of charge, to any person obtaining a copy
        of this software and associated documentation files (the "Software"), to deal
        in the Software without restriction, including without limitation the rights
        to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
        copies of the Software, and to permit persons to whom the Software is
        furnished to do so, subject to the following conditions:
        
        The above copyright notice and this permission notice shall be included in all
        copies or substantial portions of the Software.
        
        THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
        IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
        FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
        AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
        LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
        OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
        SOFTWARE.
        
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: websockets<13,>=12
Provides-Extra: director
Requires-Dist: paho-mqtt<3,>=2.1; extra == "director"
Requires-Dist: cryptography>=41; extra == "director"
Dynamic: license-file

<!-- Generated by tools/generate_pypi_readme.py; edit INTEGRATION.md / API.md instead. -->
# Xiaozhuiot Robot SDK

发行名：xiaozhuiot-robot-sdk；Python 导入名：robot_voice。许可证：MIT。

# 从安装到听见声音

需要 Python 3.10+，以及服务提供方已部署的小智后端和对应的小程序账号。
先用电脑或有 FFmpeg 的 Linux 机器人跑通；机器人已有播放器时，最后只替换播放器接口。

## 1. 安装

在自己的工作目录打开终端，依次执行：

```sh
python3 -m venv .venv
.venv/bin/python -m pip install "xiaozhuiot-robot-sdk[director]"
```

创建 sdk_config.json（已有配置则保留）：

```json
{
  "ota_url": "https://YOUR_MANAGER_API_HOST/xiaozhi/ota/",
  "identity": {},
  "speaker_name": "现场观众",
  "timeout": 10
}
```

打开 sdk_config.json，把 ota_url 设置为服务提供方给你的地址。identity 保持 `{}`，使用当前机器的 MAC；speaker_name 是现场文字的说话人名称。更换机器后需重新检查绑定。

## 2. 检查设备是否绑定

执行下面这条命令，进入 Python 交互环境。后面的 Python 代码都在这里逐条输入：

```sh
.venv/bin/python -m asyncio
```

```python
from robot_voice import DeviceSDK
sdk = DeviceSDK(config_path="sdk_config.json")
registration = await sdk.check_registration()
registration.bound
```

如果返回 True，直接进入第 3 步。

如果返回 False，输入：

```python
registration.activation.code
```

把显示的激活码填到对应小程序的添加设备页面，绑定到智能体。绑定完成后再输入：

```python
registration = await sdk.check_registration()
registration.bound
```

返回 True 后继续。

## 3. 让设备上线

```python
director = await sdk.open_director()
director.ready
```

返回 True，表示设备在线。此时没有声音是正常的，还没有启动播放任务和导播。

## 4. 接上播放器

先确认系统已安装 FFmpeg，并能在另一个终端执行 `ffplay -version`。

回到刚才的 Python 环境：

```python
import asyncio
from robot_voice import FFplayPlayer, run_robot
player_task = asyncio.create_task(run_robot(director, FFplayPlayer))
```

现在去小程序选择这台设备，启动导播。你应该听到声音，而且能连续播放多段。
这两个接口已经包含在安装的库中，不需要导入 examples，也不需要自己处理播放回执。

没有声音时，先检查播放任务：

```python
player_task.done()
```

正常待命返回 False。如果返回 True，输入 `player_task.result()` 查看错误，例如缺少 ffplay 或播放失败。
不要重复创建任务：同一个 director 只能有一个接收者。

## 5. 提交一句现场文字

保持导播运行，输入：

```python
result = await sdk.send_asr_text("这个产品多少钱？")
result["queued"]
```

True 表示已入队，回复仍由刚才的播放器播放。False 表示当前没有运行中的导播。
以后把这句话换成机器人语音识别得到的最终文字即可；SDK 不负责录音和语音识别。

## 6. 停止

```python
try:
    await sdk.stop_director()
finally:
    player_task.cancel()
    try:
        await player_task
    except asyncio.CancelledError:
        pass
    finally:
        await sdk.disconnect()
```

这会请求停止服务端导播、关闭播放器并断开设备。停止接口需要配套后端支持；也可以先在小程序停止导播，再执行播放器取消和 disconnect。

## 7. 放进机器人程序

交互验证成功后，将以下代码保存为 robot_app.py，执行 `python robot_app.py`。在小程序启停导播。

```python
import asyncio
from robot_voice import DeviceSDK, FFplayPlayer, run_robot


async def main():
    sdk = DeviceSDK(config_path="sdk_config.json")
    try:
        registration = await sdk.check_registration()
        if not registration.bound:
            print("请在小程序绑定设备，激活码：", registration.activation.code)
            print("绑定后重新运行本程序。")
            return
        director = await sdk.open_director()
        print("设备已在线，请在小程序启动导播。")
        # Replace FFplayPlayer with your own AudioPlayer implementation on a robot.
        await run_robot(director, FFplayPlayer)
    finally:
        await sdk.disconnect()


if __name__ == "__main__":
    asyncio.run(main())
```

如果机器人能使用 ffplay，上述播放器可以直接用于实机联调，仍需验证实际声卡。
如果机器人已有音频系统，只需将 FFplayPlayer 换成自己的类：

```python
class RobotPlayer:
    def __init__(self, audio_format):
        # 按 audio_format 配置你们的 Opus 解码器和音频输出。
        # 每段播报会创建一个新实例。
        raise NotImplementedError("接入机器人的音频系统")

    async def feed(self, packet: bytes):
        # 将一包原始 Opus 送入解码/播放队列，尽快返回。
        raise NotImplementedError

    async def finish(self):
        # 等待解码、播放队列和声卡中的声音全部播完，再返回。
        raise NotImplementedError

    async def stop(self):
        # 停止声音、清空队列、释放资源；允许重复调用。
        raise NotImplementedError

player_task = asyncio.create_task(run_robot(director, RobotPlayer))
```

这是需要替换硬件调用的接口模板，不能原样播放。三个异步方法不得阻塞事件循环；finish 等待期间可能被取消，随后会调用 stop。

**finish 返回就是你的程序确认“实际播完”**，run_robot 随后代你调用播放完成接口。播放失败应抛出异常，不能假装成功。插播时旧段会停止，不确认旧段。

更多参数、返回值、异常和事件见 API 接口说明。普通接入无需阅读底层协议。


# Python API

首次使用先照着 接入教程 操作。以下接口均从已安装的 robot_voice 使用；异步接口在宿主的 asyncio 循环中调用。

## DeviceSDK(config_path="sdk_config.json")

创建 SDK 对象。config_path 是配置文件路径，默认读取工作目录 sdk_config.json；产品代码建议用绝对路径。构造不连接服务器、不启动播放。

配置：ota_url 为服务地址；identity 留空使用本机 MAC，也可指定 interface；speaker_name 为现场说话人名称（默认“现场观众”，1–64 字符）；timeout 为请求超时秒数（默认 10）。不要打印完整注册结果或配置。

## await sdk.check_registration()

检查绑定并取得连接配置，返回注册对象。

```python
registration = await sdk.check_registration()
if not registration.bound:
    print(registration.activation.code)
```

bound 为 bool；未绑定时 activation.code 为前端绑定所需的激活码。用户在前端绑定后再次调用，直到 bound=True。接口不替用户绑定。

## await sdk.open_director()

前提：已 check_registration 且 bound=True。返回 director，ready=True 表示设备在线。只上线，不开播。不能与独立 WebSocket 的 sdk.connect() 同时使用。

## await run_robot(director, player_factory, *, on_event=None)

```python
from robot_voice import run_robot, FFplayPlayer
await run_robot(director, FFplayPlayer)
```

持续接收音频并交给播放器。调用它表示宿主授权使用该播放器，并在播放器 finish 成功返回后代为调用 playback_finished；不会自动启动服务端导播或重连。

| 参数 | 类型与含义 |
| --- | --- |
| director | open_director 返回的会话 |
| player_factory | 同步函数或类，接收 AudioFormat，返回一个新的播放器；每段调用一次 |
| on_event | 可选同步函数，接收事件 dict，例如 `lambda e: print(e['type'])`；必须快速返回 |

一直运行至取消或发生错误，无业务返回值。异常会向调用方传播，并清理播放器；on_event 抛异常也会结束任务。宿主需要 await/监督任务，不能丢弃后台任务错误。同一个 director 只运行一个 run_robot，不再调用 receive 或手动 playback_finished。

取消时停止本地播放；已发出的完成请求会等待结果，不取消或重试。取消不会停止服务端导播，也不会断开 SDK，宿主自行调用 stop_director / disconnect。

### AudioFormat 与播放器接口

```python
from robot_voice import AudioFormat, AudioPlayer
```

AudioFormat 是只读数据对象：encoding='opus'，sample_rate 为协商采样率，channels=1，frame_duration 为帧时长（毫秒，当前 60）。AudioPlayer 是类型协议，不要求继承，实现以下方法即可。

| 方法 | 契约 |
| --- | --- |
| await player.feed(packet) | packet 是单包原始 Opus bytes，不是 WAV、Ogg 或 PCM。尽快放入有界播放队列；失败抛异常 |
| await player.finish() | 等待所有音频实际播完；返回即确认完成，空音频也需能正常返回；允许取消等待 |
| await player.stop() | 停止播放并清空缓冲、释放资源；可重复调用，包括 finish 成功/失败之后 |

每段使用新的播放器实例。feed 不得等整段播完；三个方法不得阻塞宿主事件循环。适配你们自己的驱动时，要同时接好解码格式、排空通知和取消操作。

### FFplayPlayer(audio_format)

库自带的可选播放器，使用系统 PATH 中的 ffplay，无需导入 examples。显式传给 run_robot 后，收到音频才启动播放进程。支持单声道 Opus / 60ms，finish 等待播放进程正常结束；播放收尾超时（30 秒）或失败会抛异常，不确认播完。需要预装 FFmpeg，实际声卡仍需实机验证。

## await sdk.start_director()

前提：已 open_director，并已在前端配置自动启动目标或存在可用实例。

```python
result = await sdk.start_director()
result["started"]
```

返回 dict：started 为 bool，code 为业务码，runtime_type 为后端运行类型。没有可用目标时 started=False，不会替你创建目标。首次联调也可以直接在小程序开播。

## await sdk.send_asr_text(text)

前提：已 open_director。text 是去除首尾空白后 1–4000 字符的现场文字。

```python
result = await sdk.send_asr_text("这个产品多少钱？")
result["queued"]
```

返回 dict：queued 为 bool，code、runtime_type 为后端状态。queued=True 只表示入队；False 表示没有运行中的导播。按配置组合“说话人说：正文”，不自动开播，不实现语音识别；重复提交会重复入队。

## await sdk.stop_director()

前提：已 open_director。停止服务端导播，清理旧音频并通知播放器中断，设备保持在线。
返回 dict：stopped 为 bool，code、runtime_type 为后端状态。启停及 ASR 接口需要配套 manager-api 支持。

## await sdk.disconnect()

关闭设备连接和 SDK 内部任务。先取消并等待 run_robot，再调用本接口。不代替 stop_director。

## 错误处理

SDKError 是 SDK 异常基类；StateError 表示调用时机错误，TransportError 表示传输失败，ProtocolError 表示响应不符合协议；无效参数可能抛 ValueError。播放器错误原样传播，例如 RuntimeError / OSError。

连接断开后由宿主决定是否执行 disconnect → check_registration → open_director → 重建播放任务。播放完成请求失败不会自动重试，避免重复释放下一段。

## 进阶：直接接收事件

使用 run_robot 时无需自行处理事件；需要完全自管播放时才调用 `await director.receive()`。返回值是 **dict，事件类型取 event['type']**。

| type | 常用字段 / 含义 |
| --- | --- |
| prompt | speech_id、text；新段开始 |
| text | speech_id、text；生成文本 |
| audio | speech_id、data、format、sample_rate、channels、frame_duration、sequence |
| audio_end | speech_id；接收结束，声音可能仍在播放 |
| interrupted | 旧 speech_id 作废；停止旧段 |
| audio_gap | missing；音频包缺失。run_robot 默认继续播放，可在 on_event 中抛异常停止 |
| audio_session_closed | 音频会话结束，设备可继续待命 |
| director_stopped | 导播已停止 |
| disconnected | 连接断开 |

完全自管时，实际播完调用 `await director.playback_finished(speech_id)`。只允许确认已收完且仍有效的一段，不重复确认。当前传输缺少可靠逐段结束标记，audio_end 使用尾部静默判断，强网络抖动仍可能影响音频完整性；播放适配层不改变该边界。
