modify: 插件初始化

2026-08-12 10:24:32 +08:00
parent 3e1c568d04
commit 56dd4f0ebb
3 changed files with 516 additions and 91 deletions
@@ -8,7 +8,7 @@

 ```text
 <plugin_name>/
-├── __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 *` 或显式导入）
@@ -26,8 +26,8 @@
 ## 已有插件概览

 | 插件                     | 特点                                         | 适合参考   |
-| ------------------------ | ------------------------------------------------- | ---------- |
-| **example**              | 最简示例，`auto_bind_models` + Database + cleanup | 入门学习   |
+| ------------------------ | -------------------------------------------- | ---------- |
+| **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 <name> [operation ...]
+> ```
+>
+> 以下为手动创建方式（了解结构时参考）。
+
 ```bash
 mkdir -p src/plugins/<name>/router
 mkdir -p src/plugins/<name>/models
@@ -124,38 +132,44 @@ from . import my_action
 __all__ = ["my_action"]
 ```

-### 第七步：实现工厂函数
+### 第七步：实现插件入口（推荐 Plugin 基类）

-`src/plugins/<name>/__init__.py`：
+`src/plugins/<name>/__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("<name>")  # 自动绑定生成的 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")) / "<name>.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 = "<name>"
+    db_name = "<name>.db"   # 自动在 data_dir 下建 SQLite 库（不需要则置 None）
+    auto_bind = True        # 首次初始化时自动 auto_bind_models("<name>")
+
+    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<br/>self.db = ..."] -->|"self.set_var('db', self.db)"| GV["global_var 模块<br/>db / require()"]
+    R1["router/*.py<br/>@rpc_method 函数"] -->|"from ..global_var import require"| GV
+    R2["routes.py<br/>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/<name>/__init__.py`：

 ```python
-from ..global_var import db
+class MyPlugin(Plugin):
+    type = "<name>"
+    db_name = "<name>.db"

-@rpc_method(subject="<name>.action", method="GET", ...)
-async def handler(request):
+    async def on_init(self):
+        self.set_var("db", self.db)   # 发布到 global_var 模块
+```
+
+### 路由一：TOML 驱动标准路由（@rpc_method）
+
+`router/<op>.py`（请求经 Zenoh → Worker Queryable → 调用函数）：
+
+```python
+from chongming_worker.decorators import rpc_method
+from ..models.pydantic import OpGetRequest, OpGetResponse
+
+@rpc_method(subject="<name>.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="/<name>", tags=["<name>"])
+
+@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/<op>.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 装饰器
@@ -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 颜色格式化器。
@@ -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 基类统一管理生命周期

 关键约定:
  - 模型命名: <Operation><Method>{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