# aibox_sophon_adapter — AI Box Sophon 设备适配器

基于 **MQTT + Zenoh RPC** 的 AI Box（Sophon 算力盒子）设备接入适配器。
盒子通过 MQTT 主题 `/board_ping` 周期上报系统状态心跳，插件负责解析、持久化、对外提供
查询 API，并在设备离线时自动告警；同时支持告警视频上传、违规事件上报与外部平台转发。

> 插件类型：`aibox_sophon_adapter`　|　Zenoh subject 前缀：`aibox_sophon_adapter.*`　|　路由前缀：`/aibox_sophon_adapter`

---

## 核心功能

| #   | 功能                 | 说明                                                                                       |
| --- | -------------------- | ------------------------------------------------------------------------------------------ |
| 1   | **MQTT 心跳上报**    | 通过网关 MQTT Bridge 订阅 `box/aibox_sophon/board_ping`，解析设备系统参数并持久化到 SQLite |
| 2   | **设备状态查询**     | 提供 HTTP API 查询设备在线状态、详情、离线设备列表                                         |
| 3   | **离线检测告警**     | 后台任务周期扫描，超过 `heartbeat.timeout` 未上报心跳的设备标记为离线并输出告警日志        |
| 4   | **视频流管理**       | 心跳中状态为「正常」的通道自动通过 MediaMTX 拉取 RTSP 视频流（`StreamManager`）            |
| 5   | **视频上传接口**     | 盒子告警时通过 HTTP FORM-DATA 上传录制视频，返回唯一 `VideoId`                             |
| 6   | **告警上报接口**     | 盒子违规事件 JSON 报文字动解析入库（`alarm_record` 表）                                    |
| 7   | **告警查询接口**     | 告警列表（分页/筛选）与单条告警详情查询                                                    |
| 8   | **豆包 AI 分析**     | 告警图片可交由 `doubao` 插件判定是否为误报（可配置开关）                                   |
| 9   | **外部 MQTT 转发**   | 心跳数据转发到外部 MQTT Broker（`mqtt_forward`）                                           |
| 10  | **外部 MQTT 订阅**   | 直接订阅外部 Broker，实现 `edge_app_controller` 请求-响应协议（`mqtt_subscribe`）          |
| 11  | **告警转发外部平台** | 先上传视频再上报告警到外部开放平台（`alarm_forward`）                                      |

---

## 架构与数据流

### 心跳上报链路

```mermaid
sequenceDiagram
    participant Box as AI Box (Sophon)
    participant Broker as MQTT Broker (网关 Bridge)
    participant Zenoh as Zenoh RPC
    participant Plugin as aibox_sophon_adapter
    participant DB as SQLite
    participant MTX as MediaMTX

    Box->>Broker: publish box/aibox_sophon/board_ping
    Broker->>Zenoh: 路由为 aibox_sophon_adapter.board_ping
    Zenoh->>Plugin: 调用 board_ping(request)
    Plugin->>DB: upsert board_heartbeat 记录
    Plugin->>MTX: 状态正常通道 start_stream(RTSP)
    Plugin->>Broker: 心跳转发外部 MQTT（可选）
```

### 告警上报链路

```mermaid
sequenceDiagram
    participant Box as AI Box
    participant GW as 网关 HTTP
    participant Plugin as aibox_sophon_adapter
    participant DB as SQLite
    participant Doubao as doubao 插件
    participant Ext as 外部开放平台

    Box->>GW: POST /api/v1/aibox_sophon_adapter/video_upload (FORM-DATA)
    GW->>Plugin: 保存视频 → 返回 VideoId
    Box->>GW: POST /api/v1/aibox_sophon_adapter/alarm_report (JSON)
    GW->>Plugin: 手动 Pydantic 验证报文
    Plugin->>Doubao: 豆包 AI 分析（可关闭）
    Plugin->>DB: 写入 alarm_record（按 AlarmId 去重 + 裁剪）
    Plugin->>Ext: 上传视频 + 转发告警（可选）
```

