modify: mqtt 插件返回问题

aibox_sophon_adapter/__init__.py

@@ -282,12 +282,24 @@ async def create_workers(session, config, count=1, worker_manager=None):
     int_mqtt_cfg = config.get("internal_mqtt", {})
     int_mqtt_enabled = int_mqtt_cfg.get("enabled", False)
     int_mqtt_client = None
+    # 内部请求-响应主题（edge_app_controller 的 alg_media_fetch 命令使用）。
+    # 注意：勿用 chongming/{插件}/{操作} 保留命名空间——网关 MQTT Bridge 订阅了
+    # chongming/+/+，会拦截该命名空间并按 Worker RPC 路由，而 edge_app_controller
+    # 不是 @rpc_method（Worker 无对应 Queryable），导致 worker_manager 报
+    # "Worker ... 无回复，尝试重试"。请与对端应用（设备）约定一致的主题。
+    int_req_topic = int_mqtt_cfg.get(
+        "request_topic", "/internal/aibox_sophon_adapter/edge_app_controller"
+    )
+    int_resp_topic = int_mqtt_cfg.get(
+        "response_topic", "/internal/aibox_sophon_adapter/edge_app_controller_reply"
+    )
     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}"
+            f"[InternalMQTT] 配置已加载: enabled={int_mqtt_enabled}, uri={int_mqtt_uri}, "
+            f"request_topic={int_req_topic}, response_topic={int_resp_topic}"
         )
         int_mqtt_client = InternalMQTTClient(uri=int_mqtt_uri)
         try:
@@ -307,6 +319,8 @@ async def create_workers(session, config, count=1, worker_manager=None):
     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("internal_request_topic", int_req_topic)
+    set_global_var("internal_response_topic", int_resp_topic)
     set_global_var("upload_root", upload_root_path)
     set_global_var("alarm_max_records", alarm_max)
     set_global_var("doubao_analysis_enabled", doubao_enabled)

aibox_sophon_adapter/edge_app_controller.py

@@ -56,12 +56,12 @@ async def edge_app_controller(payload: dict, topic: str) -> dict | None:
     )
 
     if command is None:
-        return None
+        return _reply(-1, "缺少 command 字段")
 
     handler = _COMMANDS.get(str(command))
     if handler is None:
         logger.warning(f"[EdgeAppController] 未知命令: {command}")
-        return None
+        return _reply(-2, f"未知命令: {command}")
 
     try:
         data = await handler(payload)
@@ -73,7 +73,7 @@ async def edge_app_controller(payload: dict, topic: str) -> dict | None:
         return data
     except Exception as e:
         logger.error(f"[EdgeAppController] 命令 {command} 执行失败: {e}")
-        return None
+        return _reply(-3, f"命令执行异常: {e}")
 
 
 # ============================================================
