From 5e5dee83dfe3e545628baba386cd0c2c607671df Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Thu, 20 Aug 2026 16:51:00 +0800 Subject: [PATCH] =?UTF-8?q?modify:=20cli=E5=91=BD=E4=BB=A4=E6=9B=B4?= =?UTF-8?q?=E6=96=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 4 +- aibox_sophon_adapter/README.md | 2 +- aibox_sophon_adapter/__init__.py | 9 +++ aibox_sophon_adapter/global_var.py | 4 ++ aibox_sophon_adapter/router/board_ping.py | 75 +++++++++++++---------- aibox_sophon_adapter/routes.py | 2 +- 6 files changed, 61 insertions(+), 35 deletions(-) diff --git a/README.md b/README.md index 8f355ae..3c18e9c 100644 --- a/README.md +++ b/README.md @@ -102,7 +102,7 @@ message = { type = "str", default = "" } ### 第四步：生成 Pydantic 模型 ```bash -uv run python -m chongming.scripts.gen_models +uv run gen_models ``` 自动在 `src/plugins//models/` 下生成 `pydantic.py` 和 `__init__.py`。 @@ -209,7 +209,7 @@ from .routes import router # 网关通过 getattr 自动发现 ## 为现有插件添加新端点 1. 创建 `config/workers//.toml` -2. 运行 `uv run python -m chongming.scripts.gen_models` +2. 运行 `uv run gen_models` 3. 创建 `router/.py`，引用生成的模型 4. 更新 `router/__init__.py` 5. 重启服务 diff --git a/aibox_sophon_adapter/README.md b/aibox_sophon_adapter/README.md index dadbad2..854fe99 100644 --- a/aibox_sophon_adapter/README.md +++ b/aibox_sophon_adapter/README.md @@ -345,7 +345,7 @@ tags = ["aibox_sophon_adapter", "sophon", "adapter"] ### 新增 RPC 端点 1. 在 `config/workers/aibox_sophon_adapter/` 下创建 `<操作>.toml`（参考 `list_devices.toml`） -2. 运行 `uv run python -m chongming.scripts.gen_models` 生成模型 +2. 运行 `uv run gen_models` 生成模型 3. 在 `src/plugins/aibox_sophon_adapter/router/` 下创建 `<操作>.py`，使用 `@rpc_method` 4. 更新 `router/__init__.py` 的导入与 `__all__` 5. 重启服务 diff --git a/aibox_sophon_adapter/__init__.py b/aibox_sophon_adapter/__init__.py index 2686af8..202ac9a 100644 --- a/aibox_sophon_adapter/__init__.py +++ b/aibox_sophon_adapter/__init__.py @@ -21,6 +21,7 @@ MQTT 客户端启动块已抽取到 mqtt_clients 能力组件。 """ import asyncio import logging +import shutil from datetime import datetime from pathlib import Path @@ -156,6 +157,14 @@ class AiboxSophonAdapterPlugin(Plugin): set_global_var("alarm_forward_video_url", alarm_fwd_video_url) set_global_var("alarm_forward_timeout", alarm_fwd_timeout) + # 探测 ffmpeg（RTSP 拉流依赖）；缺失时心跳仍正常入库，仅跳过拉流 + ffmpeg_available = shutil.which("ffmpeg") is not None + set_global_var("ffmpeg_available", ffmpeg_available) + if not ffmpeg_available: + logger.warning( + "[aibox_sophon_adapter] 未检测到 ffmpeg，设备正常通道的 RTSP 拉流将跳过" + ) + logger.info( "[aibox_sophon_adapter] 已初始化: " f"心跳间隔={hb_interval}s, 超时={hb_timeout}s, 检查间隔={check_interval}s" diff --git a/aibox_sophon_adapter/global_var.py b/aibox_sophon_adapter/global_var.py index cc277c3..3cb8ab4 100644 --- a/aibox_sophon_adapter/global_var.py +++ b/aibox_sophon_adapter/global_var.py @@ -57,6 +57,9 @@ heartbeat_timeout: float = 15.0 # 心跳上报间隔（秒） heartbeat_interval: float = 5.0 +# ffmpeg 是否可用（on_init 时探测；不可用时跳过 RTSP 拉流，不阻断心跳） +ffmpeg_available: bool = True + def set_global_var(name: str, value: Any) -> None: """ @@ -85,5 +88,6 @@ __all__ = [ "alarm_forward_url", "alarm_forward_video_url", "alarm_forward_timeout", + "ffmpeg_available", "set_global_var", ] diff --git a/aibox_sophon_adapter/router/board_ping.py b/aibox_sophon_adapter/router/board_ping.py index 650932d..492d912 100644 --- a/aibox_sophon_adapter/router/board_ping.py +++ b/aibox_sophon_adapter/router/board_ping.py @@ -27,10 +27,14 @@ 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, mqtt_forward + from ..global_var import ( + db, heartbeat_timeout, stream_manager, link_path, mqtt_forward, ffmpeg_available, + ) board_id = request.BoardId + ping_json = request.model_dump_json() + # ---- 心跳落库（核心步骤，失败才视为心跳处理失败） ---- try: if db is None: raise RuntimeError("Database not initialized") @@ -43,7 +47,6 @@ async def board_ping(request: BoardPingRequest, worker_id: str = "") -> dict: record: BoardHeartbeat | None = result.scalar_one_or_none() now = datetime.now() - ping_json = request.model_dump_json() if record is None: # 新设备，创建记录 @@ -81,37 +84,47 @@ async def board_ping(request: BoardPingRequest, worker_id: str = "") -> dict: record.updated_at = now await session.commit() - - if stream_manager is None: - raise RuntimeError("StreamManager not initialized") - - ping_data = json.loads(ping_json) - medias = ping_data.get("Medias", {}) - for media in medias: - if "正常" in media.get("MediaStatus", {}).get("label", ""): - stream_id = media["MediaName"] - 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: - success = await mqtt_forward.publish(ping_data) - if not success: - logger.warning("[MQTTForward] 心跳转发失败") - else: - logger.warning(f"[MQTTForward] 心跳已转发 ({len(ping_data)} keys)") - else: - logger.warning("[MQTTForward] mqtt_forward 客户端为 None，跳过转发") - - 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)} + # ---- 以下为可选能力：失败只告警，不阻断心跳 ---- + ping_data = json.loads(ping_json) + + # RTSP 拉流（依赖 ffmpeg；缺失时跳过并在启动日志提示） + medias = ping_data.get("Medias", {}) + pending = [ + media for media in medias + if "正常" in media.get("MediaStatus", {}).get("label", "") + ] + if pending and stream_manager is None: + logger.warning("[Heartbeat] StreamManager 未初始化，跳过 RTSP 拉流") + elif pending and not ffmpeg_available: + logger.warning("[Heartbeat] ffmpeg 不可用，跳过 RTSP 拉流（安装 ffmpeg 后恢复）") + else: + for media in pending: + stream_id = media["MediaName"] + source_url = media["MediaUrl"].replace("/Link", link_path) + try: + await stream_manager.start_stream(f"/{stream_id}", source_url) + except Exception as e: + logger.warning(f"[Heartbeat] 通道 {stream_id} 拉流失败: {e}") + + # 转发心跳数据到外部 MQTT Broker + if mqtt_forward is not None: + success = await mqtt_forward.publish(ping_data) + if not success: + logger.warning("[MQTTForward] 心跳转发失败") + else: + logger.warning(f"[MQTTForward] 心跳已转发 ({len(ping_data)} keys)") + else: + logger.warning("[MQTTForward] mqtt_forward 客户端为 None，跳过转发") + + 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)}" + ) + return {"received": True, "board_id": board_id} diff --git a/aibox_sophon_adapter/routes.py b/aibox_sophon_adapter/routes.py index 936330b..44746c2 100644 --- a/aibox_sophon_adapter/routes.py +++ b/aibox_sophon_adapter/routes.py @@ -94,7 +94,7 @@ async def _analyze_with_doubao(zenoh_session, full_data: dict) -> bool: logger.info("[AlarmReport] 豆包分析已关闭 (doubao_analysis.enabled=false)") return True - async with Client(session=zenoh_session) as client: + async with Client(session=zenoh_session, logger=logger) as client: try: result = await client.call( worker_type="doubao",