239 lines
10 KiB
Python
239 lines
10 KiB
Python
"""
|
||
AI Box Sophon Adapter 插件
|
||
|
||
基于 MQTT 的 AI Box 设备心跳适配器，实现：
|
||
1. 通过 MQTT 主题接收设备心跳上报
|
||
2. 解析设备系统参数（温度、硬盘、内存、通道、任务、TPU 等）
|
||
3. 持久化心跳数据到 SQLite 数据库
|
||
4. 提供 HTTP API 查询设备在线状态和详情
|
||
5. 后台定时检查离线设备，触发异常告警
|
||
6. 视频上传接口 — 盒子告警时录制视频上传（FORM-DATA）
|
||
7. 告警上报接口 — 违规事件 JSON 报文字动解析入库
|
||
8. 告警查询接口 — 告警列表/详情查询
|
||
9. IN2 任务启停开关 — DI 输入有电启动全部任务，断电停止全部任务
|
||
|
||
心跳超时机制:
|
||
- 设备按配置间隔（默认 5 秒）上报心跳
|
||
- 服务端每 check_interval 秒检查一次
|
||
- 超过 heartbeat_timeout 秒未收到心跳的设备标记为离线
|
||
|
||
生命周期: 基于 Plugin 基类统一管理（数据库/后台任务/清理钩子由基类处理），
|
||
MQTT 客户端启动块已抽取到 mqtt_clients 能力组件。
|
||
"""
|
||
import asyncio
|
||
import logging
|
||
import shutil
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
from chongming_worker.plugin import Plugin
|
||
|
||
from .router import * # noqa: F401,F403 触发 @rpc_method 注册
|
||
from .routes import router # noqa: F401 自定义 FastAPI 路由（网关通过包属性自动发现）
|
||
from .edge_app_controller import edge_app_controller # noqa: F401
|
||
from .mqtt_clients import start_mqtt_forward, start_mqtt_subscribe, start_internal_mqtt
|
||
|
||
# 模块加载时即注册 SQLModel 元数据 —— 基类 _bootstrap 建表（create_all）只创建
|
||
# 已导入的表，而 global_var 的导入是懒加载的（on_init 才执行），若拖到建表之后
|
||
# 会导致全新库缺少 board_heartbeat / alarm_record 表（no such table）。
|
||
from .models.orm import BoardHeartbeat, AlarmRecord # noqa: F401
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
async def _check_offline_devices(interval: float):
|
||
"""
|
||
后台任务：定期检查离线设备
|
||
|
||
每隔 interval 秒扫描数据库，将超过 heartbeat_timeout 秒
|
||
未收到心跳的设备标记为离线，并输出告警日志。
|
||
"""
|
||
from .global_var import db, heartbeat_timeout
|
||
|
||
while True:
|
||
await asyncio.sleep(interval)
|
||
try:
|
||
if db is None:
|
||
continue
|
||
|
||
async with db.get_session() as session:
|
||
from .models.orm import BoardHeartbeat
|
||
from sqlalchemy import select
|
||
|
||
stmt = select(BoardHeartbeat).where(BoardHeartbeat.is_online) # type: ignore
|
||
result = await session.execute(stmt)
|
||
records = result.scalars().all()
|
||
|
||
now = datetime.now()
|
||
offline_count = 0
|
||
for record in records:
|
||
elapsed = (now - record.last_ping_time).total_seconds()
|
||
if elapsed > heartbeat_timeout:
|
||
record.is_online = False
|
||
record.updated_at = now
|
||
offline_count += 1
|
||
logger.warning(
|
||
f"[Offline] 设备 {record.board_id} (IP: {record.board_ip}) "
|
||
f"已离线 {elapsed:.0f} 秒（超时阈值: {heartbeat_timeout}s）"
|
||
)
|
||
|
||
if offline_count > 0:
|
||
await session.commit()
|
||
logger.info(f"[Offline] 本轮标记 {offline_count} 个设备为离线")
|
||
|
||
except Exception as e:
|
||
logger.error(f"[OfflineChecker] 检查异常: {e}")
|
||
|
||
|
||
class AiboxSophonAdapterPlugin(Plugin):
|
||
"""AI Box 适配插件 — 统一生命周期。"""
|
||
|
||
type = "aibox_sophon_adapter"
|
||
db_name = "aibox_sophon_adapter.db" # 自动在 data_dir 下建 SQLite 库
|
||
auto_bind = True # 首次初始化时自动 auto_bind_models
|
||
|
||
async def on_init(self):
|
||
"""首次 create_workers 时调用一次：初始化数据库客户端、流控制器、后台任务。"""
|
||
config = self.config
|
||
from .global_var import set_global_var, heartbeat_timeout as _default_timeout
|
||
from chongming_stream_manager import StreamManager
|
||
|
||
# 所有运行时配置均从 [workers.config] 及其子节读取
|
||
data_dir = Path(config.get("data_dir", "./data"))
|
||
mediamtx_addr = config.get("mediamtx_addr", "rtsp://127.0.0.1:8554")
|
||
link_path = config.get("link_path", "/data/bm-app")
|
||
|
||
# 心跳配置
|
||
heartbeat = config.get("heartbeat", {})
|
||
hb_interval = heartbeat.get("interval", 5)
|
||
hb_timeout = heartbeat.get("timeout", _default_timeout)
|
||
check_interval = heartbeat.get("check_interval", 10)
|
||
|
||
# 告警保留配置
|
||
alarm_retention = config.get("alarm_retention", {})
|
||
alarm_max = alarm_retention.get("max_records", 0)
|
||
|
||
# 豆包 AI 分析开关
|
||
doubao_cfg = config.get("doubao_analysis", {})
|
||
doubao_enabled = doubao_cfg.get("enabled", True)
|
||
|
||
# ip 音柱 TTS 播放开关
|
||
ip_sound_column_cfg = config.get("ip_sound_column", {})
|
||
ip_sound_column_enabled = ip_sound_column_cfg.get("enabled", True)
|
||
ip_sound_column_volume = ip_sound_column_cfg.get("volume", 50) # 播放音量 0-100
|
||
ip_sound_column_tts = ip_sound_column_cfg.get("tts", {}) # 播放内容（key: Summary，value: tts 文本）
|
||
|
||
# 告警转发配置
|
||
alarm_fwd_cfg = config.get("alarm_forward", {})
|
||
alarm_fwd_enabled = alarm_fwd_cfg.get("enabled", False)
|
||
alarm_fwd_url = alarm_fwd_cfg.get("url", "http://dg.datagoo.cn:60014/open-api/event-alarm/receive")
|
||
alarm_fwd_video_url = alarm_fwd_cfg.get("video_upload_url", "http://dg.datagoo.cn:60014/open-api/event-alarm/receiveVideoUpload")
|
||
alarm_fwd_timeout = alarm_fwd_cfg.get("timeout", 10.0)
|
||
|
||
# IN2 数字输入控制任务启停
|
||
switch_cfg = config.get("task_switch", {})
|
||
|
||
# 数据库已由基类初始化（data_dir/aibox_sophon_adapter.db）
|
||
db = self.db
|
||
|
||
# 初始化视频流控制器
|
||
stream_manager = StreamManager(mediamtx_addr)
|
||
|
||
# upload 根目录（用于清理告警关联视频）
|
||
upload_root_path = str(data_dir / "uploads")
|
||
|
||
# MQTT 客户端能力组件（启动失败降级为 None，仅警告）
|
||
mqtt_fwd_client = await start_mqtt_forward(config.get("mqtt_forward", {}))
|
||
mqtt_sub_client = await start_mqtt_subscribe(config.get("mqtt_subscribe", {}))
|
||
int_mqtt_client, int_req_topic, int_resp_topic = await start_internal_mqtt(
|
||
config.get("internal_mqtt", {})
|
||
)
|
||
|
||
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("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)
|
||
set_global_var("ip_sound_column_enabled", ip_sound_column_enabled)
|
||
set_global_var("ip_sound_column_volume", ip_sound_column_volume)
|
||
set_global_var("ip_sound_column_tts", ip_sound_column_tts)
|
||
set_global_var("alarm_forward_enabled", alarm_fwd_enabled)
|
||
set_global_var("alarm_forward_url", alarm_fwd_url)
|
||
set_global_var("alarm_forward_video_url", alarm_fwd_video_url)
|
||
set_global_var("alarm_forward_timeout", alarm_fwd_timeout)
|
||
|
||
# 探测 ffmpeg（RTSP 拉流依赖）；缺失时心跳仍正常入库，仅跳过拉流
|
||
ffmpeg_available = shutil.which("ffmpeg") is not None
|
||
set_global_var("ffmpeg_available", ffmpeg_available)
|
||
if not ffmpeg_available:
|
||
logger.warning(
|
||
"[aibox_sophon_adapter] 未检测到 ffmpeg，设备正常通道的 RTSP 拉流将跳过"
|
||
)
|
||
|
||
# IN2 任务启停开关（DI 有电 → 启动全部任务；断电 → 停止全部任务）
|
||
# 找不到 IN2 状态文件（非盒子环境）时降级为 None，仅警告
|
||
task_switch = None
|
||
if switch_cfg.get("enabled", False):
|
||
from .task_switch import In2TaskSwitch
|
||
|
||
task_switch = In2TaskSwitch(
|
||
api_base=switch_cfg.get("api_base", "http://127.0.0.1"),
|
||
start_command=switch_cfg.get("start_command", 1),
|
||
stop_command=switch_cfg.get("stop_command", 0),
|
||
poll_interval=switch_cfg.get("poll_interval", 0.05),
|
||
debounce=switch_cfg.get("debounce", 3),
|
||
active_low=switch_cfg.get("active_low", False),
|
||
device=switch_cfg.get("device"),
|
||
key=switch_cfg.get("key"),
|
||
)
|
||
if not task_switch.start(asyncio.get_running_loop()):
|
||
task_switch = None
|
||
set_global_var("task_switch", task_switch)
|
||
|
||
logger.info(
|
||
"[aibox_sophon_adapter] 已初始化: "
|
||
f"心跳间隔={hb_interval}s, 超时={hb_timeout}s, 检查间隔={check_interval}s"
|
||
)
|
||
|
||
# 后台任务由基类跟踪，cleanup 时自动取消
|
||
self.spawn_task(_check_offline_devices(check_interval))
|
||
|
||
async def on_cleanup(self):
|
||
"""cleanup 时调用：关闭流控制器、外部 MQTT 客户端与任务启停开关（db 与后台任务由基类处理）。"""
|
||
from .global_var import (
|
||
stream_manager, mqtt_forward, mqtt_subscribe, internal_mqtt,
|
||
task_switch,
|
||
)
|
||
|
||
if stream_manager is not None:
|
||
await stream_manager.stop_all()
|
||
|
||
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()
|
||
|
||
if task_switch is not None:
|
||
task_switch.stop()
|
||
|
||
logger.info("[aibox_sophon_adapter] 清理完成")
|
||
|
||
|
||
# 暴露给 app lifespan 的模块级入口（bound method，签名与旧写法完全一致）
|
||
# 注意: 实例名必须为 plugin（router 已被 routes.router APIRouter 占用）
|
||
plugin = AiboxSophonAdapterPlugin()
|
||
create_workers = plugin.create_workers
|
||
cleanup = plugin.cleanup
|