@@ -115,19 +115,24 @@ async def _cmd_alg_media_fetch(payload: dict) -> dict | None:
     请求: {"Event": "/alg_media_fetch", "BoardId": "..."}
     返回: {"board_id": ..., "medias": [...], "tasks": [...]}
     """
-    from .global_var import db, internal_mqtt
+    from .global_var import (
+        db,
+        internal_mqtt,
+        internal_request_topic,
+        internal_response_topic,
+    )
 
     board_id = payload.get("BoardId")
     if not board_id:
-        return None
+        return _reply(-4, "缺少 BoardId 参数")
 
     print(f"[alg_media_fetch]Processing for board: {board_id}")
     print(f"[alg_media_fetch]Expected payload: {payload}")
 
     if db is None:
-        raise RuntimeError("数据库未初始化")
+        return _reply(-5, "数据库未初始化")
     if internal_mqtt is None:
-        raise RuntimeError("internal_mqtt 未初始化")
+        return _reply(-5, "internal_mqtt 未初始化")
 
     async with db.get_session() as session:
         from .models.orm import BoardHeartbeat
@@ -139,17 +144,24 @@ async def _cmd_alg_media_fetch(payload: dict) -> dict | None:
 
         if record is None:
             print(f"[alg_media_fetch]Board {board_id} not found in database.")
-            return None
+            return _reply(-4, f"设备 {board_id} 不在数据库中")
 
+        # 内部请求-响应主题：勿用 chongming/{插件}/{操作} 保留命名空间。
+        # 网关 MQTT Bridge 订阅了 chongming/+/+，会拦截该命名空间并按 Worker RPC
+        # 路由，而 edge_app_controller 不是 @rpc_method（Worker 无对应 Queryable），
+        # 会导致 worker_manager 报 "Worker ... 无回复，尝试重试"。
+        # 主题由 [workers.config.internal_mqtt] 配置，需与对端应用（设备）约定一致。
         resp = await internal_mqtt.request(
-            request_topic="/chongming/aibox_sophon_adapter/edge_app_controller",
+            request_topic=internal_request_topic,
-            response_topic="/chongming/aibox_sophon_adapter/edge_app_controller_reply",
+            response_topic=internal_response_topic,
             payload=payload,
             timeout=5
         )
 
         print(f"[alg_media_fetch]Response from internal_mqtt: {resp}")
 
+        if resp is None:
+            return _reply(-5, "获取媒体信息超时或失败")
         return resp
 
 async def _cmd_alg_task_fetch(payload: dict) -> dict | None:
@@ -163,13 +175,13 @@ async def _cmd_alg_task_fetch(payload: dict) -> dict | None:
 
     board_id = payload.get("BoardId")
     if not board_id:
-        return None
+        return _reply(-4, "缺少 BoardId 参数")
 
     print(f"[alg_task_fetch]Processing for board: {board_id}")
     print(f"[alg_task_fetch]Expected payload: {payload}")
 
     if db is None:
-        raise RuntimeError("数据库未初始化")
+        return _reply(-5, "数据库未初始化")
 
     async with db.get_session() as session:
         from .models.orm import BoardHeartbeat
@@ -181,7 +193,7 @@ async def _cmd_alg_task_fetch(payload: dict) -> dict | None:
 
         if record is None:
             print(f"[alg_task_fetch]Board {board_id} not found in database.")
-            return None
+            return _reply(-4, f"设备 {board_id} 不在数据库中")
 
         import json
         ping_data = {}

aibox_sophon_adapter/global_var.py

@@ -30,6 +30,11 @@ mqtt_subscribe: "MQTTSubscribeClient | None" = None
 # 内部 MQTT 请求-响应客户端（handler 内对任意应用进程发起请求-响应）
 internal_mqtt: "InternalMQTTClient | None" = None
 
+# 内部请求-响应主题（edge_app_controller 的 alg_media_fetch 命令使用）
+# 注意：勿用 chongming/{插件}/{操作} 保留命名空间（网关 Bridge 会拦截并按 Worker RPC 路由）
+internal_request_topic: str = "/internal/aibox_sophon_adapter/edge_app_controller"
+internal_response_topic: str = "/internal/aibox_sophon_adapter/edge_app_controller_reply"
+
 # 上传文件根目录（用于清理告警关联视频）
 upload_root: str = "./data/uploads"
 
@@ -70,6 +75,8 @@ __all__ = [
     "mqtt_forward",
     "mqtt_subscribe",
     "internal_mqtt",
+    "internal_request_topic",
+    "internal_response_topic",
     "upload_root",
     "alarm_max_records",
     "doubao_analysis_enabled",

example/__init__.py

@@ -33,8 +33,17 @@ auto_bind_models("example")
 # 详见 src/chongming/app/__init__.py 的 lifespan。
 
 
+# 幂等标记：防止 create_workers 被重复调用时重复初始化数据库等资源
+_initialized = False
+
+
 async def create_workers(session, config, count=1, worker_manager=None):
     """插件入口 — 委托给 chongming_worker 运行时创建 Worker 协程"""
+    global _initialized
+    if _initialized:
+        print("[example] create_workers 重复调用，跳过重复初始化")
+        return _create_workers(session, config, count, worker_manager)
+
     from .global_var import set_global_var
     from chongming_database import Database
 
@@ -44,11 +53,16 @@ async def create_workers(session, config, count=1, worker_manager=None):
     await db.initialize()
     set_global_var("db", db)
 
+    _initialized = True
     return _create_workers(session, config, count, worker_manager)
 
+
 async def cleanup(worker_item: str):
-    from .global_var import db
+    global _initialized
+    from .global_var import db, set_global_var
 
     if db is not None:
         await db.dispose()
+        set_global_var("db", None)
+    _initialized = False
     print(f"{worker_item} 清理完毕")

example/global_var.py

@@ -1,22 +1,57 @@
+"""Example 插件全局运行时状态 — 统一容器
+
+约定:
+  - 模块只做"声明默认值"，不产生任何导入期副作用
+  - 所有值统一在 create_workers 中通过 set_global_var 显式写入
+  - set_global_var 仅允许写入 _ALLOWED 白名单，拼写错误会立即抛错
+  - require() 用于取"必须已初始化"的值，未初始化时抛清晰错误
+"""
 from typing import Any
-from .models import *
+
 from chongming_database import Database
 
+# 允许写入的全局变量白名单（防止拼写错误静默创建无用的新变量）
+_ALLOWED = frozenset({"db"})
 
+# 显式声明所有运行时状态（默认 None，运行时由 create_workers 写入）
 db: Database | None = None
 
 
 def set_global_var(name: str, value: Any) -> None:
     """
-    设置当前模块的全局变量。
+    白名单校验版 setter — 设置模块级运行时状态。
+
+    拼错变量名会立即抛 KeyError，而不是静默创建一个无用的新变量。
 
     Args:
-        name: 变量名（字符串）
+        name: 必须在 _ALLOWED 白名单中的变量名
         value: 任意值
     """
+    if name not in _ALLOWED:
+        raise KeyError(
+            f"global_var: 未知变量 '{name}'（允许: {sorted(_ALLOWED)}）"
+        )
     globals()[name] = value
 
 
-__all__ = [
+def require(name: str) -> Any:
-    "set_global_var",
+    """
-]
+    获取必须已初始化的全局变量，未初始化时抛清晰错误。
+
+    Args:
+        name: 必须在 _ALLOWED 白名单中的变量名
+
+    Returns:
+        对应全局变量的值
+    """
+    if name not in _ALLOWED:
+        raise KeyError(f"global_var: 未知变量 '{name}'（允许: {sorted(_ALLOWED)}）")
+    value = globals().get(name)
+    if value is None:
+        raise RuntimeError(
+            f"global_var: '{name}' 未初始化 —— 请确认插件 create_workers 已执行且配置正确"
+        )
+    return value
+
+
+__all__ = sorted(_ALLOWED) + ["set_global_var", "require"]