### 外部 MQTT 请求-响应（edge_app_controller）

```mermaid
sequenceDiagram
    participant C as 外部客户端
    participant Broker as 外部 Broker
    participant Plugin as aibox_sophon_adapter

    C->>Broker: publish /edge_app_controller {"command": "..."}
    Broker->>Plugin: 订阅回调 edge_app_controller(payload, topic)
    Plugin->>Plugin: 按 command 分发到注册表 _COMMANDS
    Plugin-->>Broker: publish /edge_app_controller_reply {"code":0,"msg":"ok","data":{...}}
```

---

## 目录结构

```text
src/plugins/aibox_sophon_adapter/
├── __init__.py            # 插件入口：AiboxSophonAdapterPlugin（Plugin 基类）
│                          #   - on_init: DB / StreamManager / MQTT 客户端 / 全局变量 / 后台任务
│                          #   - on_cleanup: 流控制器与 MQTT 客户端关闭（db/后台任务由基类处理）
├── mqtt_clients.py        # MQTT 客户端能力组件（从 __init__ 抽取，统一"启动失败降级为 None"模式）
│                          #   - start_mqtt_forward / start_mqtt_subscribe / start_internal_mqtt
│                          #   - build_subscribe_routes() 构建订阅路由
├── global_var.py          # 模块级全局变量（db / stream_manager / mqtt_* / 各配置）
├── routes.py              # 自定义 FastAPI 路由（video_upload / video / alarm_report）
├── edge_app_controller.py # 外部 MQTT 请求-响应处理器（command 注册表分发）
├── alarm_trim.py          # 告警记录清理（超限裁剪 + 删除关联视频）
├── router/                # Zenoh RPC 端点（@rpc_method）
│   ├── __init__.py        # 导出全部端点模块
│   ├── board_ping.py      # MQTT 心跳上报
│   ├── list_devices.py    # 查询所有设备状态
│   ├── get_device.py      # 查询单个设备详情
│   ├── get_offline_devices.py  # 查询离线/异常设备
│   ├── list_alarms.py     # 查询告警记录列表
│   └── get_alarm.py       # 查询单条告警详情
└── models/                # 数据模型
    ├── board_models.py    # 心跳数据结构（手动维护，嵌套过深无法自动生成）
    ├── alarm_models.py    # 告警上报请求/响应模型
    ├── video_models.py    # 视频上传响应模型
    ├── orm.py             # SQLModel ORM（board_heartbeat / alarm_record）
    ├── pydantic.py        # gen_models 自动生成（如存在）
    └── __init__.py        # 模型导出
```

---

## 快速开始

### 1. 注册插件

插件已注册于 `config/workers/aibox_sophon_adapter/config.toml`（新架构由框架自动扫描）：

```toml
[workers]
type = "aibox_sophon_adapter"
module = "plugins.aibox_sophon_adapter"
factory = "create_workers"
cleanup = "cleanup"
count = 1
```

### 2. 启动服务

```bash
uv run serve
```

启动时插件会自动：初始化 SQLite → 连接 MediaMTX → 启动 MQTT 转发/订阅/内部客户端 →
创建 Worker 副本 → 启动离线检查后台任务。

### 3. 验证

```bash
# 查看设备列表
curl "http://127.0.0.1:8000/api/v1/aibox_sophon_adapter/list_devices"
```

---

## HTTP API 接口

### 自动生成 RPC 端点（前缀 `/api/v1/aibox_sophon_adapter`）

| 方法 | 路径                   | 说明                  | 参数                                                           |
| ---- | ---------------------- | --------------------- | -------------------------------------------------------------- |
| GET  | `/list_devices`        | 查询所有设备状态      | 无                                                             |
| GET  | `/get_device`          | 查询单个设备详情      | `board_id: str`                                                |
| GET  | `/get_offline_devices` | 查询离线/异常设备列表 | 无                                                             |
| GET  | `/list_alarms`         | 告警列表（分页/筛选） | `board_id: str`、`summary: str`、`page: int`、`page_size: int` |
| GET  | `/get_alarm`           | 查询单条告警详情      | `alarm_id: str`                                                |

