modify: chongming-mqtt 应用

aibox_sophon_adapter/__init__.py

@@ -20,6 +20,7 @@ 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
@@ -27,8 +28,102 @@ 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")
 
@@ -138,7 +233,7 @@ async def create_workers(session, config, count=1, worker_manager=None):
     mqtt_fwd_enabled = mqtt_fwd.get("enabled", False)
     mqtt_fwd_client = None
     if mqtt_fwd_enabled:
-        from .mqtt_forward import MQTTForwardClient
+        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}")
@@ -153,12 +248,65 @@ async def create_workers(session, config, count=1, worker_manager=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
+    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}"
+        )
+        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("upload_root", upload_root_path)
     set_global_var("alarm_max_records", alarm_max)
     set_global_var("doubao_analysis_enabled", doubao_enabled)
@@ -186,7 +334,9 @@ async def cleanup(worker_item: str):
 
     在 Worker 关闭时调用，关闭数据库连接和后台任务。
     """
-    from .global_var import db, stream_manager, mqtt_forward
+    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():
@@ -206,4 +356,10 @@ async def cleanup(worker_item: str):
     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} 清理完毕")

aibox_sophon_adapter/edge_app_controller.py

@@ -0,0 +1,207 @@
+"""
+edge_app_controller — 外部 MQTT 请求处理（请求-响应协议）
+
+订阅主题:  /edge_app_controller
+回复主题:  /edge_app_controller_reply
+
+请求消息 (JSON): {"command": "<命令名>", ...业务字段}
+                兼容协议: {"Event": "/alg_media_fetch", "BoardId": "..."}（Event 取最后一段作为命令）
+回复消息 (JSON): {"code": 0, "msg": "ok", "data": {...}}
+  - code: 0 成功；非 0 失败
+      -1 缺少 command 字段
+      -2 未知命令
+      -3 命令执行异常
+
+不回复约定:
+  命令处理函数返回 None 时，该命令只处理、不回复（静默命令，如心跳/日志类）。
+  适用于按命令/Event 控制是否回复，无需修改配置。
+
+命令通过 _COMMANDS 注册表分发，新增命令只需在注册表登记一个 async 处理函数。
+"""
+import logging
+from typing import Any, Callable, Coroutine
+
+logger = logging.getLogger(__name__)
+
+
+def _reply(code: int, msg: str, data: Any = None) -> dict:
+    """统一回复封装"""
+    return {"code": code, "msg": msg, "data": data}
+
+
+async def edge_app_controller(payload: dict, topic: str) -> dict | None:
+    """
+    处理来自外部 Broker 的边缘应用控制器请求（订阅 /edge_app_controller）
+
+    Args:
+        payload: 请求消息（已解析为 dict），期望含 command 字段
+        topic:   实际命中的主题
+
+    Returns:
+        回复消息（统一封装 code / msg / data），由订阅客户端发布到 reply_topic；
+        命令函数返回 None 时返回 None（不回复）。
+    """
+    command = (
+        payload.get("command")
+        or payload.get("cmd")
+        or payload.get("action")
+        or payload.get("Event")
+    )
+    # Event 可能是路径形式（如 /alg_media_fetch），取最后一段作为命令名
+    if isinstance(command, str) and command.startswith("/"):
+        command = command.rsplit("/", 1)[-1]
+
+    logger.info(
+        f"[EdgeAppController] 收到请求 topic={topic}, command={command}, payload={payload}"
+    )
+
+    if command is None:
+        return None
+
+    handler = _COMMANDS.get(str(command))
+    if handler is None:
+        logger.warning(f"[EdgeAppController] 未知命令: {command}")
+        return None
+
+    try:
+        data = await handler(payload)
+        # 命令函数返回 None 表示该命令只处理、不回复（如心跳/日志类静默命令）
+        if data is None:
+            logger.info(f"[EdgeAppController] 命令 {command} 处理完成（配置为不回复）")
+            return None
+        logger.info(f"[EdgeAppController] 命令 {command} 执行成功")
+        return data
+    except Exception as e:
+        logger.error(f"[EdgeAppController] 命令 {command} 执行失败: {e}")
+        return None
+
+
+# ============================================================
+# 内置命令（示例）
+# ============================================================
+
+async def _cmd_ping(payload: dict) -> dict | None:
+    """存活探测：返回服务信息"""
+    return {"service": "aibox_sophon_adapter", "alive": True}
+
+
+async def _cmd_ack(payload: dict) -> None:
+    """
+    确认命令（示例）：仅记录日志，不回复
+
+    返回 None 表示不回复 —— 需要"某些命令静默处理"时，让命令函数返回 None 即可。
+    """
+    logger.info(f"[EdgeAppController] 命令 ack 已收到（不回复），payload={payload}")
+    return None
+
+
+async def _cmd_query_status(payload: dict) -> dict | None:
+    """
+    查询边缘应用/设备状态
+
+    复用插件已有的设备查询逻辑（router.list_devices），返回设备在线状态列表。
+    后续可按需扩展：查询指定应用运行状态、通道状态等。
+    """
+    from .router.list_devices import list_devices
+
+    result = await list_devices()
+    return result
+
+
+async def _cmd_alg_media_fetch(payload: dict) -> dict | None:
+    """
+    算法媒体抓取：按 BoardId 返回设备最近一次心跳中的算法/通道（媒体）信息
+
+    请求: {"Event": "/alg_media_fetch", "BoardId": "..."}
+    返回: {"board_id": ..., "medias": [...], "tasks": [...]}
+    """
+    from .global_var import db
+
+    board_id = payload.get("BoardId")
+    if not board_id:
+        return None
+
+    print(f"[alg_media_fetch]Processing for board: {board_id}")
+    print(f"[alg_media_fetch]Expected payload: {payload}")
+
+    if db is None:
+        raise RuntimeError("数据库未初始化")
+
+    async with db.get_session() as session:
+        from .models.orm import BoardHeartbeat
+        from sqlalchemy import select
+
+        stmt = select(BoardHeartbeat).where(BoardHeartbeat.board_id == board_id) # type: ignore
+        result = await session.execute(stmt)
+        record: BoardHeartbeat | None = result.scalar_one_or_none()
+
+        if record is None:
+            print(f"[alg_media_fetch]Board {board_id} not found in database.")
+            return None
+
+        import json
+        ping_data = {}
+        try:
+            ping_data = json.loads(record.ping_data) if record.ping_data else {}
+        except json.JSONDecodeError:
+            pass
+
+        return {
+            "board_id": board_id,
+            "board_ip": record.board_ip,
+            "medias": ping_data.get("Medias", []),
+            "tasks": ping_data.get("Tasks", []),
+        }
+
+async def _cmd_alg_task_fetch(payload: dict) -> dict | None:
+    """
+    算法任务抓取：按 BoardId 返回设备最近一次心跳中的算法任务信息
+
+    请求: {"Event": "/alg_task_fetch", "BoardId": "..."}
+    返回: {"board_id": ..., "tasks": [...]}
+    """
+    from .global_var import db
+
+    board_id = payload.get("BoardId")
+    if not board_id:
+        return None
+
+    print(f"[alg_task_fetch]Processing for board: {board_id}")
+    print(f"[alg_task_fetch]Expected payload: {payload}")
+
+    if db is None:
+        raise RuntimeError("数据库未初始化")
+
+    async with db.get_session() as session:
+        from .models.orm import BoardHeartbeat
+        from sqlalchemy import select
+
+        stmt = select(BoardHeartbeat).where(BoardHeartbeat.board_id == board_id) # type: ignore
+        result = await session.execute(stmt)
+        record: BoardHeartbeat | None = result.scalar_one_or_none()
+
+        if record is None:
+            print(f"[alg_task_fetch]Board {board_id} not found in database.")
+            return None
+
+        import json
+        ping_data = {}
+        try:
+            ping_data = json.loads(record.ping_data) if record.ping_data else {}
+        except json.JSONDecodeError:
+            pass
+
+        return {
+            "board_id": board_id,
+            "tasks": ping_data.get("Tasks", []),
+        }
+
+
+# 命令注册表：command -> async handler(payload) -> data（返回 None 表示不回复）
+_COMMANDS: dict[str, Callable[[dict], Coroutine[Any, Any, Any]]] = {
+    "ping": _cmd_ping,
+    "ack": _cmd_ack,
+    "query_status": _cmd_query_status,
+    "alg_media_fetch": _cmd_alg_media_fetch,
+    "alg_task_fetch": _cmd_alg_task_fetch,
+}

aibox_sophon_adapter/global_var.py

@@ -10,7 +10,9 @@ from .models.orm import *
 from chongming_stream_manager import StreamManager
 
 if TYPE_CHECKING:
-    from .mqtt_forward import MQTTForwardClient
+    from chongming_mqtt.mqtt_forward import MQTTForwardClient
+    from chongming_mqtt.mqtt_subscribe import MQTTSubscribeClient
+    from chongming_mqtt.internal_mqtt import InternalMQTTClient
 
 # 数据库实例
 db: Database | None = None
@@ -22,6 +24,12 @@ link_path: str | None = None
 # 外部 MQTT 转发客户端
 mqtt_forward: "MQTTForwardClient | None" = None
 
+# 外部 MQTT 订阅客户端
+mqtt_subscribe: "MQTTSubscribeClient | None" = None
+
+# 内部 MQTT 请求-响应客户端（handler 内对任意应用进程发起请求-响应）
+internal_mqtt: "InternalMQTTClient | None" = None
+
 # 上传文件根目录（用于清理告警关联视频）
 upload_root: str = "./data/uploads"
 
@@ -60,6 +68,8 @@ __all__ = [
     "heartbeat_timeout",
     "heartbeat_interval",
     "mqtt_forward",
+    "mqtt_subscribe",
+    "internal_mqtt",
     "upload_root",
     "alarm_max_records",
     "doubao_analysis_enabled",

aibox_sophon_adapter/mqtt_forward.py

@@ -1,120 +0,0 @@
-"""
-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,
-                    "connection_timeout": 10,        # 连接超时（秒），避免永久挂起
-                }
-            )
-            # 手动初始化 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