chongming/chongming-plugins
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

modify: 统一插件风格

Browse Source
This commit is contained in:
chakcy
2026-08-14 16:04:39 +08:00
parent 56dd4f0ebb
commit 4fa066432d
10 changed files with 516 additions and 436 deletions
Download Patch File Download Diff File Expand all files Collapse all files
aibox_sophon_adapter/README.md
+7 -5
View File
@@ -87,10 +87,12 @@ sequenceDiagram
```text
src/plugins/aibox_sophon_adapter/
├── __init__.py # 插件入口：create_workers() 工厂 + cleanup()
│ # - 初始化 DB / StreamManager / 各 MQTT 客户端
│ # - _build_subscribe_routes() 构建订阅路由
│ # - 启动离线检查后台任务
├── __init__.py # 插件入口：AiboxSophonAdapterPlugin（Plugin 基类）
│ # - on_init: DB / StreamManager / MQTT 客户端 / 全局变量 / 后台任务
│ # - on_cleanup: 流控制器与 MQTT 客户端关闭（db/后台任务由基类处理）
├── mqtt_clients.py # MQTT 客户端能力组件（从 __init__ 抽取，统一"启动失败降级为 None"模式）
│ # - start_mqtt_forward / start_mqtt_subscribe / start_internal_mqtt
│ # - build_subscribe_routes() 构建订阅路由
├── global_var.py # 模块级全局变量（db / stream_manager / mqtt_* / 各配置）
├── routes.py # 自定义 FastAPI 路由（video_upload / video / alarm_report）
├── edge_app_controller.py # 外部 MQTT 请求-响应处理器（command 注册表分发）
@@ -356,7 +358,7 @@ tags = ["aibox_sophon_adapter", "sophon", "adapter"]
### 修改配置
编辑 `config/workers/aibox_sophon_adapter/config.toml` 后重启服务即可生效，无需重新生成模型
（运行时配置均在 `create_workers` 中读取）。
（运行时配置均在 `AiboxSophonAdapterPlugin.on_init` 中读取；MQTT 客户端启动逻辑见 `mqtt_clients.py`）。
---
aibox_sophon_adapter/__init__.py
+95 -288
View File
@@ -2,7 +2,7 @@
AI Box Sophon Adapter 插件
基于 MQTT 的 AI Box 设备心跳适配器，实现：
1. 通过 MQTT 主题 /board_ping 接收设备心跳上报
1. 通过 MQTT 主题接收设备心跳上报
2. 解析设备系统参数（温度、硬盘、内存、通道、任务、TPU 等）
3. 持久化心跳数据到 SQLite 数据库
4. 提供 HTTP API 查询设备在线状态和详情
@@ -15,122 +15,25 @@ AI Box Sophon Adapter 插件
- 设备按配置间隔（默认 5 秒）上报心跳
- 服务端每 check_interval 秒检查一次
- 超过 heartbeat_timeout 秒未收到心跳的设备标记为离线
生命周期: 基于 Plugin 基类统一管理（数据库/后台任务/清理钩子由基类处理），
MQTT 客户端启动块已抽取到 mqtt_clients 能力组件。
"""
import asyncio
import logging
from datetime import datetime
from pathlib import Path
from typing import Any
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 .routes import router # 自定义 FastAPI 路由（网关自动发现）
# 外部 MQTT 订阅请求处理函数（订阅 /edge_app_controller 等主题的 handler）
from .router import * # noqa: F401,F403 触发 @rpc_method 注册
from .routes import router # noqa: F401 自定义 FastAPI 路由（网关通过包属性自动发现）
from .edge_app_controller import edge_app_controller # noqa: F401
from .mqtt_clients import start_mqtt_forward, start_mqtt_subscribe, start_internal_mqtt
logger = logging.getLogger(__name__)
def _build_subscribe_routes(topics_cfg: list) -> list:
"""
根据配置的订阅主题列表构建 MQTTSubscribeClient 路由。
topics_cfg: [{"topic": "...", "handler": "board_ping", "reply_topic": "..."}, ...]
- topic: 订阅主题（支持 + / # 通配符）
- handler: 处理函数名，按序从插件 router 模块、插件包自身查找
- reply_topic: 可选，回复主题；处理后返回值将发布到该主题（支持 {1}{2} 占位符）
适配约定（按首个位置参数类型）:
- Pydantic 模型: 用 payload 构造模型实例后调用（如 board_ping -> BoardPingRequest(**payload)）
- dict: 直接传整个 payload 和 topic（如 edge_app_controller(payload, topic)）
- 其他: 将 payload 作为关键字参数展开调用
返回: [{"topic": ..., "handler": async callable(payload, topic), "reply_topic": ...}]
"""
import importlib
from typing import get_origin
from pydantic import BaseModel
# 注意: 不能用 "from . import router"，因为 __init__.py 中
# "from .routes import router" 已将 router 设为包级属性（FastAPI APIRouter 对象），
# 会遮蔽子模块包 aibox_sophon_adapter.router，故用 importlib 显式按名导入。
router_mod = importlib.import_module(f"{__name__}.router")
pkg_mod = importlib.import_module(__name__)
routes = []
for t in topics_cfg or []:
topic = t.get("topic", "")
handler_name = t.get("handler", "")
reply_topic = t.get("reply_topic")
if not topic or not handler_name:
logger.warning(f"[MQTTSubscribe] 跳过无效订阅项: {t}")
continue
fn = getattr(router_mod, handler_name, None)
if fn is None:
fn = getattr(pkg_mod, handler_name, None)
if fn is None:
logger.warning(
f"[MQTTSubscribe] 未知 handler '{handler_name}'，跳过 {topic}"
)
continue
# 探测首个位置参数的类型注解
hints = getattr(fn, "__annotations__", {})
first_type: Any = None
for pname, ptype in hints.items():
if pname in ("worker_id", "return"):
continue
first_type = ptype
break
if (
isinstance(first_type, type)
and issubclass(first_type, BaseModel)
):
model_cls = first_type
async def _cb_model(data, topic, _fn=fn, _model=model_cls):
return await _fn(_model(**data))
routes.append({
"topic": topic,
"handler": _cb_model,
"reply_topic": reply_topic,
})
elif first_type is dict or get_origin(first_type) is dict:
# 处理函数期望整个 dict 作为第一个参数（如 edge_app_controller(payload, topic)）
async def _cb_payload(data, topic, _fn=fn):
return await _fn(data, topic)
routes.append({
"topic": topic,
"handler": _cb_payload,
"reply_topic": reply_topic,
})
else:
# 其他签名：将 payload 按关键字参数展开调用
async def _cb_kwargs(data, topic, _fn=fn):
if isinstance(data, dict):
return await _fn(**data)
return await _fn()
routes.append({
"topic": topic,
"handler": _cb_kwargs,
"reply_topic": reply_topic,
})
return routes
# 自动绑定 Pydantic 模型（用于 gen_models 生成的模型）
auto_bind_models("aibox_sophon_adapter")
# 离线检查任务引用
_offline_check_task: asyncio.Task | None = None
async def _check_offline_devices(interval: float):
"""
后台任务：定期检查离线设备
@@ -150,7 +53,7 @@ async def _check_offline_devices(interval: float):
from .models.orm import BoardHeartbeat
from sqlalchemy import select
stmt = select(BoardHeartbeat).where(BoardHeartbeat.is_online == True) # type: ignore
stmt = select(BoardHeartbeat).where(BoardHeartbeat.is_online)
result = await session.execute(stmt)
records = result.scalars().all()
@@ -175,206 +78,110 @@ async def _check_offline_devices(interval: float):
logger.error(f"[OfflineChecker] 检查异常: {e}")
async def create_workers(session, config, count=1, worker_manager=None):
"""
插件入口 — 初始化数据库和全局资源，创建 Worker 副本
class AiboxSophonAdapterPlugin(Plugin):
"""AI Box 适配插件 — 统一生命周期。"""
由 FastAPI lifespan 在启动时调用。
"""
from .global_var import set_global_var, heartbeat_timeout as _default_timeout
from chongming_database import Database
from chongming_stream_manager import StreamManager
type = "aibox_sophon_adapter"
db_name = "aibox_sophon_adapter.db" # 自动在 data_dir 下建 SQLite 库
auto_bind = True # 首次初始化时自动 auto_bind_models
# 所有运行时配置均从 [workers.config] 及其子节读取
data_dir = Path(config.get("data_dir", "./data"))
mediamtx_addr = config.get("mediamtx_addr", "rtsp://127.0.0.1:8554")
link_path = config.get("link_path", "/data/bm-app")
async def on_init(self):
"""首次 create_workers 时调用一次：初始化数据库客户端、流控制器、后台任务。"""
config = self.config
from .global_var import set_global_var, heartbeat_timeout as _default_timeout
from chongming_stream_manager import StreamManager
# 心跳配置
heartbeat = config.get("heartbeat", {})
hb_interval = heartbeat.get("interval", 5)
hb_timeout = heartbeat.get("timeout", _default_timeout)
check_interval = heartbeat.get("check_interval", 10)
# 所有运行时配置均从 [workers.config] 及其子节读取
data_dir = Path(config.get("data_dir", "./data"))
mediamtx_addr = config.get("mediamtx_addr", "rtsp://127.0.0.1:8554")
link_path = config.get("link_path", "/data/bm-app")
# 告警保留配置
alarm_retention = config.get("alarm_retention", {})
alarm_max = alarm_retention.get("max_records", 0)
# 心跳配置
heartbeat = config.get("heartbeat", {})
hb_interval = heartbeat.get("interval", 5)
hb_timeout = heartbeat.get("timeout", _default_timeout)
check_interval = heartbeat.get("check_interval", 10)
# 豆包 AI 分析开关
doubao_cfg = config.get("doubao_analysis", {})
doubao_enabled = doubao_cfg.get("enabled", True)
# 告警保留配置
alarm_retention = config.get("alarm_retention", {})
alarm_max = alarm_retention.get("max_records", 0)
# 告警转发配置
alarm_fwd_cfg = config.get("alarm_forward", {})
alarm_fwd_enabled = alarm_fwd_cfg.get("enabled", False)
alarm_fwd_url = alarm_fwd_cfg.get("url", "http://dg.datagoo.cn:60014/open-api/event-alarm/receive")
alarm_fwd_video_url = alarm_fwd_cfg.get("video_upload_url", "http://dg.datagoo.cn:60014/open-api/event-alarm/receiveVideoUpload")
alarm_fwd_timeout = alarm_fwd_cfg.get("timeout", 10.0)
# 豆包 AI 分析开关
doubao_cfg = config.get("doubao_analysis", {})
doubao_enabled = doubao_cfg.get("enabled", True)
# 初始化数据库
db_path = data_dir / "aibox_sophon_adapter.db"
db_path.parent.mkdir(parents=True, exist_ok=True)
# 告警转发配置
alarm_fwd_cfg = config.get("alarm_forward", {})
alarm_fwd_enabled = alarm_fwd_cfg.get("enabled", False)
alarm_fwd_url = alarm_fwd_cfg.get("url", "http://dg.datagoo.cn:60014/open-api/event-alarm/receive")
alarm_fwd_video_url = alarm_fwd_cfg.get("video_upload_url", "http://dg.datagoo.cn:60014/open-api/event-alarm/receiveVideoUpload")
alarm_fwd_timeout = alarm_fwd_cfg.get("timeout", 10.0)
db = Database(db_path)
await db.initialize()
# 数据库已由基类初始化（data_dir/aibox_sophon_adapter.db）
db = self.db
# 初始化视频流控制器
stream_manager = StreamManager(mediamtx_addr)
# 初始化视频流控制器
stream_manager = StreamManager(mediamtx_addr)
# upload 根目录（用于清理告警关联视频）
upload_root_path = str(data_dir / "uploads")
# upload 根目录（用于清理告警关联视频）
upload_root_path = str(data_dir / "uploads")
# 设定模块级全局变量（供 _check_offline_devices 使用）
global heartbeat_timeout
heartbeat_timeout = hb_timeout
# 初始化外部 MQTT 转发客户端
mqtt_fwd = config.get("mqtt_forward", {})
mqtt_fwd_enabled = mqtt_fwd.get("enabled", False)
mqtt_fwd_client = None
if mqtt_fwd_enabled:
from chongming_mqtt.mqtt_forward import MQTTForwardClient
mqtt_fwd_uri = mqtt_fwd.get("uri", "")
mqtt_fwd_topic = mqtt_fwd.get("topic", "/board_ping")
logger.warning(f"[MQTTForward] 配置已加载: enabled={mqtt_fwd_enabled}, uri={mqtt_fwd_uri}, topic={mqtt_fwd_topic}")
if mqtt_fwd_uri:
mqtt_fwd_client = MQTTForwardClient(uri=mqtt_fwd_uri, topic=mqtt_fwd_topic)
try:
await mqtt_fwd_client.start()
logger.warning(f"[MQTTForward] 外部转发已启用: {mqtt_fwd_uri} → {mqtt_fwd_topic}")
except Exception as e:
logger.warning(f"[MQTTForward] 启动失败，将跳过转发: {e}")
mqtt_fwd_client = None
else:
logger.warning(f"[MQTTForward] mqtt_forward 未启用，配置内容: {mqtt_fwd}")
# 初始化外部 MQTT 订阅客户端（本插件直接订阅外部 Broker，不走网关 Bridge）
mqtt_sub = config.get("mqtt_subscribe", {})
mqtt_sub_enabled = mqtt_sub.get("enabled", False)
mqtt_sub_client = None
if mqtt_sub_enabled:
from chongming_mqtt.mqtt_subscribe import MQTTSubscribeClient
mqtt_sub_uri = mqtt_sub.get("uri", "")
mqtt_sub_topics = mqtt_sub.get("topics", []) or []
logger.warning(
f"[MQTTSubscribe] 配置已加载: enabled={mqtt_sub_enabled}, "
f"uri={mqtt_sub_uri}, topics={mqtt_sub_topics}"
# MQTT 客户端能力组件（启动失败降级为 None，仅警告）
mqtt_fwd_client = await start_mqtt_forward(config.get("mqtt_forward", {}))
mqtt_sub_client = await start_mqtt_subscribe(config.get("mqtt_subscribe", {}))
int_mqtt_client, int_req_topic, int_resp_topic = await start_internal_mqtt(
config.get("internal_mqtt", {})
)
if mqtt_sub_uri and mqtt_sub_topics:
routes = _build_subscribe_routes(mqtt_sub_topics)
mqtt_sub_client = MQTTSubscribeClient(uri=mqtt_sub_uri, routes=routes)
try:
await mqtt_sub_client.start()
logger.warning(f"[MQTTSubscribe] 外部订阅已启用: {mqtt_sub_uri}")
except Exception as e:
logger.warning(f"[MQTTSubscribe] 启动失败，将跳过订阅: {e}")
mqtt_sub_client = None
else:
logger.warning(
"[MQTTSubscribe] 未配置 uri 或 topics，跳过外部订阅"
)
else:
logger.warning(f"[MQTTSubscribe] mqtt_subscribe 未启用，配置内容: {mqtt_sub}")
# 初始化内部 MQTT 请求-响应客户端
# （handler 内对任意应用进程发起请求-响应，主题显式传入，默认连内部 127.0.0.1:1883）
int_mqtt_cfg = config.get("internal_mqtt", {})
int_mqtt_enabled = int_mqtt_cfg.get("enabled", False)
int_mqtt_client = None
# 内部请求-响应主题（edge_app_controller 的 alg_media_fetch 命令使用）。
# 与盒子侧约定一致（盒子只设根主题 box，子路径固定 aibox_sophon/...）；
# 勿用 chongming/{插件}/{操作} 保留命名空间——网关 MQTT Bridge 订阅了
# chongming/+/+，会拦截 3 级主题并按 Worker RPC 路由，而 edge_app_controller
# 不是 @rpc_method（Worker 无对应 Queryable），导致 worker_manager 报
# "Worker ... 无回复，尝试重试"。
int_req_topic = int_mqtt_cfg.get(
"request_topic", "box/aibox_sophon/edge_app_controller"
)
int_resp_topic = int_mqtt_cfg.get(
"response_topic", "box/aibox_sophon/edge_app_controller_reply"
)
if int_mqtt_enabled:
from chongming_mqtt.internal_mqtt import InternalMQTTClient
set_global_var("db", db)
set_global_var("heartbeat_timeout", hb_timeout)
set_global_var("heartbeat_interval", hb_interval)
set_global_var("stream_manager", stream_manager)
set_global_var("link_path", link_path)
set_global_var("mqtt_forward", mqtt_fwd_client)
set_global_var("mqtt_subscribe", mqtt_sub_client)
set_global_var("internal_mqtt", int_mqtt_client)
set_global_var("internal_request_topic", int_req_topic)
set_global_var("internal_response_topic", int_resp_topic)
set_global_var("upload_root", upload_root_path)
set_global_var("alarm_max_records", alarm_max)
set_global_var("doubao_analysis_enabled", doubao_enabled)
set_global_var("alarm_forward_enabled", alarm_fwd_enabled)
set_global_var("alarm_forward_url", alarm_fwd_url)
set_global_var("alarm_forward_video_url", alarm_fwd_video_url)
set_global_var("alarm_forward_timeout", alarm_fwd_timeout)
int_mqtt_uri = int_mqtt_cfg.get("uri", "mqtt://127.0.0.1:1883")
logger.warning(
f"[InternalMQTT] 配置已加载: enabled={int_mqtt_enabled}, uri={int_mqtt_uri}, "
f"request_topic={int_req_topic}, response_topic={int_resp_topic}"
logger.info(
"[aibox_sophon_adapter] 已初始化: "
f"心跳间隔={hb_interval}s, 超时={hb_timeout}s, 检查间隔={check_interval}s"
)
int_mqtt_client = InternalMQTTClient(uri=int_mqtt_uri)
try:
await int_mqtt_client.start()
logger.warning(f"[InternalMQTT] 内部请求-响应客户端已启用: {int_mqtt_uri}")
except Exception as e:
logger.warning(f"[InternalMQTT] 启动失败，将跳过: {e}")
int_mqtt_client = None
else:
logger.warning(f"[InternalMQTT] internal_mqtt 未启用，配置内容: {int_mqtt_cfg}")
set_global_var("db", db)
set_global_var("heartbeat_timeout", hb_timeout)
set_global_var("heartbeat_interval", hb_interval)
set_global_var("stream_manager", stream_manager)
set_global_var("link_path", link_path)
set_global_var("mqtt_forward", mqtt_fwd_client)
set_global_var("mqtt_subscribe", mqtt_sub_client)
set_global_var("internal_mqtt", int_mqtt_client)
set_global_var("internal_request_topic", int_req_topic)
set_global_var("internal_response_topic", int_resp_topic)
set_global_var("upload_root", upload_root_path)
set_global_var("alarm_max_records", alarm_max)
set_global_var("doubao_analysis_enabled", doubao_enabled)
set_global_var("alarm_forward_enabled", alarm_fwd_enabled)
set_global_var("alarm_forward_url", alarm_fwd_url)
set_global_var("alarm_forward_video_url", alarm_fwd_video_url)
set_global_var("alarm_forward_timeout", alarm_fwd_timeout)
# 后台任务由基类跟踪，cleanup 时自动取消
self.spawn_task(_check_offline_devices(check_interval))
logger.info(
f"[aibox_sophon_adapter] 已初始化: "
f"心跳间隔={hb_interval}s, 超时={hb_timeout}s, 检查间隔={check_interval}s"
)
async def on_cleanup(self):
"""cleanup 时调用：关闭流控制器与外部 MQTT 客户端（db 与后台任务由基类处理）。"""
from .global_var import (
stream_manager, mqtt_forward, mqtt_subscribe, internal_mqtt,
)
# 启动离线设备检查后台任务
global _offline_check_task
_offline_check_task = asyncio.create_task(_check_offline_devices(check_interval))
if stream_manager is not None:
await stream_manager.stop_all()
# 创建 Worker 副本
return _create_workers(session, config, count, worker_manager)
if mqtt_forward is not None:
await mqtt_forward.stop()
if mqtt_subscribe is not None:
await mqtt_subscribe.stop()
if internal_mqtt is not None:
await internal_mqtt.stop()
logger.info("[aibox_sophon_adapter] 清理完成")
async def cleanup(worker_item: str):
"""
清理资源
在 Worker 关闭时调用，关闭数据库连接和后台任务。
"""
from .global_var import (
db, stream_manager, mqtt_forward, mqtt_subscribe, internal_mqtt,
)
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()
if stream_manager is not None:
await stream_manager.stop_all()
if mqtt_forward is not None:
await mqtt_forward.stop()
if mqtt_subscribe is not None:
await mqtt_subscribe.stop()
if internal_mqtt is not None:
await internal_mqtt.stop()
logger.info(f"[aibox_sophon_adapter] {worker_item} 清理完毕")
# 暴露给 app lifespan 的模块级入口（bound method，签名与旧写法完全一致）
# 注意: 实例名必须为 plugin（router 已被 routes.router APIRouter 占用）
plugin = AiboxSophonAdapterPlugin()
create_workers = plugin.create_workers
cleanup = plugin.cleanup
aibox_sophon_adapter/mqtt_clients.py
+219
View File
@@ -0,0 +1,219 @@
"""
AI Box Sophon Adapter — MQTT 客户端能力组件
从插件 __init__.py 抽取的三个外部 MQTT 客户端启动块，统一模式:
- enabled=False → 仅警告"未启用"并返回 None
- enabled=True → 懒加载客户端类，启动失败降级为 None（仅警告，不阻断插件）
内部 MQTT 主题约定（勿改）:
与盒子侧约定一致（盒子只设根主题 box，子路径固定 aibox_sophon/...）；
勿用 chongming/{插件}/{操作} 保留命名空间——网关 MQTT Bridge 订阅了
chongming/+/+，会拦截 3 级主题并按 Worker RPC 路由，而 edge_app_controller
不是 @rpc_method（Worker 无对应 Queryable），导致 worker_manager 报
"Worker ... 无回复，尝试重试"。
"""
import logging
from typing import Any
logger = logging.getLogger(__name__)
# 插件包名（作为 importlib 探测的基准，兼容 plugins.xxx 与 xxx 两种导入路径）
_PKG = __package__ or "plugins.aibox_sophon_adapter"
def _import_first(candidates: list) -> Any:
"""按序尝试导入模块，全部失败返回 None。"""
import importlib
for name in candidates:
try:
return importlib.import_module(name)
except ImportError:
continue
return None
def build_subscribe_routes(topics_cfg: list) -> list:
"""
根据配置的订阅主题列表构建 MQTTSubscribeClient 路由。
topics_cfg: [{"topic": "...", "handler": "board_ping", "reply_topic": "..."}, ...]
- topic: 订阅主题（支持 + / # 通配符）
- handler: 处理函数名，按序从插件 router 模块、插件包自身查找
- reply_topic: 可选，回复主题；处理后返回值将发布到该主题（支持 {1}{2} 占位符）
适配约定（按首个位置参数类型）:
- Pydantic 模型: 用 payload 构造模型实例后调用（如 board_ping -> BoardPingRequest(**payload)）
- dict: 直接传整个 payload 和 topic（如 edge_app_controller(payload, topic)）
- 其他: 将 payload 作为关键字参数展开调用
返回: [{"topic": ..., "handler": async callable(payload, topic), "reply_topic": ...}]
"""
from typing import get_origin
from pydantic import BaseModel
# 注意: 不能用 "from .. import router" 取子模块，因为插件包属性 router
# 已被 routes.router（FastAPI APIRouter）遮蔽，故用 importlib 按名导入。
pkg_base = _PKG
pkg_leaf = _PKG.rsplit(".", 1)[-1]
router_mod = _import_first([f"{pkg_base}.router", f"{pkg_leaf}.router"])
pkg_mod = _import_first([pkg_base, pkg_leaf])
routes = []
for t in topics_cfg or []:
topic = t.get("topic", "")
handler_name = t.get("handler", "")
reply_topic = t.get("reply_topic")
if not topic or not handler_name:
logger.warning("[MQTTSubscribe] 跳过无效订阅项: %s", t)
continue
fn = None
if router_mod is not None:
fn = getattr(router_mod, handler_name, None)
if fn is None and pkg_mod is not None:
fn = getattr(pkg_mod, handler_name, None)
if fn is None:
logger.warning(
"[MQTTSubscribe] 未知 handler '%s'，跳过 %s", handler_name, topic
)
continue
# 探测首个位置参数的类型注解
hints = getattr(fn, "__annotations__", {})
first_type: Any = None
for pname, ptype in hints.items():
if pname in ("worker_id", "return"):
continue
first_type = ptype
break
if (
isinstance(first_type, type)
and issubclass(first_type, BaseModel)
):
model_cls = first_type
async def _cb_model(data, topic, _fn=fn, _model=model_cls):
return await _fn(_model(**data))
routes.append({
"topic": topic,
"handler": _cb_model,
"reply_topic": reply_topic,
})
elif first_type is dict or get_origin(first_type) is dict:
# 处理函数期望整个 dict 作为第一个参数（如 edge_app_controller(payload, topic)）
async def _cb_payload(data, topic, _fn=fn):
return await _fn(data, topic)
routes.append({
"topic": topic,
"handler": _cb_payload,
"reply_topic": reply_topic,
})
else:
# 其他签名：将 payload 按关键字参数展开调用
async def _cb_kwargs(data, topic, _fn=fn):
if isinstance(data, dict):
return await _fn(**data)
return await _fn()
routes.append({
"topic": topic,
"handler": _cb_kwargs,
"reply_topic": reply_topic,
})
return routes
async def start_mqtt_forward(cfg: dict) -> Any:
"""启动外部 MQTT 转发客户端，返回客户端或 None。"""
enabled = cfg.get("enabled", False)
if not enabled:
logger.warning("[MQTTForward] mqtt_forward 未启用，配置内容: %s", cfg)
return None
from chongming_mqtt.mqtt_forward import MQTTForwardClient
uri = cfg.get("uri", "")
topic = cfg.get("topic", "/board_ping")
logger.warning(
"[MQTTForward] 配置已加载: enabled=%s, uri=%s, topic=%s", enabled, uri, topic
)
if not uri:
return None
client = MQTTForwardClient(uri=uri, topic=topic)
try:
await client.start()
logger.warning("[MQTTForward] 外部转发已启用: %s → %s", uri, topic)
return client
except Exception as e:
logger.warning("[MQTTForward] 启动失败，将跳过转发: %s", e)
return None
async def start_mqtt_subscribe(cfg: dict) -> Any:
"""启动外部 MQTT 订阅客户端（本插件直接订阅外部 Broker，不走网关 Bridge）。"""
enabled = cfg.get("enabled", False)
if not enabled:
logger.warning("[MQTTSubscribe] mqtt_subscribe 未启用，配置内容: %s", cfg)
return None
from chongming_mqtt.mqtt_subscribe import MQTTSubscribeClient
uri = cfg.get("uri", "")
topics = cfg.get("topics", []) or []
logger.warning(
"[MQTTSubscribe] 配置已加载: enabled=%s, uri=%s, topics=%s",
enabled, uri, topics,
)
if not (uri and topics):
logger.warning("[MQTTSubscribe] 未配置 uri 或 topics，跳过外部订阅")
return None
routes = build_subscribe_routes(topics)
client = MQTTSubscribeClient(uri=uri, routes=routes)
try:
await client.start()
logger.warning("[MQTTSubscribe] 外部订阅已启用: %s", uri)
return client
except Exception as e:
logger.warning("[MQTTSubscribe] 启动失败，将跳过订阅: %s", e)
return None
async def start_internal_mqtt(cfg: dict) -> tuple[Any, str, str]:
"""
启动内部 MQTT 请求-响应客户端
（handler 内对任意应用进程发起请求-响应，主题显式传入，默认连内部 127.0.0.1:1883）。
返回 (client | None, request_topic, response_topic)：
主题无论是否启用都按配置解析并返回（由调用方写入 global_var）。
"""
req_topic = cfg.get(
"request_topic", "box/aibox_sophon/edge_app_controller"
)
resp_topic = cfg.get(
"response_topic", "box/aibox_sophon/edge_app_controller_reply"
)
enabled = cfg.get("enabled", False)
if not enabled:
logger.warning("[InternalMQTT] internal_mqtt 未启用，配置内容: %s", cfg)
return None, req_topic, resp_topic
from chongming_mqtt.internal_mqtt import InternalMQTTClient
uri = cfg.get("uri", "mqtt://127.0.0.1:1883")
logger.warning(
"[InternalMQTT] 配置已加载: enabled=%s, uri=%s, request_topic=%s, response_topic=%s",
enabled, uri, req_topic, resp_topic,
)
client = InternalMQTTClient(uri=uri)
try:
await client.start()
logger.warning("[InternalMQTT] 内部请求-响应客户端已启用: %s", uri)
return client, req_topic, resp_topic
except Exception as e:
logger.warning("[InternalMQTT] 启动失败，将跳过: %s", e)
return None, req_topic, resp_topic