chongming/chongming-plugins
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
fb158d75cab2bb3808377884d8a62625bbf02a03
chongming-plugins/aibox_sophon_adapter/README.md
T

422 lines
23 KiB
Markdown
Raw Normal View History

modify: 插件初始化
2026-08-12 10:24:32 +08:00
# 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`
---
## 核心功能
| # | 功能 | 说明 |
| --- | -------------------- | ------------------------------------------------------------------------------------------ |
modify: IN2 DI 输入监听
2026-09-13 20:34:10 +08:00
| 1 | **MQTT 心跳上报** | 通过网关 MQTT Bridge 订阅 `box/aibox_sophon/board_ping`，解析设备系统参数并持久化到数据库 |
modify: 插件初始化
2026-08-12 10:24:32 +08:00
| 2 | **设备状态查询** | 提供 HTTP API 查询设备在线状态、详情、离线设备列表 |
| 3 | **离线检测告警** | 后台任务周期扫描，超过 `heartbeat.timeout` 未上报心跳的设备标记为离线并输出告警日志 |
| 4 | **视频流管理** | 心跳中状态为「正常」的通道自动通过 MediaMTX 拉取 RTSP 视频流（`StreamManager`） |
| 5 | **视频上传接口** | 盒子告警时通过 HTTP FORM-DATA 上传录制视频，返回唯一 `VideoId` |
| 6 | **告警上报接口** | 盒子违规事件 JSON 报文字动解析入库（`alarm_record` 表） |
| 7 | **告警查询接口** | 告警列表（分页/筛选）与单条告警详情查询 |
| 8 | **豆包 AI 分析** | 告警图片可交由 `doubao` 插件判定是否为误报（可配置开关） |
modify: IN2 DI 输入监听
2026-09-13 20:34:10 +08:00
| 9 | **IP 音柱 TTS 播报** | 告警时按违规类型调用 `ip_sound_column` 插件播放 TTS 语音（可配置开关/音量/文案） |
| 10 | **外部 MQTT 转发** | 心跳数据转发到外部 MQTT Broker（`mqtt_forward`） |
| 11 | **外部 MQTT 订阅** | 直接订阅外部 Broker，实现 `edge_app_controller` 请求-响应协议（`mqtt_subscribe`） |
| 12 | **告警转发外部平台** | 先上传视频再上报告警到外部开放平台（`alarm_forward`） |
| 13 | **IN2 任务启停开关** | 监听盒子 IN2 数字输入：有电启动全部算法任务，断电停止全部任务（`task_switch`） |
modify: 插件初始化
2026-08-12 10:24:32 +08:00
---
## 架构与数据流
### 心跳上报链路
```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/
modify: 统一插件风格
2026-08-14 16:04:39 +08:00
├── __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() 构建订阅路由
modify: 插件初始化
2026-08-12 10:24:32 +08:00
├── global_var.py # 模块级全局变量（db / stream_manager / mqtt_* / 各配置）
├── routes.py # 自定义 FastAPI 路由（video_upload / video / alarm_report）
├── edge_app_controller.py # 外部 MQTT 请求-响应处理器（command 注册表分发）
modify: IN2 DI 输入监听
2026-09-13 20:34:10 +08:00
├── task_switch.py # IN2 数字输入控制任务启停（复用 diwatch 的 IN2 通道）
modify: 插件初始化
2026-08-12 10:24:32 +08:00
├── alarm_trim.py # 告警记录清理（超限裁剪 + 删除关联视频）
modify: IN2 DI 输入监听
2026-09-13 20:34:10 +08:00
├── diwatch/ # DI 数字输入监听库（SE5/SE9 的 IN1/IN2，独立 CLI：python3 -m diwatch；task_switch 复用其 IN2 通道）
├── in1.py / in11.py # diwatch 前身：IN1 事件脚本 / IN2 轮询脚本（保留备用）
modify: 插件初始化
2026-08-12 10:24:32 +08:00
├── 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 + "忽略告警"）
modify: IN2 DI 输入监听
2026-09-13 20:34:10 +08:00
3. **IP 音柱 TTS 播报**：按 `Summary` 匹配文案调用 `ip_sound_column.tts` 播放语音（失败仅记录日志，不阻断）
4. **写库**：按 `AlarmId/UniqueId` 去重写入 `alarm_record`，超限自动裁剪（`alarm_trim`）
5. **外部转发**：先上传视频替换 `VideoFile` 为外部 `VideoId`，再上报告警
modify: 插件初始化
2026-08-12 10:24:32 +08:00
> 盒子仅当 **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` 函数。
---
modify: IN2 DI 输入监听
2026-09-13 20:34:10 +08:00
## 数据库表
modify: 插件初始化
2026-08-12 10:24:32 +08:00
modify: IN2 DI 输入监听
2026-09-13 20:34:10 +08:00
数据库默认为 SQLite 文件 `{data_dir}/aibox_sophon_adapter.db`（默认 `./data/`）；
配置了 `[workers.config.database].url` 时优先走 PostgreSQL（当前 config.toml 即此模式），
由 `chongming_worker.Plugin` 基类统一初始化。
modify: 插件初始化
2026-08-12 10:24:32 +08:00
### `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 分析告警图片 |
modify: IN2 DI 输入监听
2026-09-13 20:34:10 +08:00
| `ip_sound_column` | `enabled` / `volume` / `tts` | `true` / `50` / `{}` | 告警时 IP 音柱 TTS 播报（`tts` 为 Summary → 播报文案映射，未命中用默认文案） |
| `task_switch` | `enabled` / `api_base` / `start_command` / `stop_command` / `poll_interval` / `debounce` / `active_low` | true / `http://127.0.0.1` / 1 / 0 / 0.05 / 3 / false | IN2 数字输入控制任务启停（见下节） |
modify: 插件初始化
2026-08-12 10:24:32 +08:00
| `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 秒）未上报的设备标记为离线并输出告警日志。
---
modify: IN2 DI 输入监听
2026-09-13 20:34:10 +08:00
## IN2 任务启停开关（task_switch）
监听盒子 IN2 数字输入（diwatch `SysfsBackend` 轮询 `gpio_status`，含抗抖），
电平变化时通过盒子本地 HTTP 接口控制**全部**算法任务：
- IN2 **有电**（电平 1）→ 启动全部任务（`ControlCommand = start_command`，默认 1）
- IN2 **断电**（电平 0）→ 停止全部任务（`ControlCommand = stop_command`，默认 0）
控制流程：`POST {api_base}/api/alg_task_fetch`（查全部任务的 `AlgTaskSession`）→
逐个 `POST {api_base}/api/alg_task_control`（启停控制，按返回 `Result.Code` 判定成败并记日志）。
modify 修复前端展示配置与实际插件配置不一致的bug
2026-09-17 15:34:34 +08:00
### IN2 后端自适应（SE5 / SE9）
由 `diwatch.make_in2_backend` 自动选择，优先级：显式 `device` → sysfs 状态文件（SE9）→ gpio-keys event 设备（SE5）→ 放弃启动。
| 板卡 | 后端 | 说明 |
| ---- | ---- | ---- |
| SE9 | `SysfsBackend` | 轮询 `/sys/bus/platform/devices/gpio-in-2/gpio_status`，`poll_interval` / `debounce` 生效 |
| SE5 | `EventBackend` | 无 gpio-in-2 节点，IN2 与 IN1 同走 gpio-keys 事件设备（`event1`），键码 `key`（默认 `KEY_DOWN=108`） |
SE5 首次部署需确认两件事：
1. **event 读权限**：`linaro` 用户运行 diwatch 报 `Permission denied: '/dev/input/event1'` 时，
执行 `sudo usermod -aG input linaro` 并重新登录（服务以 root 运行则不受影响）。
2. **IN2 实际键码**：操作 IN2 的同时运行 `python3 -m diwatch detect -t 5`，输出里的 `code=` 即键码；
写入 `task_switch.key`（或 CLI `in2 -k <键码>`）。若 detect 无任何输出，说明该 SE5 的 IN2
未接入 gpio-keys，需查 `sudo cat /sys/kernel/debug/gpio` 确认接线。
modify: IN2 DI 输入监听
2026-09-13 20:34:10 +08:00
行为约定：
1. **只响应电平跳变**，不按插件启动时的初始电平动作（避免插件重启时误触发全量启停）。
2. DI 轮询跑在独立线程，HTTP 控制调度回事件循环并加锁串行，快速抖动不会并发穿插。
3. 找不到 IN2 状态文件（非盒子环境）时降级为不启动，仅输出警告，不影响插件其他功能。
---
modify: 插件初始化
2026-08-12 10:24:32 +08:00
## 开发指南
### 新增 RPC 端点
1. 在 `config/workers/aibox_sophon_adapter/` 下创建 `<操作>.toml`（参考 `list_devices.toml`）
modify: cli命令更新
2026-08-20 16:51:00 +08:00
2. 运行 `uv run gen_models` 生成模型
modify: 插件初始化
2026-08-12 10:24:32 +08:00
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` 后重启服务即可生效，无需重新生成模型
modify: 统一插件风格
2026-08-14 16:04:39 +08:00
（运行时配置均在 `AiboxSophonAdapterPlugin.on_init` 中读取；MQTT 客户端启动逻辑见 `mqtt_clients.py`）。
modify: 插件初始化
2026-08-12 10:24:32 +08:00
---
## 注意事项
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 颜色格式化器。
Reference in New Issue Copy Permalink