Files

T

2026-09-13 20:34:10 +08:00

22 KiB

Raw Blame History

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`）

架构与数据流

心跳上报链路

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（可选）

告警上报链路

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）

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":{...}}

目录结构

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（新架构由框架自动扫描）：

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

2. 启动服务

uv run serve

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

3. 验证

# 查看设备列表
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` 处理流程

手动 Pydantic 验证基础字段（alarm_models.AlarmReportRequest，失败记录完整请求体并返回 422）
豆包 AI 分析：判定为 False 时忽略告警（返回 Code 0 + "忽略告警"）
IP 音柱 TTS 播报：按 Summary 匹配文案调用 ip_sound_column.tts 播放语音（失败仅记录日志，不阻断）
写库：按 AlarmId/UniqueId 去重写入 alarm_record，超限自动裁剪（alarm_trim）
外部转发：先上传视频替换 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]`）

[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 判定成败并记日志）。

行为约定：

只响应电平跳变，不按插件启动时的初始电平动作（避免插件重启时误触发全量启停）。
DI 轮询跑在独立线程，HTTP 控制调度回事件循环并加锁串行，快速抖动不会并发穿插。
找不到 IN2 状态文件（非盒子环境）时降级为不启动，仅输出警告，不影响插件其他功能。

开发指南

新增 RPC 端点

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

新增 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）。

注意事项

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

22 KiB Raw Blame History Unescape Escape