381 lines
15 KiB
Python
381 lines
15 KiB
Python
"""
|
||
AI Box Sophon Adapter 插件
|
||
|
||
基于 MQTT 的 AI Box 设备心跳适配器，实现：
|
||
1. 通过 MQTT 主题 /board_ping 接收设备心跳上报
|
||
2. 解析设备系统参数（温度、硬盘、内存、通道、任务、TPU 等）
|
||
3. 持久化心跳数据到 SQLite 数据库
|
||
4. 提供 HTTP API 查询设备在线状态和详情
|
||
5. 后台定时检查离线设备，触发异常告警
|
||
6. 视频上传接口 — 盒子告警时录制视频上传（FORM-DATA）
|
||
7. 告警上报接口 — 违规事件 JSON 报文字动解析入库
|
||
8. 告警查询接口 — 告警列表/详情查询
|
||
|
||
心跳超时机制:
|
||
- 设备按配置间隔（默认 5 秒）上报心跳
|
||
- 服务端每 check_interval 秒检查一次
|
||
- 超过 heartbeat_timeout 秒未收到心跳的设备标记为离线
|
||
"""
|
||
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 .router import *
|
||
from .routes import router # 自定义 FastAPI 路由（网关自动发现）
|
||
|
||
# 外部 MQTT 订阅请求处理函数（订阅 /edge_app_controller 等主题的 handler）
|
||
from .edge_app_controller import edge_app_controller # noqa: F401
|
||
|
||
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):
|
||
"""
|
||
后台任务：定期检查离线设备
|
||
|
||
每隔 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) # type: ignore
|
||
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
|
||
from chongming_stream_manager import StreamManager
|
||
|
||
# 所有运行时配置均从 [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")
|
||
|
||
# 心跳配置
|
||
heartbeat = config.get("heartbeat", {})
|
||
hb_interval = heartbeat.get("interval", 5)
|
||
hb_timeout = heartbeat.get("timeout", _default_timeout)
|
||
check_interval = heartbeat.get("check_interval", 10)
|
||
|
||
# 告警保留配置
|
||
alarm_retention = config.get("alarm_retention", {})
|
||
alarm_max = alarm_retention.get("max_records", 0)
|
||
|
||
# 豆包 AI 分析开关
|
||
doubao_cfg = config.get("doubao_analysis", {})
|
||
doubao_enabled = doubao_cfg.get("enabled", 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_path = data_dir / "aibox_sophon_adapter.db"
|
||
db_path.parent.mkdir(parents=True, exist_ok=True)
|
||
|
||
db = Database(db_path)
|
||
await db.initialize()
|
||
|
||
# 初始化视频流控制器
|
||
stream_manager = StreamManager(mediamtx_addr)
|
||
|
||
# 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}"
|
||
)
|
||
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
|
||
|
||
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}"
|
||
)
|
||
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)
|
||
|
||
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, 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} 清理完毕")
|