diff --git a/aibox_sophon_adapter/__init__.py b/aibox_sophon_adapter/__init__.py index 0e3b152..ae5054f 100644 --- a/aibox_sophon_adapter/__init__.py +++ b/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} 清理完毕") diff --git a/aibox_sophon_adapter/edge_app_controller.py b/aibox_sophon_adapter/edge_app_controller.py new file mode 100644 index 0000000..0a7e3d3 --- /dev/null +++ b/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, +} diff --git a/aibox_sophon_adapter/global_var.py b/aibox_sophon_adapter/global_var.py index 6310adf..dc0f7d8 100644 --- a/aibox_sophon_adapter/global_var.py +++ b/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", diff --git a/aibox_sophon_adapter/mqtt_forward.py b/aibox_sophon_adapter/mqtt_forward.py deleted file mode 100644 index 509e131..0000000 --- a/aibox_sophon_adapter/mqtt_forward.py +++ /dev/null @@ -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