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

modify: mqtt心跳转发

Browse Source
This commit is contained in:
chakcy
2026-08-03 15:06:29 +08:00
parent 84addb71ff
commit e43c25f5dc
6 changed files with 296 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
aibox_sophon_adapter/__init__.py
+31 -1
View File
@@ -113,11 +113,38 @@ async def create_workers(session, config, count=1, worker_manager=None):
# 初始化视频流控制器
stream_manager = StreamManager(mediamtx_addr)
# 读取告警保留配置
alarm_retention_config = worker_config.get("alarm_retention", {}) if isinstance(config, dict) else {}
alarm_max = alarm_retention_config.get("max_records", 0)
# upload 根目录（用于清理告警关联视频）
upload_root_path = config.get("data_dir", "./data/uploads") if isinstance(config, dict) else "./data/uploads"
# 初始化外部 MQTT 转发客户端
mqtt_fwd_config = worker_config.get("mqtt_forward", {}) if isinstance(config, dict) else {}
mqtt_fwd_enabled = mqtt_fwd_config.get("enabled", False)
mqtt_fwd_client = None
if mqtt_fwd_enabled:
from .mqtt_forward import MQTTForwardClient
mqtt_fwd_uri = mqtt_fwd_config.get("uri", "")
mqtt_fwd_topic = mqtt_fwd_config.get("topic", "/board_ping")
if mqtt_fwd_uri:
mqtt_fwd_client = MQTTForwardClient(uri=mqtt_fwd_uri, topic=mqtt_fwd_topic)
try:
await mqtt_fwd_client.start()
logger.info(f"[MQTTForward] 外部转发已启用: {mqtt_fwd_uri} → {mqtt_fwd_topic}")
except Exception as e:
logger.warning(f"[MQTTForward] 启动失败，将跳过转发: {e}")
mqtt_fwd_client = None
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("upload_root", upload_root_path)
set_global_var("alarm_max_records", alarm_max)
logger.info(
f"[aibox_sophon_adapter] 已初始化: "
@@ -138,7 +165,7 @@ async def cleanup(worker_item: str):
在 Worker 关闭时调用，关闭数据库连接和后台任务。
"""
from .global_var import db, stream_manager
from .global_var import db, stream_manager, mqtt_forward
global _offline_check_task
if _offline_check_task is not None and not _offline_check_task.done():
@@ -155,4 +182,7 @@ async def cleanup(worker_item: str):
if stream_manager is not None:
await stream_manager.stop_all()
if mqtt_forward is not None:
await mqtt_forward.stop()
logger.info(f"[aibox_sophon_adapter] {worker_item} 清理完毕")
aibox_sophon_adapter/alarm_trim.py
+94
View File
@@ -0,0 +1,94 @@
"""
AI Box Sophon Adapter — 告警记录清理
当告警记录超过配置的 max_records 时，自动清理最早的记录及其关联的视频文件。
"""
import logging
import os
import shutil
from sqlalchemy import select, func, asc
from .models.orm import AlarmRecord
logger = logging.getLogger(__name__)
async def trim_alarm_records(db_session, upload_root: str, max_records: int) -> int:
"""
裁剪告警记录，保留最新的 max_records 条
删除超出限制的最早记录，同时清理关联的视频文件目录。
Args:
db_session: SQLAlchemy 异步 session
upload_root: 上传文件根目录（如 ./data/uploads）
max_records: 最大保留条数，<= 0 表示不限制
Returns:
实际删除的记录数
"""
if max_records <= 0:
return 0
# 查询当前总记录数
count_result = await db_session.execute(
select(func.count()).select_from(AlarmRecord)
)
total = count_result.scalar() or 0
if total <= max_records:
return 0
# 超出数量
overflow = total - max_records
# 查询最早的 overflow 条记录（按 created_at 升序）
stmt = (
select(AlarmRecord)
.order_by(asc(AlarmRecord.created_at)) # type: ignore
.limit(overflow)
)
result = await db_session.execute(stmt)
old_records = result.scalars().all()
deleted_count = 0
for record in old_records:
# 清理关联的视频文件目录
if record.video_file:
_delete_video_dir(upload_root, record.video_file)
await db_session.delete(record)
deleted_count += 1
await db_session.commit()
if deleted_count > 0:
logger.info(
f"[AlarmTrim] 已清理 {deleted_count} 条告警记录 "
f"（保留最新 {max_records} 条，原共 {total} 条）"
)
return deleted_count
def _delete_video_dir(upload_root: str, video_id: str) -> None:
"""
删除告警关联的视频文件目录
视频存储路径: {upload_root}/aibox_sophon_adapter/{video_id}/
Args:
upload_root: 上传根目录
video_id: 视频唯一标识（VideoId）
"""
video_dir = os.path.join(upload_root, "aibox_sophon_adapter", video_id)
if not os.path.isdir(video_dir):
return
try:
shutil.rmtree(video_dir)
logger.debug(f"[AlarmTrim] 已删除视频目录: {video_dir}")
except OSError as e:
logger.warning(f"[AlarmTrim] 删除视频目录失败 [{video_dir}]: {e}")
aibox_sophon_adapter/global_var.py
+16 -1
View File
@@ -3,12 +3,15 @@ AI Box Sophon Adapter — 全局变量
模块级共享资源管理。
"""
from typing import Any
from typing import Any, TYPE_CHECKING
from chongming_database import Database
from .models.orm import *
from chongming_stream_manager import StreamManager
if TYPE_CHECKING:
from .mqtt_forward import MQTTForwardClient
# 数据库实例
db: Database | None = None
@@ -16,6 +19,15 @@ db: Database | None = None
stream_manager: StreamManager | None = None
link_path: str | None = None
# 外部 MQTT 转发客户端
mqtt_forward: "MQTTForwardClient | None" = None
# 上传文件根目录（用于清理告警关联视频）
upload_root: str = "./data/uploads"
# 告警保留配置
alarm_max_records: int = 0 # 0 表示不限制
# 心跳超时配置（秒）
heartbeat_timeout: float = 15.0
@@ -38,5 +50,8 @@ __all__ = [
"db",
"heartbeat_timeout",
"heartbeat_interval",
"mqtt_forward",
"upload_root",
"alarm_max_records",
"set_global_var",
]
aibox_sophon_adapter/mqtt_forward.py
+119
View File
@@ -0,0 +1,119 @@
"""
AI Box Sophon Adapter — 外部 MQTT 转发客户端
建立到外部 MQTT Broker 的长连接，将设备心跳数据转发到指定主题。
"""
import asyncio
import json
import logging
from urllib.parse import urlparse, unquote
from amqtt.client import MQTTClient
from amqtt.mqtt.constants import QOS_1
logger = logging.getLogger(__name__)
class MQTTForwardClient:
"""
外部 MQTT 转发客户端
在插件启动时建立到外部 MQTT Broker 的持久连接，
提供 publish() 方法将数据转发到指定主题。
支持 URI 中的 percent-encoded 凭据（如密码中的 @ 写为 %40）。
使用方式:
client = MQTTForwardClient(uri="mqtt://user:pass@host:port")
await client.start()
await client.publish("/topic", {"key": "value"})
await client.stop()
"""
def __init__(self, uri: str, topic: str = "/board_ping"):
self._uri = uri
self._topic = topic
self._client: MQTTClient | None = None
self._running = False
self._lock = asyncio.Lock()
# 预解析 URI 并解码凭据（amqtt 的 urlparse 不会自动 unquote）
parsed = urlparse(uri)
self._clean_uri = f"{parsed.scheme}://{parsed.hostname}:{parsed.port or 1883}"
self._username: str | None = unquote(parsed.username) if parsed.username else None
self._password: str | None = unquote(parsed.password) if parsed.password else None
async def start(self) -> None:
"""建立到外部 MQTT Broker 的连接"""
if self._running:
return
try:
self._client = MQTTClient(
config={
"auto_reconnect": True,
"reconnect_retries": -1, # 无限重连
"reconnect_max_interval": 30, # 最大重连间隔（秒）
"keep_alive": 60,
}
)
# 手动初始化 session 并设置解码后的凭据（绕过 amqtt urlparse 不解码 %40 的问题）
self._client.session = self._client._init_session(self._clean_uri)
if self._username:
self._client.session.username = self._username
if self._password:
self._client.session.password = self._password
await self._client._do_connect()
self._running = True
logger.info(f"[MQTTForward] 已连接到外部 Broker: {self._clean_uri}")
logger.info(f"[MQTTForward] 转发目标主题: {self._topic}")
except Exception as e:
logger.error(f"[MQTTForward] 连接外部 Broker 失败: {e}")
self._client = None
raise
async def stop(self) -> None:
"""断开与外部 MQTT Broker 的连接"""
if not self._running:
return
self._running = False
if self._client:
try:
await self._client.disconnect()
except Exception as e:
logger.warning(f"[MQTTForward] 断开连接时异常: {e}")
finally:
self._client = None
logger.info("[MQTTForward] 已断开外部 Broker 连接")
async def publish(self, payload: dict, topic: str | None = None) -> bool:
"""
发布消息到外部 MQTT Broker
Args:
payload: 要发送的 JSON 数据
topic: 目标主题，默认使用初始化时配置的主题
Returns:
是否发送成功
"""
if not self._running or self._client is None:
logger.warning("[MQTTForward] 客户端未连接，跳过转发")
return False
target = topic or self._topic
async with self._lock:
try:
data = json.dumps(payload, ensure_ascii=False, default=str)
await self._client.publish(target, data.encode("utf-8"), QOS_1)
logger.debug(f"[MQTTForward] 已转发到 {target}: {len(data)} 字节")
return True
except Exception as e:
logger.error(f"[MQTTForward] 发布失败 [{target}]: {e}")
return False
@property
def is_running(self) -> bool:
return self._running
aibox_sophon_adapter/router/board_ping.py
+5 -1
View File
@@ -27,7 +27,7 @@ async def board_ping(request: BoardPingRequest, worker_id: str = "") -> dict:
设备通过 MQTT 主题 /board_ping 定期（默认 5 秒）上报系统状态。
收到心跳后更新数据库中的设备记录，标记为在线。
"""
from ..global_var import db, heartbeat_timeout, stream_manager, link_path
from ..global_var import db, heartbeat_timeout, stream_manager, link_path, mqtt_forward
board_id = request.BoardId
@@ -93,6 +93,10 @@ async def board_ping(request: BoardPingRequest, worker_id: str = "") -> dict:
source_url = media["MediaUrl"].replace("/Link", link_path)
await stream_manager.start_stream(f"/{stream_id}", source_url)
# 转发心跳数据到外部 MQTT Broker
if mqtt_forward is not None:
await mqtt_forward.publish(ping_data)
logger.debug(
f"[Heartbeat] {board_id}: temp={request.BoardTemp}, "
f"disk={request.HostDisk.Used}/{request.HostDisk.Total} kB, "
aibox_sophon_adapter/routes.py
+31 -1
View File
@@ -19,6 +19,8 @@ from fastapi import APIRouter, UploadFile, File, Form, Request, Body, HTTPExcept
from fastapi.responses import FileResponse
from pydantic import ValidationError
from chongming_request import Client
from .models.video_models import VideoUploadResponse, VideoUploadResult
from .models.alarm_models import (
AlarmReportRequest,
@@ -30,7 +32,7 @@ logger = logging.getLogger(__name__)
# 自定义路由器，前缀与 config.toml 中的 router_prefix 保持一致
# 最终路径由网关 register_routes() 加上全局 prefix (/api/v1) 组合而成
router = APIRouter(prefix="/aibox", tags=["aibox_sophon_adapter"])
router = APIRouter(prefix="/aibox_sophon_adapter", tags=["aibox_sophon_adapter"])
@router.post(
@@ -168,6 +170,8 @@ async def alarm_report(
返回格式：{"Result": {"Code": 0, "Desc": ""}}
仅当 HTTP 200 + Code 非 0 时，盒子认为上报失败。
"""
zenoh_session = request.app.state.zenoh_session
# 手动 Pydantic 验证，失败时记录完整原始请求体
try:
body = AlarmReportRequest.model_validate(payload)
@@ -182,6 +186,27 @@ async def alarm_report(
# 获取完整报文（含 extra 字段，排除 None 值以生成整洁 JSON）
full_data = body.model_dump(exclude_none=True)
async with Client(session=zenoh_session) as client:
try:
result = await client.call(
worker_type="doubao",
subject="doubao.analyze",
payload={
"image_base64": full_data.get("ImageData", ""),
"prompt": full_data.get("Summary", ""),
},
timeout=30.0,
retries=1
)
analysis = result.get("result", "")
error = result.get("error", "")
if error:
logger.warning(f"Doubao 分析失败: {error}")
else:
logger.info(f"Doubao 分析结果: {analysis}")
except Exception as e:
logger.error(f"Doubao 分析调用失败: {type(e).__name__}: {e}")
board_id = full_data.get("BoardId", "")
board_ip = full_data.get("BoardIp", "")
alarm_id = full_data.get("AlarmId") or full_data.get("UniqueId", "")
@@ -228,6 +253,11 @@ async def alarm_report(
f"BoardId={board_id}, Summary={summary}, "
f"Media={media_name}, VideoFile={video_file}"
)
# 裁剪告警记录，保留最新 N 条
from .global_var import alarm_max_records, upload_root as _upload_root
if alarm_max_records > 0:
from .alarm_trim import trim_alarm_records
await trim_alarm_records(session, _upload_root, alarm_max_records)
else:
logger.warning(
f"[AlarmReport] 重复告警，已忽略: AlarmId={alarm_id}"