diff --git a/README.md b/README.md index 4b23a2a..261c2c8 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,7 @@ ```text / -├── __init__.py # 入口：create_workers() 工厂 + cleanup() + auto_bind_models() +├── __init__.py # 入口：Plugin 子类 + create_workers/cleanup 绑定 ├── global_var.py # 模块级全局状态（Database、API client 等） ├── router/ # RPC 端点实现 │ ├── __init__.py # 导入所有端点模块（`from . import *` 或显式导入） @@ -25,12 +25,12 @@ ## 已有插件概览 -| 插件 | 特点 | 适合参考 | -| ------------------------ | ------------------------------------------------- | ---------- | -| **example** | 最简示例，`auto_bind_models` + Database + cleanup | 入门学习 | -| **auth** | SQLModel ORM + 管理员初始化 + 多进程幂等 | 数据库操作 | -| **doubao** | 外部 API 客户端（Ark/OpenAI）+ AICC 安全通道 | 外部集成 | -| **aibox_sophon_adapter** | 自定义 `APIRouter` + MQTT 转发 + 后台任务 | 高级模式 | +| 插件 | 特点 | 适合参考 | +| ------------------------ | -------------------------------------------- | ---------- | +| **example** | 最简示例，Plugin 基类 + Database + 生命周期 | 入门学习 | +| **auth** | SQLModel ORM + 管理员初始化 + 多进程幂等 | 数据库操作 | +| **doubao** | 外部 API 客户端（Ark/OpenAI）+ AICC 安全通道 | 外部集成 | +| **aibox_sophon_adapter** | 自定义 `APIRouter` + MQTT 转发 + 后台任务 | 高级模式 | --- @@ -38,6 +38,14 @@ ### 第一步：创建目录结构 +> **推荐**：直接用脚手架生成统一风格的插件骨架，无需手动创建目录： +> +> ```bash +> uv run gen_plugin [operation ...] +> ``` +> +> 以下为手动创建方式（了解结构时参考）。 + ```bash mkdir -p src/plugins//router mkdir -p src/plugins//models @@ -124,38 +132,44 @@ from . import my_action __all__ = ["my_action"] ``` -### 第七步：实现工厂函数 +### 第七步：实现插件入口（推荐 Plugin 基类） -`src/plugins//__init__.py`： +`src/plugins//__init__.py`（由 `gen_plugin` 脚手架生成）： ```python -from pathlib import Path -from chongming_worker import auto_bind_models -from chongming_worker.runtime import create_workers as _create_workers -from .router import * # 触发 @rpc_method 注册 +import logging -auto_bind_models("") # 自动绑定生成的 Pydantic 模型 +from chongming_worker.plugin import Plugin -async def create_workers(session, config, count, worker_manager): - """插件入口：初始化资源，启动 Worker""" - from chongming_database import Database - from .global_var import set_db +from . import router # noqa: F401 触发 @rpc_method 注册 +# from .routes import router # 可选：自定义 FastAPI 路由（网关自动发现） - db_path = Path(config.get("data_dir", "./data")) / ".db" - db = Database(db_path) - await db.initialize() - set_db(db) +logger = logging.getLogger(__name__) - return _create_workers(session, config, count, worker_manager) -async def cleanup(worker_manager): - """清理资源""" - from .global_var import db - if db: - await db.dispose() +class MyPlugin(Plugin): + type = "" + db_name = ".db" # 自动在 data_dir 下建 SQLite 库（不需要则置 None） + auto_bind = True # 首次初始化时自动 auto_bind_models("") + + async def on_init(self): + """首次 create_workers 时调用一次：初始化资源、启动后台任务。""" + self.set_var("db", self.db) # 发布到 global_var 供业务代码读取 + # self.spawn_task(self._background_loop()) + + async def on_cleanup(self): + """cleanup 时调用：关闭自定义客户端等（后台任务与数据库由基类处理）。""" + + +# 暴露给 app lifespan 的模块级入口（bound method，签名与旧写法完全一致） +_plugin = MyPlugin() +create_workers = _plugin.create_workers +cleanup = _plugin.cleanup ``` -> **注意**：`create_workers` 必须是 `async` 函数，因为 `Database.initialize()` 是异步的。框架在 `asyncio.gather` 中并发调用所有插件的工厂函数。 +基类自动处理：幂等、统一日志、数据库建库/释放、后台任务取消、清理钩子、模型自动绑定。详见 `utils/worker/README.md` 的「Plugin 基类」章节。 + +> **兼容旧写法**：不依赖基类的传统 `create_workers` / `cleanup` 模块级函数写法仍受支持（见 `utils/worker/README.md`）。 --- @@ -199,32 +213,89 @@ from .routes import router # 网关通过 getattr 自动发现 --- -## 全局变量模式 +## 共享对象（global_var）在两种路由中的使用 -每个插件通过 `global_var.py` 管理模块级共享资源： +### 数据流 -```python -# global_var.py -from chongming_database import Database +共享对象（Database、API client 等）在 `Plugin.on_init` 中初始化，通过 `set_var` 发布到 `global_var` 模块；两种路由的业务代码（模块级函数）在函数体内 `from .global_var import ...` 读取。 -db: Database | None = None - -def set_db(database: Database): - global db - db = database +```mermaid +flowchart LR + P["Plugin.on_init
self.db = ..."] -->|"self.set_var('db', self.db)"| GV["global_var 模块
db / require()"] + R1["router/*.py
@rpc_method 函数"] -->|"from ..global_var import require"| GV + R2["routes.py
APIRouter 处理器"] -->|"from .global_var import db"| GV ``` -在端点中直接引用： +> **为什么需要 `global_var.py`**：`@rpc_method` 函数和 APIRouter 处理器都是模块级函数，拿不到 Plugin 实例，只能通过模块级容器读取；`global_var.py` 独立成模块，可避免与 `__init__.py` 循环导入。`set_var()` 内部会调用 `global_var.set_global_var()`（白名单校验）完成发布。 + +### 注入：on_init → global_var + +`src/plugins//__init__.py`： ```python -from ..global_var import db +class MyPlugin(Plugin): + type = "" + db_name = ".db" -@rpc_method(subject=".action", method="GET", ...) -async def handler(request): + async def on_init(self): + self.set_var("db", self.db) # 发布到 global_var 模块 +``` + +### 路由一：TOML 驱动标准路由（@rpc_method） + +`router/.py`（请求经 Zenoh → Worker Queryable → 调用函数）： + +```python +from chongming_worker.decorators import rpc_method +from ..models.pydantic import OpGetRequest, OpGetResponse + +@rpc_method(subject=".op", method="GET", response_model=OpGetResponse) +async def op(request: OpGetRequest) -> OpGetResponse: + from ..global_var import require + db = require("db") # 未初始化立即抛清晰错误 async with db.get_session() as session: ... + return OpGetResponse(...) ``` +### 路由二：插件自定义路由（APIRouter） + +`routes.py`（FastAPI 直接调用，适合 multipart 上传等 TOML 无法描述的端点）： + +```python +from fastapi import APIRouter, UploadFile, File + +router = APIRouter(prefix="/", tags=[""]) + +@router.post("/upload") +async def upload(file: UploadFile = File(...)): + from .global_var import db + if db is None: # 容忍未初始化，优雅降级 + return {"error": "db 未初始化"} + async with db.get_session() as session: + ... + return {"ok": True} +``` + +在 `__init__.py` 顶部暴露给网关：`from .routes import router`（网关通过 `getattr(module, "router")` 自动发现）。 + +### 两种路由对比 + +| | 路由一 `@rpc_method` | 路由二 `APIRouter` | +| ---------- | ----------------------------------------- | --------------------------------------------- | +| 文件位置 | `router/.py` | `routes.py` | +| 入口发现 | 装饰器注册 → Worker Queryable | `__init__.py` 暴露 `router` 属性 | +| 函数签名 | `(request: Pydantic模型) -> Pydantic模型` | FastAPI 风格（`Request`/`UploadFile`/`Form`） | +| 适用场景 | 标准 JSON / query 端点 | multipart 上传、手动解析报文 | +| 读共享对象 | `from ..global_var import require` | `from .global_var import db` | +| 请求路径 | HTTP → Zenoh → Worker 回调 | FastAPI 直接调用 | + +### 经验 + +- **在函数体内 import**（不要放文件顶部）：模块 import 时 `create_workers` 尚未执行、`global_var` 还是空；请求到达时变量才注入完成。 +- **两种读取风格**：`require("db")`（必须有，未初始化立即报错）vs `from .global_var import db` + `if db is None`（允许优雅降级）。 +- **参考实现**：`aibox_sophon_adapter/routes.py` 的 `_save_alarm_record`（自定义路由中 `from .global_var import db` + `async with db.get_session()`）。 + --- ## @rpc_method 装饰器 diff --git a/aibox_sophon_adapter/README.md b/aibox_sophon_adapter/README.md new file mode 100644 index 0000000..b5e36ce --- /dev/null +++ b/aibox_sophon_adapter/README.md @@ -0,0 +1,373 @@ +# 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 # 插件入口：create_workers() 工厂 + cleanup() +│ # - 初始化 DB / StreamManager / 各 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 python -m chongming.scripts.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` 后重启服务即可生效，无需重新生成模型 +（运行时配置均在 `create_workers` 中读取）。 + +--- + +## 注意事项 + +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 颜色格式化器。 diff --git a/example/__init__.py b/example/__init__.py index 8dea313..34af96a 100644 --- a/example/__init__.py +++ b/example/__init__.py @@ -1,13 +1,14 @@ """ -Example Worker 插件 — 演示 @rpc_method + 显式模型绑定用法 +Example Worker 插件 — 演示 @rpc_method + Plugin 基类统一生命周期 -本插件展示如何使用 chongming_worker 框架定义 RPC 端点。 +本插件展示如何使用 chongming_worker 框架定义 RPC 端点，以及如何 +基于 Plugin 基类编写风格统一的插件初始化 / 清理。 核心工作流: 1. 编写 config/workers/example/*.toml 配置文件 2. 执行 uv run gen_models 生成 Pydantic 模型 - 3. 在 __init__.py 中用 @rpc_method 定义业务函数 - 4. 显式导入模型并绑定到对应端点（替代 auto_bind_models） + 3. 在 router/*.py 中用 @rpc_method 定义业务函数 + 4. 在 __init__.py 中用 Plugin 基类统一管理生命周期 关键约定: - 模型命名: {Request|Response} @@ -15,54 +16,34 @@ Example Worker 插件 — 演示 @rpc_method + 显式模型绑定用法 - 装饰器 subject 中的操作名 → 模型前缀（驼峰） example.add → Add, example.subtract → Subtract """ -from pathlib import Path +import logging -from chongming_worker import auto_bind_models -from chongming_worker.runtime import create_workers as _create_workers +from chongming_worker.plugin import Plugin -from .router import * +from .router import * # noqa: F401,F403 触发 @rpc_method 注册 -auto_bind_models("example") +logger = logging.getLogger(__name__) -# ============================================================ -# 工厂函数 — 供 app lifespan 动态加载 -# ============================================================ -# 框架通过 __import__ + getattr 获取此函数， -# 然后调用 create_workers(session, config, count, worker_manager=worker_manager)。 -# 详见 src/chongming/app/__init__.py 的 lifespan。 +class ExamplePlugin(Plugin): + """Example 插件 — 统一生命周期。""" + + type = "example" + db_name = "example.db" # 自动在 data_dir 下建 SQLite 库 + auto_bind = True # 首次初始化时自动 auto_bind_models("example") + + async def on_init(self): + """首次 create_workers 时调用一次：初始化资源。""" + # 数据库已由基类自动初始化（self.db），写入全局变量供 router 使用 + self.set_var("db", self.db) + logger.info("[example] 初始化完成") + + async def on_cleanup(self): + """cleanup 时调用：关闭自定义客户端等（数据库由基类处理）。""" + logger.info("[example] 清理完成") -# 幂等标记：防止 create_workers 被重复调用时重复初始化数据库等资源 -_initialized = False - - -async def create_workers(session, config, count=1, worker_manager=None): - """插件入口 — 委托给 chongming_worker 运行时创建 Worker 协程""" - global _initialized - if _initialized: - print("[example] create_workers 重复调用，跳过重复初始化") - return _create_workers(session, config, count, worker_manager) - - from .global_var import set_global_var - from chongming_database import Database - - db_path = Path(config.get("data_dir", "./data")) / "example.db" - - db = Database(db_path) - await db.initialize() - set_global_var("db", db) - - _initialized = True - return _create_workers(session, config, count, worker_manager) - - -async def cleanup(worker_item: str): - global _initialized - from .global_var import db, set_global_var - - if db is not None: - await db.dispose() - set_global_var("db", None) - _initialized = False - print(f"{worker_item} 清理完毕") +# 暴露给 app lifespan 的模块级入口（bound method，签名与旧写法完全一致） +plugin = ExamplePlugin() +create_workers = plugin.create_workers +cleanup = plugin.cleanup