""" AI Box Sophon Adapter — 全局变量 模块级共享资源管理。 """ from typing import Any, TYPE_CHECKING from chongming_database import Database from .models.orm import * from chongming_stream_manager import StreamManager if TYPE_CHECKING: from .mqtt_forward import MQTTForwardClient # 数据库实例 db: Database | None = None # 视频流管理器 stream_manager: StreamManager | None = None link_path: str | None = None # 外部 MQTT 转发客户端 mqtt_forward: "MQTTForwardClient | None" = None # 上传文件根目录(用于清理告警关联视频) upload_root: str = "./data/uploads" # 告警保留配置 alarm_max_records: int = 0 # 0 表示不限制 # 豆包 AI 分析开关 doubao_analysis_enabled: bool = True # 告警转发配置 alarm_forward_enabled: bool = False alarm_forward_url: str = "http://dg.datagoo.cn:60014/open-api/event-alarm/receive" alarm_forward_video_url: str = "http://dg.datagoo.cn:60014/open-api/event-alarm/receiveVideoUpload" alarm_forward_timeout: float = 10.0 # 心跳超时配置(秒) heartbeat_timeout: float = 15.0 # 心跳上报间隔(秒) heartbeat_interval: float = 5.0 def set_global_var(name: str, value: Any) -> None: """ 设置当前模块的全局变量。 Args: name: 变量名(字符串) value: 任意值 """ globals()[name] = value __all__ = [ "db", "heartbeat_timeout", "heartbeat_interval", "mqtt_forward", "upload_root", "alarm_max_records", "doubao_analysis_enabled", "alarm_forward_enabled", "alarm_forward_url", "alarm_forward_video_url", "alarm_forward_timeout", "set_global_var", ]