> `board_ping` 为 MQTT 端点，不产生 HTTP 路由。

### 自定义 FastAPI 路由（`routes.py`，同样挂在 `/api/v1` 下）

| 方法 | 路径                | 说明                                                                        |
| ---- | ------------------- | --------------------------------------------------------------------------- |
| POST | `/video_upload`     | 视频上传（multipart/form-data），返回 `{"Result": {...}, "VideoId": "..."}` |
| GET  | `/video/{video_id}` | 视频下载/流式播放（`Content-Disposition: inline`）                          |
| POST | `/alarm_report`     | 告警上报（JSON），返回 `{"Result": {"Code": 0, "Desc": ""}}`                |

#### `video_upload` 表单字段

| 字段          | 必填 | 说明                   |
| ------------- | ---- | ---------------------- |
| `BoardIp`     | 是   | 盒子网络地址           |
| `BoardId`     | 是   | 盒子唯一标识           |
| `TaskSession` | 是   | 所属任务标识           |
| `GBDeviceId`  | 否   | 国标设备号（无则空串） |
| `GBTaskChnId` | 否   | 国标通道号（无则空串） |
| `Video`       | 是   | 视频文件               |

存储路径约定：`{upload_root}/aibox_sophon_adapter/{video_id}/{filename}`（流式写入 + MD5 校验）。

#### `alarm_report` 处理流程

1. 手动 Pydantic 验证基础字段（`alarm_models.AlarmReportRequest`，失败记录完整请求体并返回 422）
2. **豆包 AI 分析**：判定为 `False` 时忽略告警（返回 Code 0 + "忽略告警"）
3. **写库**：按 `AlarmId/UniqueId` 去重写入 `alarm_record`，超限自动裁剪（`alarm_trim`）
4. **外部转发**：先上传视频替换 `VideoFile` 为外部 `VideoId`，再上报告警

> 盒子仅当 **HTTP 200 且 `Result.Code` 非 0** 时才认为上报失败。

---

## MQTT 主题

### 网关 MQTT Bridge 订阅（`[[workers.mqtt_topics]]`）

| 主题                          | 目标 Zenoh subject                | 说明         |
| ----------------------------- | --------------------------------- | ------------ |
| `box/aibox_sophon/board_ping` | `aibox_sophon_adapter.board_ping` | 盒子心跳上报 |

> 盒子只设根主题 `box`（非 `chongming` 保留命名空间，避免被网关 Bridge 按 Worker RPC 拦截）；
> 子路径固定 `aibox_sophon/...`。

### 外部 MQTT 订阅（`[workers.config.mqtt_subscribe]`，插件直连外部 Broker）

| 主题                   | 处理函数              | 回复主题                     |
| ---------------------- | --------------------- | ---------------------------- |
| `/edge_app_controller` | `edge_app_controller` | `/edge_app_controller_reply` |

### 内部 MQTT 请求-响应（`[workers.config.internal_mqtt]`）

`edge_app_controller` 的 `alg_media_fetch` / `alg_task_fetch` 命令通过内部客户端向
任意应用进程发起请求-响应：

- 请求主题：`box/aibox_sophon/edge_app_controller`
- 回复主题：`box/aibox_sophon/edge_app_controller_reply`

> 切勿使用 `chongming/{插件}/{操作}` —— 网关 Bridge 订阅 `chongming/+/+` 会拦截 3 级主题并按
> Worker RPC 路由，而 `edge_app_controller` 不是 `@rpc_method`，会导致
> `worker_manager` 报 "Worker ... 无回复，尝试重试"。

---

## edge_app_controller 请求-响应协议

订阅 `/edge_app_controller`，回复发布到 `/edge_app_controller_reply`。

