diff --git a/aibox_sophon_adapter/__init__.py b/aibox_sophon_adapter/__init__.py index ae5054f..d946541 100644 --- a/aibox_sophon_adapter/__init__.py +++ b/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) diff --git a/aibox_sophon_adapter/edge_app_controller.py b/aibox_sophon_adapter/edge_app_controller.py index c19987f..82de95b 100644 --- a/aibox_sophon_adapter/edge_app_controller.py +++ b/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", - response_topic="/chongming/aibox_sophon_adapter/edge_app_controller_reply", + request_topic=internal_request_topic, + 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 = {} diff --git a/aibox_sophon_adapter/global_var.py b/aibox_sophon_adapter/global_var.py index dc0f7d8..9054514 100644 --- a/aibox_sophon_adapter/global_var.py +++ b/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", diff --git a/example/__init__.py b/example/__init__.py index 6771bde..8dea313 100644 --- a/example/__init__.py +++ b/example/__init__.py @@ -33,22 +33,36 @@ 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 - + db_path = Path(config.get("data_dir", "./data")) / "example.db" db = Database(db_path) 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} 清理完毕") diff --git a/example/global_var.py b/example/global_var.py index 212d3bc..22fc6c8 100644 --- a/example/global_var.py +++ b/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__ = [ - "set_global_var", -] + +def require(name: str) -> Any: + """ + 获取必须已初始化的全局变量，未初始化时抛清晰错误。 + + 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"]