From e43c25f5dc71b28df308211a0f04fe042a56212f Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Mon, 3 Aug 2026 15:06:29 +0800 Subject: [PATCH] =?UTF-8?q?modify:=20mqtt=E5=BF=83=E8=B7=B3=E8=BD=AC?= =?UTF-8?q?=E5=8F=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- aibox_sophon_adapter/__init__.py | 32 +++++- aibox_sophon_adapter/alarm_trim.py | 94 +++++++++++++++++ aibox_sophon_adapter/global_var.py | 17 +++- aibox_sophon_adapter/mqtt_forward.py | 119 ++++++++++++++++++++++ aibox_sophon_adapter/router/board_ping.py | 6 +- aibox_sophon_adapter/routes.py | 32 +++++- 6 files changed, 296 insertions(+), 4 deletions(-) create mode 100644 aibox_sophon_adapter/alarm_trim.py create mode 100644 aibox_sophon_adapter/mqtt_forward.py diff --git a/aibox_sophon_adapter/__init__.py b/aibox_sophon_adapter/__init__.py index 7c67e70..2c1c8d0 100644 --- a/aibox_sophon_adapter/__init__.py +++ b/aibox_sophon_adapter/__init__.py @@ -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} 清理完毕") diff --git a/aibox_sophon_adapter/alarm_trim.py b/aibox_sophon_adapter/alarm_trim.py new file mode 100644 index 0000000..0d27c00 --- /dev/null +++ b/aibox_sophon_adapter/alarm_trim.py @@ -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}") diff --git a/aibox_sophon_adapter/global_var.py b/aibox_sophon_adapter/global_var.py index 8c8f95e..c9c1c21 100644 --- a/aibox_sophon_adapter/global_var.py +++ b/aibox_sophon_adapter/global_var.py @@ -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", ] diff --git a/aibox_sophon_adapter/mqtt_forward.py b/aibox_sophon_adapter/mqtt_forward.py new file mode 100644 index 0000000..6765de0 --- /dev/null +++ b/aibox_sophon_adapter/mqtt_forward.py @@ -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 diff --git a/aibox_sophon_adapter/router/board_ping.py b/aibox_sophon_adapter/router/board_ping.py index 2764b0e..6ef60cb 100644 --- a/aibox_sophon_adapter/router/board_ping.py +++ b/aibox_sophon_adapter/router/board_ping.py @@ -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, " diff --git a/aibox_sophon_adapter/routes.py b/aibox_sophon_adapter/routes.py index e6c6c49..cec91c2 100644 --- a/aibox_sophon_adapter/routes.py +++ b/aibox_sophon_adapter/routes.py @@ -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}"