Files

422 lines
23 KiB
Markdown
Raw Permalink Blame History

This file contains invisible Unicode characters
This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# 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`,解析设备系统参数并持久化到数据库 |
| 2 | **设备状态查询** | 提供 HTTP API 查询设备在线状态、详情、离线设备列表 |
| 3 | **离线检测告警** | 后台任务周期扫描,超过 `heartbeat.timeout` 未上报心跳的设备标记为离线并输出告警日志 |
| 4 | **视频流管理** | 心跳中状态为「正常」的通道自动通过 MediaMTX 拉取 RTSP 视频流(`StreamManager`) |
| 5 | **视频上传接口** | 盒子告警时通过 HTTP FORM-DATA 上传录制视频,返回唯一 `VideoId` |
| 6 | **告警上报接口** | 盒子违规事件 JSON 报文字动解析入库(`alarm_record` 表) |
| 7 | **告警查询接口** | 告警列表(分页/筛选)与单条告警详情查询 |
| 8 | **豆包 AI 分析** | 告警图片可交由 `doubao` 插件判定是否为误报(可配置开关) |
| 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`) |
---
## 架构与数据流
### 心跳上报链路
```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 注册表分发)
├── task_switch.py # IN2 数字输入控制任务启停(复用 diwatch 的 IN2 通道)
├── alarm_trim.py # 告警记录清理(超限裁剪 + 删除关联视频)
├── diwatch/ # DI 数字输入监听库(SE5/SE9 的 IN1/IN2,独立 CLI:python3 -m diwatch;task_switch 复用其 IN2 通道)
├── in1.py / in11.py # diwatch 前身:IN1 事件脚本 / IN2 轮询脚本(保留备用)
├── 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. **IP 音柱 TTS 播报**:按 `Summary` 匹配文案调用 `ip_sound_column.tts` 播放语音(失败仅记录日志,不阻断)
4. **写库**:按 `AlarmId/UniqueId` 去重写入 `alarm_record`,超限自动裁剪(`alarm_trim`)
5. **外部转发**:先上传视频替换 `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/`);
配置了 `[workers.config.database].url` 时优先走 PostgreSQL(当前 config.toml 即此模式),
由 `chongming_worker.Plugin` 基类统一初始化。
### `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 分析告警图片 |
| `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 数字输入控制任务启停(见下节) |
| `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 秒)未上报的设备标记为离线并输出告警日志。
---
## 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` 判定成败并记日志)。
### 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` 确认接线。
行为约定:
1. **只响应电平跳变**,不按插件启动时的初始电平动作(避免插件重启时误触发全量启停)。
2. DI 轮询跑在独立线程,HTTP 控制调度回事件循环并加锁串行,快速抖动不会并发穿插。
3. 找不到 IN2 状态文件(非盒子环境)时降级为不启动,仅输出警告,不影响插件其他功能。
---
## 开发指南
### 新增 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 颜色格式化器。