**请求**（JSON）：`{"command": "<命令名>", ...业务字段}`；兼容 `{"Event": "/alg_media_fetch", ...}`
（`Event` 取最后一段作为命令）。

**回复**（JSON）：`{"code": 0, "msg": "ok", "data": {...}}`

| code | 含义              |
| ---- | ----------------- |
| 0    | 成功              |
| -1   | 缺少 command 字段 |
| -2   | 未知命令          |
| -3   | 命令执行异常      |

**内置命令注册表（`_COMMANDS`）**：

| command           | 说明                                     | 是否回复   |
| ----------------- | ---------------------------------------- | ---------- |
| `ping`            | 存活探测，返回服务信息                   | 是         |
| `ack`             | 确认命令，仅记录日志                     | 否（静默） |
| `query_status`    | 复用 `list_devices` 返回设备在线状态列表 | 是         |
| `alg_media_fetch` | 按 `BoardId` 返回最近心跳的通道/媒体信息 | 是         |
| `alg_task_fetch`  | 按 `BoardId` 返回最近心跳的算法任务信息  | 是         |

> 命令处理函数返回 `None` 表示只处理、不回复（静默命令）。新增命令只需在 `_COMMANDS`
> 注册表登记一个 `async handler(payload) -> data` 函数。

---

## 数据库表（SQLite）

数据库文件：`{data_dir}/aibox_sophon_adapter.db`（默认 `./data/aibox_sophon_adapter.db`）。

### `board_heartbeat` — 设备心跳记录

| 字段                                         | 说明                               |
| -------------------------------------------- | ---------------------------------- |
| `board_id`                                   | 设备唯一标识（唯一索引）           |
| `board_ip` / `board_platform` / `board_type` | IP / 硬件类型 / 接入方式（WAN/4G） |
| `board_temp`                                 | 芯片温度                           |
| `status` / `version` / `system_info`         | 状态 / 软件版本 / 系统版本         |
| `grant_code` / `grant_desc`                  | 授权码 / 授权描述                  |
| `last_ping_time` / `is_online`               | 最后心跳时间 / 是否在线            |
| `ping_data`                                  | 完整心跳数据 JSON                  |

### `alarm_record` — 告警记录

| 字段                                         | 说明                                            |
| -------------------------------------------- | ----------------------------------------------- |
| `alarm_id`                                   | 告警唯一标识（唯一索引）                        |
| `board_id` / `board_ip`                      | 设备标识                                        |
| `summary`                                    | 违规类型（NoHelmet / Intrude / PeopleCross 等） |
| `task_session` / `media_name` / `video_file` | 任务号 / 通道名 / 关联视频标识                  |
| `alarm_time` / `alarm_timestamp`             | 告警时间 / 时间戳（微秒）                       |
| `alarm_data`                                 | 完整告警报文 JSON                               |

---

## 配置详解

配置文件：`config/workers/aibox_sophon_adapter/config.toml`

### 插件注册（`[workers]` / `[worker]` / `[worker.registration]`）

```toml
[workers]
type = "aibox_sophon_adapter"
module = "plugins.aibox_sophon_adapter"
factory = "create_workers"
cleanup = "cleanup"
count = 1                 # Worker 副本数

[worker]
defaults = { method = "GET", timeout = 30, auth_required = false, cacheable = false, internal = false }

[worker.registration]
router_prefix = "/aibox_sophon_adapter"
tags = ["aibox_sophon_adapter", "sophon", "adapter"]
```

### 运行时配置（`[workers.config]`）

