From 79eb7fe5cca38c880b008311cc17363cb4c5b601 Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Mon, 27 Jul 2026 14:28:07 +0800 Subject: [PATCH] modify: aibox_sophon_adapter plugins --- README.md | 222 +++++++++++++++++- aibox_sophon_adapter/__init__.py | 149 ++++++++++++ aibox_sophon_adapter/global_var.py | 36 +++ aibox_sophon_adapter/models/__init__.py | 45 ++++ aibox_sophon_adapter/models/board_models.py | 133 +++++++++++ aibox_sophon_adapter/models/orm.py | 31 +++ aibox_sophon_adapter/models/pydantic.py | 34 +++ aibox_sophon_adapter/router/__init__.py | 14 ++ aibox_sophon_adapter/router/board_ping.py | 94 ++++++++ aibox_sophon_adapter/router/get_device.py | 63 +++++ .../router/get_offline_devices.py | 54 +++++ aibox_sophon_adapter/router/list_devices.py | 69 ++++++ auth/README.md | 193 +++++++++++++++ doubao/README.md | 83 +++++-- doubao/__init__.py | 39 ++- doubao/global_var.py | 37 ++- doubao/router/analyze.py | 12 +- example/README.md | 67 +++++- example/models/__init__.py | 7 +- example/models/pydantic.py | 11 +- example/router/__init__.py | 4 + example/router/add_mqtt.py | 24 ++ example/router/heartbeat.py | 20 ++ 23 files changed, 1380 insertions(+), 61 deletions(-) create mode 100644 aibox_sophon_adapter/__init__.py create mode 100644 aibox_sophon_adapter/global_var.py create mode 100644 aibox_sophon_adapter/models/__init__.py create mode 100644 aibox_sophon_adapter/models/board_models.py create mode 100644 aibox_sophon_adapter/models/orm.py create mode 100644 aibox_sophon_adapter/models/pydantic.py create mode 100644 aibox_sophon_adapter/router/__init__.py create mode 100644 aibox_sophon_adapter/router/board_ping.py create mode 100644 aibox_sophon_adapter/router/get_device.py create mode 100644 aibox_sophon_adapter/router/get_offline_devices.py create mode 100644 aibox_sophon_adapter/router/list_devices.py create mode 100644 auth/README.md create mode 100644 example/router/add_mqtt.py create mode 100644 example/router/heartbeat.py diff --git a/README.md b/README.md index e887f14..54687c1 100644 --- a/README.md +++ b/README.md @@ -1,19 +1,217 @@ -# 插件目录 +# Worker 插件开发指南 -每个子目录代表一个 Worker 插件，包含以下约定结构： +每个子目录代表一个独立的 Worker 插件，封装特定领域的业务逻辑。插件通过 Zenoh RPC 对外提供服务， +HTTP 路由由框架根据 TOML 配置自动生成。 + +--- + +## 目录结构约定 ```text plugin_name/ -├── init.py # 提供 create_workers 和 cleanup 工厂函数 -├── router/ # RPC 方法实现（使用 @rpc_method） -├── models/ # 由 gen_models 生成的 Pydantic 模型（自动生成，勿手动修改） -├── global_var.py # 全局状态管理（如数据库连接） -└── README.md # 插件说明 +├── __init__.py # 入口：create_workers() 工厂 + cleanup() +├── global_var.py # 模块级全局状态（数据库连接、API 客户端等） +├── router/ # RPC 端点实现 +│ ├── __init__.py # 导入所有端点模块，__all__ 导出 +│ └── <操作名>.py # 每个端点一个文件，使用 @rpc_method 装饰器 +├── models/ # 数据模型 +│ ├── __init__.py # 模型导出（自动生成，勿手动修改） +│ ├── pydantic.py # Pydantic 请求/响应模型（由 gen_models.py 自动生成） +│ └── <表名>.py # SQLModel ORM 模型（如需要数据库） +└── README.md # 插件说明文档 ``` -## 开发新插件 +--- -1. 在 `config/workers/` 下创建同名目录，编写路由 TOML -2. 运行 `uv run gen_models config/workers/ src/plugins/` -3. 实现 `router/*.py` 中的业务逻辑 -4. 在 `__init__.py` 中实现 `create_workers` 工厂函数 +## 开发新插件：完整流程 + +### 第一步：创建目录结构 + +```bash +mkdir -p src/plugins//router +mkdir -p src/plugins//models +touch src/plugins//__init__.py +touch src/plugins//global_var.py +touch src/plugins//router/__init__.py +``` + +### 第二步：注册插件 + +在 `config/workers/config.toml` 中添加： + +```toml +[[workers]] +type = "" +module = "plugins." +factory = "create_workers" +cleanup = "cleanup" +count = 3 + +[workers.config] +type = "" +# 自定义配置项... +``` + +### 第三步：编写端点 TOML 配置 + +创建 `config/workers//config.toml`（分组配置）： + +```toml +[worker] +defaults = { timeout = 3.0, auth_required = true, cacheable = false, internal = false } + +[worker.registration] +router_prefix = "/" +``` + +创建每个操作的 `.toml` 文件，如 `config/workers//my_action.toml`： + +```toml +[POST] +summary = "我的操作" +auth_required = true + +[POST.response_model] +result = { type = "str", required = true } +message = { type = "str", default = "" } +``` + +### 第四步：生成 Pydantic 模型 + +```bash +uv run python -m chongming.scripts.gen_models +``` + +这会自动在 `src/plugins//models/` 下生成 `pydantic.py` 和更新 `__init__.py`。 + +### 第五步：实现 RPC 端点 + +`src/plugins//router/my_action.py`： + +```python +from chongming_worker.decorators import rpc_method +from ..models.pydantic import MyActionPostRequest, MyActionPostResponse + +@rpc_method( + subject=".my_action", + method="POST", + response_model=MyActionPostResponse, +) +async def my_action(request: MyActionPostRequest) -> MyActionPostResponse: + """业务逻辑实现""" + result = await do_something(request) + return MyActionPostResponse(result="success", message="done") +``` + +### 第六步：更新 router 导出 + +`src/plugins//router/__init__.py`： + +```python +from . import my_action +__all__ = ["my_action"] +``` + +### 第七步：实现工厂函数 + +`src/plugins//__init__.py`： + +```python +from chongming_worker.runtime import create_workers as _create_workers + +def create_workers(session, config, count, worker_manager): + """插件入口：初始化全局资源，启动 Worker""" + from . import router # 触发 @rpc_method 注册 + from .global_var import set_db + + # 初始化数据库、客户端等 + from chongming_database import Database + db = Database(config["data_dir"] / "mydb.db") + # await db.initialize() -- 在异步环境中 + + set_db(db) + _create_workers(session, config, count, worker_manager) + +async def cleanup(worker_manager): + """清理资源""" + from .global_var import db + if db: + await db.dispose() +``` + +--- + +## 为现有插件添加新端点 + +1. 在 `config/workers//` 下创建 `<新操作>.toml` +2. 运行 `uv run python -m chongming.scripts.gen_models` +3. 在 `router/` 下创建 `<新操作>.py`，引用生成的模型 +4. 更新 `router/__init__.py` +5. 重启服务 + +--- + +## 全局变量模式 + +每个插件通过 `global_var.py` 管理模块级共享资源，避免跨请求的状态冲突： + +```python +# global_var.py +db: "Database | None" = None + +def set_db(database: "Database"): + global db + db = database +``` + +在端点中直接引用： + +```python +from ..global_var import db + +@rpc_method(subject="plugin.action", method="GET", ...) +async def handler(request: RequestModel): + async with db.get_session() as session: + # 执行数据库操作 + ... +``` + +--- + +## 装饰器说明 + +```python +@rpc_method( + subject=".", # Zenoh 主题（必需） + method="GET", # HTTP 方法：GET 或 POST + response_model=ResponseCls, # Pydantic 响应模型类 +) +async def handler(request: RequestCls) -> ResponseCls: + ... +``` + +- **GET 端点**：参数通过 URL query string 传递，在 TOML 中用 `params` 定义 +- **POST 端点**：参数通过 JSON body 传递，自动绑定到 Pydantic 请求模型 +- **internal 端点**：设置 `internal = true` 后，不注册 HTTP 路由，仅可通过 Zenoh RPC 内部调用 + +--- + +## 命名约定 + +| 概念 | 约定 | 示例 | +| ----------------- | --------------- | ----------------------------------- | +| 插件名 | 小写下划线 | `doubao`, `auth`, `example` | +| 操作名 | 小写下划线 | `analyze`, `add`, `change_password` | +| Zenoh subject | `<插件>.<操作>` | `example.add`, `doubao.analyze` | +| TOML 文件名 | 与操作名一致 | `analyze.toml`, `add.toml` | +| Python 路由文件名 | 与操作名一致 | `analyze.py`, `add.py` | + +--- + +## 现有插件 + +| 插件 | 说明 | 文档 | +| ---------------------------- | -------------------------------------- | ---------------- | +| [auth](auth/README.md) | 用户注册、登录、JWT 鉴权、用户信息管理 | 含 SQLite 数据库 | +| [doubao](doubao/README.md) | 火山引擎豆包视觉理解模型集成 | 第三方 API 调用 | +| [example](example/README.md) | 四则运算示例 + 访问计数器 | 入门参考 | diff --git a/aibox_sophon_adapter/__init__.py b/aibox_sophon_adapter/__init__.py new file mode 100644 index 0000000..03857ba --- /dev/null +++ b/aibox_sophon_adapter/__init__.py @@ -0,0 +1,149 @@ +""" +AI Box Sophon Adapter 插件 + +基于 MQTT 的 AI Box 设备心跳适配器，实现： + 1. 通过 MQTT 主题 /board_ping 接收设备心跳上报 + 2. 解析设备系统参数（温度、硬盘、内存、通道、任务、TPU 等） + 3. 持久化心跳数据到 SQLite 数据库 + 4. 提供 HTTP API 查询设备在线状态和详情 + 5. 后台定时检查离线设备，触发异常告警 + +心跳超时机制: + - 设备按配置间隔（默认 5 秒）上报心跳 + - 服务端每 check_interval 秒检查一次 + - 超过 heartbeat_timeout 秒未收到心跳的设备标记为离线 +""" +import asyncio +import logging +from datetime import datetime +from pathlib import Path + +from chongming_worker import auto_bind_models +from chongming_worker.runtime import create_workers as _create_workers + +from .router import * + +logger = logging.getLogger(__name__) + +# 自动绑定 Pydantic 模型（用于 gen_models 生成的模型） +auto_bind_models("aibox_sophon_adapter") + +# 离线检查任务引用 +_offline_check_task: asyncio.Task | None = None + + +async def _check_offline_devices(interval: float): + """ + 后台任务：定期检查离线设备 + + 每隔 interval 秒扫描数据库，将超过 heartbeat_timeout 秒 + 未收到心跳的设备标记为离线，并输出告警日志。 + """ + from .global_var import db, heartbeat_timeout + + while True: + await asyncio.sleep(interval) + try: + if db is None: + continue + + async with db.get_session() as session: + from .models.orm import BoardHeartbeat + from sqlalchemy import select + + stmt = select(BoardHeartbeat).where(BoardHeartbeat.is_online == True) + result = await session.execute(stmt) + records = result.scalars().all() + + now = datetime.now() + offline_count = 0 + for record in records: + elapsed = (now - record.last_ping_time).total_seconds() + if elapsed > heartbeat_timeout: + record.is_online = False + record.updated_at = now + offline_count += 1 + logger.warning( + f"[Offline] 设备 {record.board_id} (IP: {record.board_ip}) " + f"已离线 {elapsed:.0f} 秒（超时阈值: {heartbeat_timeout}s）" + ) + + if offline_count > 0: + await session.commit() + logger.info(f"[Offline] 本轮标记 {offline_count} 个设备为离线") + + except Exception as e: + logger.error(f"[OfflineChecker] 检查异常: {e}") + + +async def create_workers(session, config, count=1, worker_manager=None): + """ + 插件入口 — 初始化数据库和全局资源，创建 Worker 副本 + + 由 FastAPI lifespan 在启动时调用。 + """ + from .global_var import set_global_var, heartbeat_timeout as _default_timeout + from chongming_database import Database + + # 读取心跳配置 + worker_config = config.get("worker", {}).get("heartbeat", {}) if isinstance(config, dict) else {} + hb_interval = worker_config.get("interval", 5) + hb_timeout = worker_config.get("timeout", _default_timeout) + check_interval = worker_config.get("check_interval", 10) + + # 设定全局变量 + global heartbeat_timeout + heartbeat_timeout = hb_timeout + + # 初始化数据库 + data_dir = Path(config.get("data_dir", "./data")) if isinstance(config, dict) else Path("./data") + db_path = data_dir / "storage" / "chongming" / "aibox_sophon_adapter.db" + db_path.parent.mkdir(parents=True, exist_ok=True) + + db = Database(db_path) + await db.initialize() + + # 创建表 + from .models.orm import BoardHeartbeat + async with db.get_session() as sess: + from sqlmodel import SQLModel + await sess.run_sync(lambda s: SQLModel.metadata.create_all(s.bind)) + + set_global_var("db", db) + set_global_var("heartbeat_timeout", hb_timeout) + set_global_var("heartbeat_interval", hb_interval) + + logger.info( + f"[aibox_sophon_adapter] 已初始化: " + f"心跳间隔={hb_interval}s, 超时={hb_timeout}s, 检查间隔={check_interval}s" + ) + + # 启动离线设备检查后台任务 + global _offline_check_task + _offline_check_task = asyncio.create_task(_check_offline_devices(check_interval)) + + # 创建 Worker 副本 + return _create_workers(session, config, count, worker_manager) + + +async def cleanup(worker_item: str): + """ + 清理资源 + + 在 Worker 关闭时调用，关闭数据库连接和后台任务。 + """ + from .global_var import db + + global _offline_check_task + if _offline_check_task is not None and not _offline_check_task.done(): + _offline_check_task.cancel() + try: + await _offline_check_task + except asyncio.CancelledError: + pass + logger.info("[aibox_sophon_adapter] 离线检查任务已停止") + + if db is not None: + await db.dispose() + + logger.info(f"[aibox_sophon_adapter] {worker_item} 清理完毕") diff --git a/aibox_sophon_adapter/global_var.py b/aibox_sophon_adapter/global_var.py new file mode 100644 index 0000000..9f195da --- /dev/null +++ b/aibox_sophon_adapter/global_var.py @@ -0,0 +1,36 @@ +""" +AI Box Sophon Adapter — 全局变量 + +模块级共享资源管理。 +""" +from typing import Any + +from chongming_database import Database + +# 数据库实例 +db: Database | None = None + +# 心跳超时配置（秒） +heartbeat_timeout: float = 15.0 + +# 心跳上报间隔（秒） +heartbeat_interval: float = 5.0 + + +def set_global_var(name: str, value: Any) -> None: + """ + 设置当前模块的全局变量。 + + Args: + name: 变量名（字符串） + value: 任意值 + """ + globals()[name] = value + + +__all__ = [ + "db", + "heartbeat_timeout", + "heartbeat_interval", + "set_global_var", +] diff --git a/aibox_sophon_adapter/models/__init__.py b/aibox_sophon_adapter/models/__init__.py new file mode 100644 index 0000000..2b0478a --- /dev/null +++ b/aibox_sophon_adapter/models/__init__.py @@ -0,0 +1,45 @@ +""" +模型导出 — aibox_sophon_adapter + +包含 gen_models 自动生成的模型和手动定义的 board_models。 +""" +from .pydantic import ( + AiboxSophonAdapterResponse, + GetDeviceGetRequest, + GetDeviceGetResponse, + GetOfflineDevicesGetResponse, + ListDevicesGetResponse, +) +from .board_models import ( + BoardPingRequest, + HostDiskInfo, + MediaInfo, + MediaStatusInfo, + AlgItemInfo, + AlgTaskStatusInfo, + TaskInfo, + TpuInfo, + DeviceSummary, +) +from .orm import BoardHeartbeat + +__all__ = [ + # 自动生成 + "AiboxSophonAdapterResponse", + "GetDeviceGetRequest", + "GetDeviceGetResponse", + "GetOfflineDevicesGetResponse", + "ListDevicesGetResponse", + # 手动定义 + "BoardPingRequest", + "HostDiskInfo", + "MediaInfo", + "MediaStatusInfo", + "AlgItemInfo", + "AlgTaskStatusInfo", + "TaskInfo", + "TpuInfo", + "DeviceSummary", + # ORM + "BoardHeartbeat", +] diff --git a/aibox_sophon_adapter/models/board_models.py b/aibox_sophon_adapter/models/board_models.py new file mode 100644 index 0000000..dc58375 --- /dev/null +++ b/aibox_sophon_adapter/models/board_models.py @@ -0,0 +1,133 @@ +""" +AI Box Sophon Adapter — 设备心跳数据 Pydantic 模型（手动定义） + +定义 BoardPingRequest 及其嵌套子结构。 +这些模型无法通过 gen_models 自动生成（嵌套层次过深），需要手动维护。 + +对应 MQTT 主题 /board_ping 的上报数据结构。 + +命名约定: + - Pydantic 模型类名以 Info 结尾（如 HostDiskInfo），避免与 JSON 字段名冲突 + - 字段名直接对应 JSON key，无需 alias +""" +from typing import Optional +from pydantic import BaseModel, Field + + +# ============================================================ +# 心跳数据子结构 +# ============================================================ + +class HostDiskInfo(BaseModel): + """设备硬盘使用情况""" + Available: int = Field(..., description="可用空间 (kB)") + Total: int = Field(..., description="总空间 (kB)") + Used: int = Field(..., description="已用空间 (kB)") + + +class MediaStatusInfo(BaseModel): + """通道状态描述""" + label: str = Field(..., description="状态标签，如 '正常'") + type: int = Field(..., description="状态类型码") + + +class MediaInfo(BaseModel): + """设备通道信息""" + MediaName: str = Field(..., description="通道名称") + MediaStatus: MediaStatusInfo = Field(..., description="通道状态") + MediaUrl: str = Field(..., description="通道视频流地址") + SubId: str = Field(default="", description="关联 GB28181 通道号") + + +class AlgItemInfo(BaseModel): + """算法配置信息""" + lineDesc: str = Field(default="", description="线描述") + lineRequired: bool = Field(default=False, description="是否需要线") + majorId: int = Field(..., description="主 ID") + majorOnly: bool = Field(default=False, description="仅主算法") + minorId: int = Field(..., description="副 ID") + name: str = Field(..., description="算法名称") + zoneDesc: str = Field(default="", description="区域描述") + zoneRequired: bool = Field(default=False, description="是否需要区域") + + +class AlgTaskStatusInfo(BaseModel): + """任务状态描述""" + label: str = Field(..., description="状态标签，如 '运行中'") + type: int = Field(..., description="状态类型码") + + +class TaskInfo(BaseModel): + """设备任务信息""" + AlgItem: list[AlgItemInfo] = Field(default_factory=list, description="算法配置列表") + AlgTaskSession: str = Field(..., description="任务号") + AlgTaskStatus: AlgTaskStatusInfo = Field(..., description="任务状态") + + +class TpuInfo(BaseModel): + """TPU 算力资源使用情况""" + mem_total: int = Field(..., description="内存总量") + mem_usage: int = Field(..., description="内存使用量") + tpu_usage: int = Field(..., description="TPU 使用率") + + +# ============================================================ +# 心跳请求体（MQTT 上报的完整数据结构） +# ============================================================ + +class BoardPingRequest(BaseModel): + """ + AI Box 设备心跳上报数据 + + 设备通过 MQTT 主题 /board_ping 定期上报。 + 上报间隔默认 5 秒。 + + 包含字段：设备标识、IP、温度、硬盘、内存、GPS、 + 通道列表、任务列表、TPU 算力等。 + """ + BoardId: str = Field(..., description="当前盒子唯一标识") + BoardIp: str = Field(..., description="设备 IP 地址") + BoardPlatform: str = Field(..., description="当前设备硬件类型，如 'BM-BOX'") + BoardTemp: str = Field(..., description="当前设备芯片温度，如 '44 (C)'") + BoardType: str = Field(..., description="当前设备接入方式 WAN/4G") + GBClientId: str = Field(default="", description="GB28181 设备号，配置后透传") + GrantCode: int = Field(..., description="授权码") + GrantDesc: str = Field(..., description="授权描述，如 '已授权'") + HostDisk: HostDiskInfo = Field(..., description="当前设备硬盘情况") + HostMemory: list[float] = Field(..., description="当前设备内存使用 [已使用(kB), 总利用率]") + Key: str = Field(..., description="运行状态关键字，如 '运行中'") + Latitude: float = Field(default=0.0, description="GPS 纬度") + Longitude: float = Field(default=0.0, description="GPS 经度") + Malloc: int = Field(..., description="当前应用 VmRSS 信息 (kB)") + Medias: list[MediaInfo] = Field(default_factory=list, description="当前设备通道信息") + Pid: int = Field(..., description="本次应用 PID") + Se6ip: str = Field(default="", description="特定设备时透传主控端 IP") + Status: str = Field(..., description="设备状态，如 '在线'") + System: str = Field(..., description="当前设备系统版本") + Tasks: list[TaskInfo] = Field(default_factory=list, description="当前设备任务信息") + Time: int = Field(..., description="当前时间戳（毫秒）") + Tpu: list[TpuInfo] = Field(default_factory=list, description="当前设备算力资源使用情况") + Version: str = Field(..., description="软件版本") + angleCourse: float = Field(default=0.0, description="GPS 角度") + kSpeed: float = Field(default=0.0, description="GPS 速度 (km/h)") + nSpeed: float = Field(default=0.0, description="GPS 速度 (m/s)") + + +# ============================================================ +# 设备状态摘要（API 响应用） +# ============================================================ + +class DeviceSummary(BaseModel): + """设备状态摘要（用于列表 API 响应）""" + board_id: str = Field(..., description="设备唯一标识") + board_ip: str = Field(..., description="设备 IP") + board_platform: str = Field(..., description="硬件类型") + board_type: str = Field(..., description="接入方式") + board_temp: str = Field(..., description="芯片温度") + status: str = Field(..., description="设备状态") + version: str = Field(..., description="软件版本") + is_online: bool = Field(..., description="是否在线") + last_ping_time: Optional[str] = Field(default=None, description="最后心跳时间") + grant_code: int = Field(..., description="授权码") + grant_desc: str = Field(..., description="授权描述") + diff --git a/aibox_sophon_adapter/models/orm.py b/aibox_sophon_adapter/models/orm.py new file mode 100644 index 0000000..dd59506 --- /dev/null +++ b/aibox_sophon_adapter/models/orm.py @@ -0,0 +1,31 @@ +""" +AI Box Sophon Adapter — ORM 模型 + +SQLModel 定义设备心跳记录表。 +""" +from datetime import datetime +from typing import Optional + +from sqlmodel import Field, SQLModel + + +class BoardHeartbeat(SQLModel, table=True): + """设备心跳记录表""" + __tablename__ = "board_heartbeat" + + id: Optional[int] = Field(default=None, primary_key=True) + board_id: str = Field(index=True, unique=True, description="设备唯一标识 (BoardId)") + board_ip: str = Field(default="", description="设备 IP") + board_platform: str = Field(default="", description="硬件类型") + board_temp: str = Field(default="", description="芯片温度") + board_type: str = Field(default="", description="接入方式 WAN/4G") + status: str = Field(default="未知", description="设备状态") + version: str = Field(default="", description="软件版本") + system_info: str = Field(default="", description="系统版本信息") + grant_code: int = Field(default=0, description="授权码") + grant_desc: str = Field(default="", description="授权描述") + last_ping_time: datetime = Field(default_factory=datetime.now, description="最后心跳时间") + ping_data: str = Field(default="{}", description="完整心跳数据 JSON") + is_online: bool = Field(default=True, description="是否在线") + created_at: datetime = Field(default_factory=datetime.now, description="首次上报时间") + updated_at: datetime = Field(default_factory=datetime.now, description="最后更新时间") diff --git a/aibox_sophon_adapter/models/pydantic.py b/aibox_sophon_adapter/models/pydantic.py new file mode 100644 index 0000000..50dfd38 --- /dev/null +++ b/aibox_sophon_adapter/models/pydantic.py @@ -0,0 +1,34 @@ +""" +自动生成的 Pydantic 模型 — aibox_sophon_adapter +来源: config/workers/aibox_sophon_adapter/ +""" + +from __future__ import annotations + +from typing import Any, Optional +from pydantic import BaseModel, Field + +class GetDeviceGetRequest(BaseModel): + """查询单个设备详情 — 请求体""" + board_id: str = Field(...) + +class GetDeviceGetResponse(BaseModel): + """查询单个设备详情 — 响应体""" + board_id: str = Field(...) + is_online: bool = Field(...) + last_ping_time: Optional[str] = Field(default=None) + ping_data: Optional[dict[str, Any]] = Field(default=None) + +class GetOfflineDevicesGetResponse(BaseModel): + """查询离线/异常设备列表 — 响应体""" + offline_devices: list[dict] = Field(...) + total: int = Field(...) + +class ListDevicesGetResponse(BaseModel): + """查询所有设备状态 — 响应体""" + devices: list[dict] = Field(...) + total: int = Field(...) + + +# 所有响应模型的联合类型 +AiboxSophonAdapterResponse = ListDevicesGetResponse | GetOfflineDevicesGetResponse | GetDeviceGetResponse diff --git a/aibox_sophon_adapter/router/__init__.py b/aibox_sophon_adapter/router/__init__.py new file mode 100644 index 0000000..d8cd927 --- /dev/null +++ b/aibox_sophon_adapter/router/__init__.py @@ -0,0 +1,14 @@ +""" +路由导出 — aibox_sophon_adapter +""" +from .board_ping import board_ping +from .list_devices import list_devices +from .get_device import get_device +from .get_offline_devices import get_offline_devices + +__all__ = [ + "board_ping", + "list_devices", + "get_device", + "get_offline_devices", +] diff --git a/aibox_sophon_adapter/router/board_ping.py b/aibox_sophon_adapter/router/board_ping.py new file mode 100644 index 0000000..f51b557 --- /dev/null +++ b/aibox_sophon_adapter/router/board_ping.py @@ -0,0 +1,94 @@ +""" +AI Box Sophon Adapter — MQTT 心跳上报处理器 + +监听 MQTT 主题 /board_ping（通过 MQTT Bridge 路由到 chongming/aibox_sophon_adapter/board_ping）， +解析设备心跳数据并更新数据库记录。 +""" +import json +import logging +from datetime import datetime + +from chongming_worker import rpc_method +from ..models.board_models import BoardPingRequest + +logger = logging.getLogger(__name__) + + +@rpc_method( + subject="aibox_sophon_adapter.board_ping", + method="MQTT", + summary="AI Box 设备心跳上报", + request_model=BoardPingRequest, +) +async def board_ping(request: BoardPingRequest, worker_id: str = "") -> dict: + """ + 处理设备心跳上报 + + 设备通过 MQTT 主题 /board_ping 定期（默认 5 秒）上报系统状态。 + 收到心跳后更新数据库中的设备记录，标记为在线。 + """ + from ..global_var import db, heartbeat_timeout + + board_id = request.BoardId + + try: + async with db.get_session() as session: + from ..models.orm import BoardHeartbeat + from sqlalchemy import select + + stmt = select(BoardHeartbeat).where(BoardHeartbeat.board_id == board_id) + result = await session.execute(stmt) + record: BoardHeartbeat | None = result.scalar_one_or_none() + + now = datetime.now() + ping_json = request.model_dump_json() + + if record is None: + # 新设备，创建记录 + record = BoardHeartbeat( + board_id=board_id, + board_ip=request.BoardIp, + board_platform=request.BoardPlatform, + board_temp=request.BoardTemp, + board_type=request.BoardType, + status=request.Status, + version=request.Version, + system_info=request.System, + grant_code=request.GrantCode, + grant_desc=request.GrantDesc, + last_ping_time=now, + ping_data=ping_json, + is_online=True, + ) + session.add(record) + logger.info(f"[Heartbeat] 新设备上线: {board_id} (IP: {request.BoardIp})") + else: + # 已有设备，更新心跳 + record.board_ip = request.BoardIp + record.board_platform = request.BoardPlatform + record.board_temp = request.BoardTemp + record.board_type = request.BoardType + record.status = request.Status + record.version = request.Version + record.system_info = request.System + record.grant_code = request.GrantCode + record.grant_desc = request.GrantDesc + record.last_ping_time = now + record.ping_data = ping_json + record.is_online = True + record.updated_at = now + + await session.commit() + + logger.debug( + f"[Heartbeat] {board_id}: temp={request.BoardTemp}, " + f"disk={request.HostDisk.Used}/{request.HostDisk.Total} kB, " + f"mem={request.HostMemory[0]:.1f} kB ({request.HostMemory[1]*100:.1f}%), " + f"medias={len(request.Medias)}, tasks={len(request.Tasks)}" + ) + + except Exception as e: + logger.error(f"[Heartbeat] 处理心跳失败 (board_id={board_id}): {e}") + return {"error": str(e)} + + return {"received": True, "board_id": board_id} diff --git a/aibox_sophon_adapter/router/get_device.py b/aibox_sophon_adapter/router/get_device.py new file mode 100644 index 0000000..5179a4f --- /dev/null +++ b/aibox_sophon_adapter/router/get_device.py @@ -0,0 +1,63 @@ +""" +查询单个设备详情 — GET /aibox/get_device?board_id=xxx +""" +import json +import logging +from datetime import datetime + +from chongming_worker import rpc_method + +logger = logging.getLogger(__name__) + + +@rpc_method( + subject="aibox_sophon_adapter.get_device", + method="GET", + summary="查询单个设备详情", +) +async def get_device(board_id: str, worker_id: str = "") -> dict: + """根据 BoardId 查询设备的完整心跳数据和在线状态""" + from ..global_var import db, heartbeat_timeout + + try: + async with db.get_session() as session: + from ..models.orm import BoardHeartbeat + from sqlalchemy import select + + stmt = select(BoardHeartbeat).where(BoardHeartbeat.board_id == board_id) + result = await session.execute(stmt) + record: BoardHeartbeat | None = result.scalar_one_or_none() + + if record is None: + return {"error": f"设备不存在: {board_id}"} + + now = datetime.now() + elapsed = (now - record.last_ping_time).total_seconds() + is_online = elapsed <= heartbeat_timeout + + ping_data = {} + try: + ping_data = json.loads(record.ping_data) if record.ping_data else {} + except json.JSONDecodeError: + pass + + return { + "board_id": record.board_id, + "board_ip": record.board_ip, + "board_platform": record.board_platform, + "board_type": record.board_type, + "board_temp": record.board_temp, + "status": record.status, + "version": record.version, + "system_info": record.system_info, + "grant_code": record.grant_code, + "grant_desc": record.grant_desc, + "is_online": is_online, + "last_ping_time": record.last_ping_time.isoformat() if record.last_ping_time else None, + "elapsed_seconds": round(elapsed, 1), + "ping_data": ping_data, + } + + except Exception as e: + logger.error(f"[get_device] 查询失败 (board_id={board_id}): {e}") + return {"error": str(e)} diff --git a/aibox_sophon_adapter/router/get_offline_devices.py b/aibox_sophon_adapter/router/get_offline_devices.py new file mode 100644 index 0000000..2abb18c --- /dev/null +++ b/aibox_sophon_adapter/router/get_offline_devices.py @@ -0,0 +1,54 @@ +""" +查询离线/异常设备 — GET /aibox/get_offline_devices +""" +import json +import logging +from datetime import datetime + +from chongming_worker import rpc_method + +logger = logging.getLogger(__name__) + + +@rpc_method( + subject="aibox_sophon_adapter.get_offline_devices", + method="GET", + summary="查询离线/异常设备列表", +) +async def get_offline_devices(worker_id: str = "") -> dict: + """返回所有超过心跳超时时间未上报的设备列表""" + from ..global_var import db, heartbeat_timeout + + offline_devices = [] + try: + async with db.get_session() as session: + from ..models.orm import BoardHeartbeat + from sqlalchemy import select + + stmt = select(BoardHeartbeat).order_by(BoardHeartbeat.last_ping_time.desc()) + result = await session.execute(stmt) + records = result.scalars().all() + + now = datetime.now() + for record in records: + elapsed = (now - record.last_ping_time).total_seconds() + is_online = elapsed <= heartbeat_timeout + + if not is_online: + offline_devices.append({ + "board_id": record.board_id, + "board_ip": record.board_ip, + "board_platform": record.board_platform, + "board_type": record.board_type, + "status": record.status, + "version": record.version, + "last_ping_time": record.last_ping_time.isoformat() if record.last_ping_time else None, + "elapsed_seconds": round(elapsed, 1), + "is_online": False, + }) + + except Exception as e: + logger.error(f"[get_offline_devices] 查询失败: {e}") + return {"offline_devices": [], "total": 0, "error": str(e)} + + return {"offline_devices": offline_devices, "total": len(offline_devices)} diff --git a/aibox_sophon_adapter/router/list_devices.py b/aibox_sophon_adapter/router/list_devices.py new file mode 100644 index 0000000..fee48fa --- /dev/null +++ b/aibox_sophon_adapter/router/list_devices.py @@ -0,0 +1,69 @@ +""" +查询所有设备状态 — GET /aibox/list_devices +""" +import json +import logging +from datetime import datetime + +from chongming_worker import rpc_method + +logger = logging.getLogger(__name__) + + +@rpc_method( + subject="aibox_sophon_adapter.list_devices", + method="GET", + summary="查询所有设备状态", +) +async def list_devices(worker_id: str = "") -> dict: + """返回所有已上报过心跳的设备及其在线状态""" + from ..global_var import db, heartbeat_timeout + + devices = [] + try: + async with db.get_session() as session: + from ..models.orm import BoardHeartbeat + from sqlalchemy import select + + stmt = select(BoardHeartbeat).order_by(BoardHeartbeat.updated_at.desc()) + result = await session.execute(stmt) + records = result.scalars().all() + + now = datetime.now() + for record in records: + # 检查是否超时 + elapsed = (now - record.last_ping_time).total_seconds() + is_online = elapsed <= heartbeat_timeout + + ping_data = {} + try: + ping_data = json.loads(record.ping_data) if record.ping_data else {} + except json.JSONDecodeError: + pass + + devices.append({ + "board_id": record.board_id, + "board_ip": record.board_ip, + "board_platform": record.board_platform, + "board_type": record.board_type, + "board_temp": record.board_temp, + "status": record.status, + "version": record.version, + "system_info": record.system_info, + "grant_code": record.grant_code, + "grant_desc": record.grant_desc, + "is_online": is_online, + "last_ping_time": record.last_ping_time.isoformat() if record.last_ping_time else None, + "elapsed_seconds": round(elapsed, 1), + "host_disk": ping_data.get("HostDisk", {}), + "host_memory": ping_data.get("HostMemory", []), + "medias_count": len(ping_data.get("Medias", [])), + "tasks_count": len(ping_data.get("Tasks", [])), + "tpu": ping_data.get("Tpu", []), + }) + + except Exception as e: + logger.error(f"[list_devices] 查询失败: {e}") + return {"devices": [], "total": 0, "error": str(e)} + + return {"devices": devices, "total": len(devices)} diff --git a/auth/README.md b/auth/README.md new file mode 100644 index 0000000..c6d7d0d --- /dev/null +++ b/auth/README.md @@ -0,0 +1,193 @@ +# Auth Worker 插件 + +用户认证与授权插件，提供完整的用户管理系统。 + +--- + +## 功能 + +- **用户注册**：用户名 + 邮箱 + 密码注册，自动校验唯一性 +- **用户登录**：JWT Token 签发，有效期 3 小时 +- **获取用户信息**：根据 JWT Token 获取完整用户档案 +- **修改密码**：旧密码验证 + 新密码更新 +- **更新资料**：更新全名、邮箱、头像、手机号等信息 +- **自动初始化**：首次启动自动创建 `admin` 管理员账户 + +--- + +## 数据库 + +使用 SQLite（通过 `chongming-database` 异步封装），数据库文件位于 `data/storage/chongming/auth.db`。 + +### Users 表结构 + +| 字段 | 类型 | 说明 | +| ----------------- | ----------------- | --------------------------------- | +| `id` | `str` (PK) | UUID 格式主键 | +| `username` | `str(50)` UNIQUE | 登录用户名 | +| `email` | `str(255)` UNIQUE | 电子邮箱 | +| `hashed_password` | `str(255)` | Argon2id 加密密码 | +| `full_name` | `str(100)` | 用户全名（可选） | +| `avatar_url` | `str(500)` | 头像链接（可选） | +| `phone` | `str(20)` | 手机号码（可选） | +| `is_active` | `bool` | 账户启用状态（默认 `true`） | +| `is_verified` | `bool` | 邮箱/手机验证状态（默认 `false`） | +| `is_superuser` | `bool` | 超级管理员标识（默认 `false`） | +| `created_at` | `datetime` | 创建时间 | +| `updated_at` | `datetime` | 最后更新时间 | +| `last_login_at` | `datetime` | 最后登录时间（可选） | + +密码使用 Rust 实现的 **Argon2id** 哈希，内存成本 19456 KiB，迭代 2 次，并行度 1。 + +--- + +## API 端点 + +所有端点路由前缀为 `/auth`。 + +### `POST /auth/register` — 用户注册 + +无需认证。 + +```json +// 请求 +{ + "username": "john", + "password": "secret123", + "email": "john@example.com", + "full_name": "John Doe", + "phone": "13800138000" +} + +// 响应 +{ + "result": "success", + "message": "注册成功", + "user_id": "a1b2c3d4_12345_1720000000.123", + "timestamp": 1720000000.123 +} +``` + +### `POST /auth/login` — 登录 + +无需认证。 + +```json +// 请求 +{ + "username": "admin", + "password": "admin" +} + +// 响应 +{ + "result": "success", + "token": "eyJhbGciOiJIUzI1NiIs...", + "message": "登录成功", + "timestamp": 1720000000.123 +} +``` + +JWT Token 有效期 3 小时，载荷包含：`sub`（用户ID）、`username`、`email`、`full_name`、`is_superuser` 等。 + +### `GET /auth/me?user_id=` — 获取用户信息 + +需要 JWT 鉴权（`Authorization: Bearer `）。 + +```json +// 响应 +{ + "result": "success", + "id": "a1b2c3d4_12345_1720000000.123", + "username": "john", + "email": "john@example.com", + "full_name": "John Doe", + "avatar_url": null, + "phone": "13800138000", + "is_active": true, + "is_verified": false, + "is_superuser": false, + "created_at": "2024-01-01T00:00:00+00:00", + "updated_at": "2024-01-01T00:00:00+00:00", + "last_login_at": null +} +``` + +### `POST /auth/change_password` — 修改密码 + +需要 JWT 鉴权。 + +```json +// 请求 +{ + "user_id": "a1b2c3d4_12345_1720000000.123", + "old_password": "secret123", + "new_password": "newSecret456" +} + +// 响应 +{ + "result": "success", + "message": "密码修改成功", + "timestamp": 1720000000.123 +} +``` + +### `POST /auth/update_profile` — 更新用户资料 + +需要 JWT 鉴权。仅更新非空字段。 + +```json +// 请求 +{ + "user_id": "a1b2c3d4_12345_1720000000.123", + "full_name": "John Updated", + "email": "", + "avatar_url": "https://example.com/avatar.png", + "phone": "" +} + +// 响应 +{ + "result": "success", + "message": "资料更新成功", + "timestamp": 1720000000.123 +} +``` + +--- + +## 默认管理员账户 + +首次启动时自动创建： + +| 字段 | 值 | +| ------------ | --------------------- | +| **用户名** | `admin` | +| **密码** | `admin` | +| **邮箱** | `admin@example.com` | +| **超管权限** | `is_superuser = true` | + +> ⚠️ **生产环境务必修改默认密码！** + +--- + +## 安全说明 + +- 密码使用 **Argon2id**（Rust 实现），对抗 GPU/ASIC 暴力破解 +- JWT 使用 **HS256** 算法，密钥在 `config.toml` 的 `[security]` 节配置 +- JWT 有效期 3 小时，过期后需重新登录 +- 支持用户软禁用（`is_active = false`） +- 接口级鉴权控制：每个端点可通过 TOML 的 `auth_required` 独立配置 + +--- + +## 依赖 + +| 工具包 | 用途 | +| -------------------- | ------------------------------------ | +| `chongming-database` | 异步 SQLite 数据库操作 | +| `chongming-jwt` | Rust JWT 编码/验证 | +| `chongming-password` | Rust Argon2id 密码哈希 | +| `chongming-config` | 全局配置加载 | +| `chongming-worker` | `@rpc_method` 装饰器 + Worker 运行时 | diff --git a/doubao/README.md b/doubao/README.md index 8a2e7da..5ec4e59 100644 --- a/doubao/README.md +++ b/doubao/README.md @@ -1,38 +1,79 @@ # Doubao Worker 插件 -火山引擎豆包（Doubao）视觉理解模型集成，支持图像分析。 +火山引擎豆包（Doubao）视觉理解模型集成，支持图像分析与描述生成。 + +--- ## 功能 -- 接收 base64 编码的图片和提示词 -- 调用火山引擎豆包视觉模型（支持思考模式） -- 返回模型描述结果 +- 接收 **base64 编码**的图片和**提示词**（prompt） +- 调用火山引擎豆包视觉模型 API +- 支持**思考模式**（thinking mode），模型会先推理再输出结果 +- 返回模型生成的图像描述文本 + +--- ## 配置 -在 `config/workers/doubao/config.toml` 中设置： +在 `config/workers/doubao/config.toml` 和 `config/workers/config.toml` 中设置： -- `api_key`：火山引擎 API Key（也可通过环境变量 `VOLC_ACCESSKEY` 提供） -- `base_url`：API 端点（默认 `https://ark.cn-beijing.volces.com/api/v3`） -- `model`：模型名称（如 `doubao-seed-evolving`） - -## 调用示例 - -```http -POST /api/v1/doubao/analyze -Content-Type: application/json - -{ - "image_base64": "data:image/png;base64,iVBORw0KG...", - "prompt": "描述这个图片的内容" -} +```toml +[workers.config] +type = "doubao" +api_key = "" # 火山引擎 API Key（也可通过环境变量 VOLC_ACCESSKEY 提供） +base_url = "https://ark.cn-beijing.volces.com/api/v3" # API 端点 +model = "doubao-seed-evolving" # 模型名称 ``` -响应示例： +--- + +## API 端点 + +### `POST /doubao/analyze` — 图像分析 ```json +// 请求 { - "result": "图片中是一只猫坐在沙发上...", + "image_base64": "data:image/png;base64,iVBORw0KGgoAAAANS...", + "prompt": "描述这张图片的内容" +} + +// 响应 +{ + "result": "图片中是一只橘猫坐在米色沙发上，阳光从窗户照进来...", "error": "" } ``` + +| 参数 | 类型 | 必填 | 说明 | +| -------------- | ----- | ---- | ----------------------------------------------------------- | +| `image_base64` | `str` | ✅ | Base64 编码的图片数据（可含 `data:image/...;base64,` 前缀） | +| `prompt` | `str` | ✅ | 给模型的提示词，描述期望的分析内容 | + +--- + +## 支持的模型 + +| 模型 | 说明 | +| ---------------------- | ---------------------------- | +| `doubao-seed-evolving` | 豆包视觉理解旗舰模型（默认） | + +可根据火山引擎最新文档更换 `model` 配置项。 + +--- + +## 实现细节 + +- 使用 `httpx` 异步 HTTP 客户端调用火山引擎 API +- 支持流式响应（streaming）中的思考模式 token +- API Key 优先从环境变量 `VOLC_ACCESSKEY` 读取 +- 超时时间默认 30 秒，在 `config.toml` 的 `defaults.timeout` 中配置 + +--- + +## 依赖 + +| 工具包 | 用途 | +| ------------------ | ------------------------------------ | +| `httpx` | 异步 HTTP 客户端 | +| `chongming-worker` | `@rpc_method` 装饰器 + Worker 运行时 | diff --git a/doubao/__init__.py b/doubao/__init__.py index a833cc4..620fbfb 100644 --- a/doubao/__init__.py +++ b/doubao/__init__.py @@ -2,10 +2,14 @@ 豆包视觉理解 Worker 插件 """ import os +from openai import AsyncOpenAI from volcenginesdkarkruntime import AsyncArk from chongming_worker import create_workers as _create_workers from chongming_worker import auto_bind_models -from .global_var import set_ark_client, cleanup_ark_client +from .global_var import ( + set_ark_client, set_aicc_client, set_is_aicc, + cleanup_all, +) from .router import * # 自动绑定 gen_models 生成的 Pydantic 模型 @@ -15,25 +19,42 @@ async def create_workers(session, config, count=1, worker_manager=None): """ 工厂函数：由 app 生命周期调用，创建 Worker 协程。 """ - # 从配置或环境变量获取 API Key + is_aicc = config.get("is_aicc", False) + set_is_aicc(is_aicc) + api_key = config.get("api_key") or os.environ.get("VOLC_ACCESSKEY") if not api_key: raise ValueError("缺少 API_KEY，请在配置中设置 api_key 或环境变量 VOLC_ACCESSKEY") - base_url = config.get("base_url", "https://ark.cn-beijing.volces.com/api/v3") model = config.get("model", "doubao-seed-evolving") - - # 将模型名存入环境变量，供 router 使用 os.environ["DOUBAO_MODEL"] = model - # 创建全局客户端 - client = AsyncArk(base_url=base_url, api_key=api_key) - set_ark_client(client) + if is_aicc: + # AICC 模式：懒加载，避免非 AICC 环境导入失败 + from bytedance.volcengine_aicc_sdk.secure_http_client import ( + AICCConfig, AsyncSecureHttpClient, + ) + + base_url = config.get("base_url", "https://ark.cn-beijing.volces.com/api/v3") + os.environ["DOUBAO_BASE_URL"] = base_url + + aicc_config = AICCConfig(ra_url=base_url, auth_token=api_key) + client = AsyncOpenAI( + api_key=api_key, + base_url=base_url, + http_client=AsyncSecureHttpClient(aicc_config=aicc_config), + ) + set_aicc_client(client) + else: + # Ark 模式：使用 AsyncArk + base_url = config.get("base_url", "https://ark.cn-beijing.volces.com/api/v3") + client = AsyncArk(base_url=base_url, api_key=api_key) + set_ark_client(client) # 委托给通用运行时创建 Worker 任务 return _create_workers(session, config, count, worker_manager) async def cleanup(worker_item: str): """清理资源""" - cleanup_ark_client() + cleanup_all() print(f"{worker_item} 清理完毕") \ No newline at end of file diff --git a/doubao/global_var.py b/doubao/global_var.py index ad7a56e..4a9aadb 100644 --- a/doubao/global_var.py +++ b/doubao/global_var.py @@ -1,7 +1,10 @@ -from typing import Optional +from typing import Optional, Union from volcenginesdkarkruntime import AsyncArk +from openai import AsyncOpenAI _ark_client: Optional[AsyncArk] = None +_aicc_client: Optional[AsyncOpenAI] = None +_is_aicc: bool = False def set_ark_client(client: AsyncArk): global _ark_client @@ -10,8 +13,36 @@ def set_ark_client(client: AsyncArk): def get_ark_client() -> Optional[AsyncArk]: return _ark_client +def set_aicc_client(client: AsyncOpenAI): + global _aicc_client + _aicc_client = client + +def get_aicc_client() -> Optional[AsyncOpenAI]: + return _aicc_client + +def set_is_aicc(value: bool): + global _is_aicc + _is_aicc = value + +def get_is_aicc() -> bool: + return _is_aicc + +def get_client() -> Union[AsyncArk, AsyncOpenAI, None]: + """返回当前活跃的客户端（AICC 或 Ark）""" + if _is_aicc: + return _aicc_client + return _ark_client + def cleanup_ark_client(): global _ark_client if _ark_client: - # AsyncArk 无显式 close，但可设为 None - _ark_client = None \ No newline at end of file + _ark_client = None + +def cleanup_aicc_client(): + global _aicc_client + if _aicc_client: + _aicc_client = None + +def cleanup_all(): + cleanup_ark_client() + cleanup_aicc_client() \ No newline at end of file diff --git a/doubao/router/analyze.py b/doubao/router/analyze.py index ee7164d..bd50ebe 100644 --- a/doubao/router/analyze.py +++ b/doubao/router/analyze.py @@ -1,7 +1,7 @@ import os from chongming_worker import rpc_method from ..models.pydantic import AnalyzePostRequest, AnalyzePostResponse -from ..global_var import get_ark_client +from ..global_var import get_client, get_is_aicc @rpc_method( subject="doubao.analyze", @@ -11,7 +11,7 @@ from ..global_var import get_ark_client ) async def analyze(request: AnalyzePostRequest) -> AnalyzePostResponse: print(f"[DEBUG] analyze called with prompt={request.prompt}, image_base64 length={len(request.image_base64)}") - client = get_ark_client() + client = get_client() if client is None: return AnalyzePostResponse(result="", error="客户端未初始化") @@ -29,6 +29,12 @@ async def analyze(request: AnalyzePostRequest) -> AnalyzePostResponse: prompt = request.prompt or "" try: + # AICC 模式的 AsyncOpenAI 不支持 thinking 参数，需通过 extra_body 传递 + if get_is_aicc(): + create_kwargs = {"extra_body": {"thinking": {"type": "disabled"}}} + else: + create_kwargs = {"thinking": {"type": "disabled"}} + response = await client.chat.completions.create( model=os.getenv("DOUBAO_MODEL", "doubao-seed-evolving"), messages=[ @@ -40,7 +46,7 @@ async def analyze(request: AnalyzePostRequest) -> AnalyzePostResponse: ] } # type: ignore ], - thinking={"type": "disabled"} + **create_kwargs # type: ignore ) result = response.choices[0].message.content # type: ignore return AnalyzePostResponse(result=result) # type: ignore diff --git a/example/README.md b/example/README.md index 652a219..18a1033 100644 --- a/example/README.md +++ b/example/README.md @@ -1,22 +1,71 @@ # Example Worker 插件 -演示 `@rpc_method` 与显式模型绑定的用法。 +演示 `@rpc_method` 装饰器与显式模型绑定的完整用法，是学习开发新插件的**入门参考**。 + +--- ## 实现的功能 -- 加、减、乘、除四则运算（GET / POST） -- 基于 `modelvault` 的访问计数器 +### 四则运算（GET / POST） -## 如何工作 +| 端点 | 方法 | 路径 | 说明 | +| ---- | ---------- | ---------------- | ------------------------ | +| 加法 | GET / POST | `/calc/add` | `a + b` | +| 减法 | GET | `/calc/subtract` | `a - b` | +| 乘法 | GET | `/calc/multiply` | `a * b` | +| 除法 | GET | `/calc/divide` | `a / b`（除数为 0 报错） | -1. 配置位于 `config/workers/example/*.toml` -2. 模型由 `gen_models` 生成，位于 `models/pydantic.py` -3. 路由实现在 `router/` 下，使用 `@rpc_method` 装饰 -4. 工厂函数 `create_workers` 初始化数据库并启动 Worker +### 访问计数器 + +| 端点 | 方法 | 路径 | 说明 | +| ---- | ---- | ------------- | ----------------- | +| 计数 | GET | `/calc/count` | 每次调用计数器 +1 | + +基于内存缓存实现，用于演示 GET 端点的无参数调用模式。 + +--- ## 调用示例 ```bash +# GET 加法 curl "http://localhost:8000/api/v1/calc/add?a=3&b=4" -# {"result":7,"operation":"add","timestamp":...} +# {"result":7.0,"operation":"add","timestamp":1720000000.123} + +# POST 加法 +curl -X POST "http://localhost:8000/api/v1/calc/add" \ + -H "Content-Type: application/json" \ + -d '{"a": 3.5, "b": 2.5}' +# {"result":6.0,"operation":"add","timestamp":1720000000.456} + +# 除法（除数为零） +curl "http://localhost:8000/api/v1/calc/divide?a=10&b=0" +# {"result":0.0,"operation":"divide","error":"Division by zero"} + +# 计数器 +curl "http://localhost:8000/api/v1/calc/count" +# {"count":1} ``` + +--- + +## 如何工作 + +1. **配置**：`config/workers/example/*.toml` — 定义每个端点的参数和响应结构 +2. **模型**：`models/pydantic.py` — 由 `gen_models` 根据 TOML 自动生成 +3. **路由**：`router/` 下每个文件实现一个 RPC 端点，使用 `@rpc_method` 装饰器 +4. **工厂**：`__init__.py` 中的 `create_workers` 初始化数据库并启动 Worker + +--- + +## 开发参考 + +此插件展示了以下模式： + +- GET 端点的 `params` 定义与 `@rpc_method` 参数绑定 +- POST 端点的 JSON body 自动解析 +- `response_model` 在 TOML 中定义并在装饰器中引用 +- 全局变量 (`global_var.py`) 管理共享资源 +- 错误处理（如除法除零返回错误信息） +- `auth_required` 控制端点是否需要 JWT 鉴权 +- `cacheable` 控制结果是否缓存 diff --git a/example/models/__init__.py b/example/models/__init__.py index 6dad973..e00d0ba 100644 --- a/example/models/__init__.py +++ b/example/models/__init__.py @@ -5,31 +5,32 @@ from .pydantic import AddGetRequest from .pydantic import AddGetResponse +from .pydantic import AddMqttRequest from .pydantic import AddPostRequest from .pydantic import AddPostResponse from .pydantic import CountGetResponse from .pydantic import DivideGetRequest from .pydantic import DivideGetResponse from .pydantic import ExampleResponse +from .pydantic import HeartbeatMqttRequest from .pydantic import MultiplyGetRequest from .pydantic import MultiplyGetResponse from .pydantic import SubtractGetRequest from .pydantic import SubtractGetResponse -from .counter import Counter - __all__ = [ "AddGetRequest", "AddGetResponse", + "AddMqttRequest", "AddPostRequest", "AddPostResponse", "CountGetResponse", "DivideGetRequest", "DivideGetResponse", "ExampleResponse", + "HeartbeatMqttRequest", "MultiplyGetRequest", "MultiplyGetResponse", "SubtractGetRequest", "SubtractGetResponse", - "Counter" ] diff --git a/example/models/pydantic.py b/example/models/pydantic.py index 41114be..f4fd602 100644 --- a/example/models/pydantic.py +++ b/example/models/pydantic.py @@ -30,6 +30,11 @@ class AddPostResponse(BaseModel): operation: str = Field(default='add') timestamp: Optional[float] = Field(default=0.0) +class AddMqttRequest(BaseModel): + """Add two numbers via MQTT — 请求体""" + a: float = Field(...) + b: float = Field(...) + class CountGetResponse(BaseModel): """Get visit count and increment — 响应体""" name: str = Field(...) @@ -47,6 +52,10 @@ class DivideGetResponse(BaseModel): timestamp: Optional[float] = Field(default=0.0) error: Optional[str] = Field(default='') +class HeartbeatMqttRequest(BaseModel): + """Heartbeat ping (no response) — 请求体""" + worker_id: str = Field(...) + class MultiplyGetRequest(BaseModel): """Multiply two numbers — 请求体""" a: float = Field(...) @@ -71,4 +80,4 @@ class SubtractGetResponse(BaseModel): # 所有响应模型的联合类型 -ExampleResponse = SubtractGetResponse | CountGetResponse | MultiplyGetResponse | AddPostResponse | AddGetResponse | DivideGetResponse +ExampleResponse = AddPostResponse | AddGetResponse | DivideGetResponse | SubtractGetResponse | CountGetResponse | MultiplyGetResponse diff --git a/example/router/__init__.py b/example/router/__init__.py index e6d582e..2679b1a 100644 --- a/example/router/__init__.py +++ b/example/router/__init__.py @@ -1,4 +1,6 @@ from .add import add, add_post +from .add_mqtt import add_mqtt +from .heartbeat import heartbeat from .subtract import subtract from .multiply import multiply from .divide import divide @@ -8,6 +10,8 @@ from .count import count_get __all__ = [ "add", "add_post", + "add_mqtt", + "heartbeat", "subtract", "multiply", "divide", diff --git a/example/router/add_mqtt.py b/example/router/add_mqtt.py new file mode 100644 index 0000000..0ec598f --- /dev/null +++ b/example/router/add_mqtt.py @@ -0,0 +1,24 @@ +""" +MQTT 端点处理器 — 通过 MQTT 主题 chongming/example/add 调用 + +与 GET/POST 的 add 共用同一个 subject (example.add)， +通过 _method="MQTT" 区分。 +""" +import time + +from chongming_worker import rpc_method +from ..models.pydantic import AddMqttRequest + + +@rpc_method( + subject="example.add", + method="MQTT", + summary="Add two numbers via MQTT", +) +async def add_mqtt(request: AddMqttRequest) -> dict: + """MQTT 加法 — 通过 MQTT 主题 chongming/example/add 调用，返回 dict""" + return { + "result": request.a + request.b, + "operation": "add_mqtt", + "timestamp": time.time(), + } diff --git a/example/router/heartbeat.py b/example/router/heartbeat.py new file mode 100644 index 0000000..eee7c95 --- /dev/null +++ b/example/router/heartbeat.py @@ -0,0 +1,20 @@ +""" +心跳 MQTT 端点 — 仅上报，无需响应 + +通过 MQTT 主题 chongming/example/heartbeat 发送心跳， +Worker 返回 None，MQTT Bridge 不发布响应。 +""" +import time + +from chongming_worker import rpc_method + + +@rpc_method( + subject="example.heartbeat", + method="MQTT", + summary="Heartbeat ping via MQTT (no response)", +) +async def heartbeat(worker_id: str = "") -> None: + """心跳上报 — 无返回值，MQTT Bridge 跳过响应发布""" + print(f"[Heartbeat] worker={worker_id} time={time.time()}") + return None # 显式返回 None，MQTT Bridge 不发布响应