| 配置节            | 字段                                                   | 默认值                              | 说明                           |
| ----------------- | ------------------------------------------------------ | ----------------------------------- | ------------------------------ |
| （根）            | `data_dir`                                             | `./data`                            | 数据目录（数据库 + 上传文件）  |
| （根）            | `mediamtx_addr`                                        | `rtsp://127.0.0.1/channel`          | MediaMTX 地址                  |
| （根）            | `link_path`                                            | `/data/bm-app`                      | 视频流软链接路径               |
| `heartbeat`       | `interval`                                             | `5`                                 | 设备上报间隔（秒）             |
| `heartbeat`       | `timeout`                                              | `15`                                | 超时阈值（秒），超过即视为离线 |
| `heartbeat`       | `check_interval`                                       | `10`                                | 服务端检查离线设备的间隔（秒） |
| `mqtt_forward`    | `enabled` / `uri` / `topic`                            | false / 空 / `/board_ping`          | 心跳转发到外部 Broker          |
| `mqtt_subscribe`  | `enabled` / `uri` / `topics`                           | false / 空 / 空                     | 直连外部 Broker 订阅（见上）   |
| `internal_mqtt`   | `enabled` / `uri` / `request_topic` / `response_topic` | false / `mqtt://127.0.0.1:1883` / … | 内部请求-响应客户端            |
| `alarm_retention` | `max_records`                                          | `0`                                 | 最多保留告警记录数（0 不限制） |
| `doubao_analysis` | `enabled`                                              | `true`                              | 是否启用豆包 AI 分析告警图片   |
| `alarm_forward`   | `enabled` / `url` / `video_upload_url` / `timeout`     | false / … / … / 10.0                | 告警转发外部开放平台           |

> 示例：`mqtt_forward.uri = "mqtt://admin:Mixc%402021@192.168.1.6:1883"` —— 密码中特殊字符
> 需 URL 编码（`%40` 即 `@`）。

---

## 心跳与离线机制

- 设备按 `heartbeat.interval`（默认 5 秒）通过 `box/aibox_sophon/board_ping` 上报心跳。
- `board_ping` 处理器：新设备建记录 / 已有设备更新心跳并置为在线；状态「正常」的通道
  自动通过 MediaMTX 启动 RTSP 拉流；可选转发外部 MQTT。
- 后台任务 `_check_offline_devices` 每 `heartbeat.check_interval`（默认 10 秒）扫描一次，
  将超过 `heartbeat.timeout`（默认 15 秒）未上报的设备标记为离线并输出告警日志。

---

## 开发指南

### 新增 RPC 端点

1. 在 `config/workers/aibox_sophon_adapter/` 下创建 `<操作>.toml`（参考 `list_devices.toml`）
2. 运行 `uv run gen_models` 生成模型
3. 在 `src/plugins/aibox_sophon_adapter/router/` 下创建 `<操作>.py`，使用 `@rpc_method`
4. 更新 `router/__init__.py` 的导入与 `__all__`
5. 重启服务

### 新增 edge_app_controller 命令

在 `edge_app_controller.py` 中定义 `async def _cmd_xxx(payload) -> data`，并在 `_COMMANDS`
注册表中登记即可。返回 `None` 表示静默处理（不回复）。

### 修改配置

编辑 `config/workers/aibox_sophon_adapter/config.toml` 后重启服务即可生效，无需重新生成模型
（运行时配置均在 `AiboxSophonAdapterPlugin.on_init` 中读取；MQTT 客户端启动逻辑见 `mqtt_clients.py`）。

---

## 注意事项

1. **不要使用 `chongming/` 命名空间**：盒子 MQTT 根主题固定为 `box`，子路径固定
   `aibox_sophon/...`，否则会被网关 Bridge 拦截并按 Worker RPC 路由。
2. **`edge_app_controller` 不是 `@rpc_method`**：它只通过外部 MQTT 订阅触发，Worker 上
   无对应 Zenoh Queryable。
3. **数据库模型**（`models/orm.py`）与心跳结构（`models/board_models.py`）为手动维护，
   `board_models.py` 因嵌套过深无法由 `gen_models` 自动生成。
4. **视频文件清理**：告警记录裁剪时会同步删除 `{upload_root}/aibox_sophon_adapter/{video_id}/`
   目录下的关联视频。
5. **日志**：使用标准 `logging.getLogger(__name__)`，核心已配置 ANSI 颜色格式化器。