diff --git a/aibox_sophon_adapter/README.md b/aibox_sophon_adapter/README.md index 854fe99..774641b 100644 --- a/aibox_sophon_adapter/README.md +++ b/aibox_sophon_adapter/README.md @@ -12,7 +12,7 @@ | # | 功能 | 说明 | | --- | -------------------- | ------------------------------------------------------------------------------------------ | -| 1 | **MQTT 心跳上报** | 通过网关 MQTT Bridge 订阅 `box/aibox_sophon/board_ping`，解析设备系统参数并持久化到 SQLite | +| 1 | **MQTT 心跳上报** | 通过网关 MQTT Bridge 订阅 `box/aibox_sophon/board_ping`，解析设备系统参数并持久化到数据库 | | 2 | **设备状态查询** | 提供 HTTP API 查询设备在线状态、详情、离线设备列表 | | 3 | **离线检测告警** | 后台任务周期扫描，超过 `heartbeat.timeout` 未上报心跳的设备标记为离线并输出告警日志 | | 4 | **视频流管理** | 心跳中状态为「正常」的通道自动通过 MediaMTX 拉取 RTSP 视频流（`StreamManager`） | @@ -20,9 +20,11 @@ | 6 | **告警上报接口** | 盒子违规事件 JSON 报文字动解析入库（`alarm_record` 表） | | 7 | **告警查询接口** | 告警列表（分页/筛选）与单条告警详情查询 | | 8 | **豆包 AI 分析** | 告警图片可交由 `doubao` 插件判定是否为误报（可配置开关） | -| 9 | **外部 MQTT 转发** | 心跳数据转发到外部 MQTT Broker（`mqtt_forward`） | -| 10 | **外部 MQTT 订阅** | 直接订阅外部 Broker，实现 `edge_app_controller` 请求-响应协议（`mqtt_subscribe`） | -| 11 | **告警转发外部平台** | 先上传视频再上报告警到外部开放平台（`alarm_forward`） | +| 9 | **IP 音柱 TTS 播报** | 告警时按违规类型调用 `ip_sound_column` 插件播放 TTS 语音（可配置开关/音量/文案） | +| 10 | **外部 MQTT 转发** | 心跳数据转发到外部 MQTT Broker（`mqtt_forward`） | +| 11 | **外部 MQTT 订阅** | 直接订阅外部 Broker，实现 `edge_app_controller` 请求-响应协议（`mqtt_subscribe`） | +| 12 | **告警转发外部平台** | 先上传视频再上报告警到外部开放平台（`alarm_forward`） | +| 13 | **IN2 任务启停开关** | 监听盒子 IN2 数字输入：有电启动全部算法任务，断电停止全部任务（`task_switch`） | --- @@ -96,7 +98,10 @@ src/plugins/aibox_sophon_adapter/ ├── global_var.py # 模块级全局变量（db / stream_manager / mqtt_* / 各配置） ├── routes.py # 自定义 FastAPI 路由（video_upload / video / alarm_report） ├── edge_app_controller.py # 外部 MQTT 请求-响应处理器（command 注册表分发） +├── task_switch.py # IN2 数字输入控制任务启停（复用 diwatch 的 IN2 通道） ├── alarm_trim.py # 告警记录清理（超限裁剪 + 删除关联视频） +├── diwatch/ # DI 数字输入监听库（SE5/SE9 的 IN1/IN2，独立 CLI：python3 -m diwatch；task_switch 复用其 IN2 通道） +├── in1.py / in11.py # diwatch 前身：IN1 事件脚本 / IN2 轮询脚本（保留备用） ├── router/ # Zenoh RPC 端点（@rpc_method） │ ├── __init__.py # 导出全部端点模块 │ ├── board_ping.py # MQTT 心跳上报 @@ -188,8 +193,9 @@ curl "http://127.0.0.1:8000/api/v1/aibox_sophon_adapter/list_devices" 1. 手动 Pydantic 验证基础字段（`alarm_models.AlarmReportRequest`，失败记录完整请求体并返回 422） 2. **豆包 AI 分析**：判定为 `False` 时忽略告警（返回 Code 0 + "忽略告警"） -3. **写库**：按 `AlarmId/UniqueId` 去重写入 `alarm_record`，超限自动裁剪（`alarm_trim`） -4. **外部转发**：先上传视频替换 `VideoFile` 为外部 `VideoId`，再上报告警 +3. **IP 音柱 TTS 播报**：按 `Summary` 匹配文案调用 `ip_sound_column.tts` 播放语音（失败仅记录日志，不阻断） +4. **写库**：按 `AlarmId/UniqueId` 去重写入 `alarm_record`，超限自动裁剪（`alarm_trim`） +5. **外部转发**：先上传视频替换 `VideoFile` 为外部 `VideoId`，再上报告警 > 盒子仅当 **HTTP 200 且 `Result.Code` 非 0** 时才认为上报失败。 @@ -257,9 +263,11 @@ curl "http://127.0.0.1:8000/api/v1/aibox_sophon_adapter/list_devices" --- -## 数据库表（SQLite） +## 数据库表 -数据库文件：`{data_dir}/aibox_sophon_adapter.db`（默认 `./data/aibox_sophon_adapter.db`）。 +数据库默认为 SQLite 文件 `{data_dir}/aibox_sophon_adapter.db`（默认 `./data/`）； +配置了 `[workers.config.database].url` 时优先走 PostgreSQL（当前 config.toml 即此模式）， +由 `chongming_worker.Plugin` 基类统一初始化。 ### `board_heartbeat` — 设备心跳记录 @@ -323,6 +331,8 @@ tags = ["aibox_sophon_adapter", "sophon", "adapter"] | `internal_mqtt` | `enabled` / `uri` / `request_topic` / `response_topic` | false / `mqtt://127.0.0.1:1883` / … | 内部请求-响应客户端 | | `alarm_retention` | `max_records` | `0` | 最多保留告警记录数（0 不限制） | | `doubao_analysis` | `enabled` | `true` | 是否启用豆包 AI 分析告警图片 | +| `ip_sound_column` | `enabled` / `volume` / `tts` | `true` / `50` / `{}` | 告警时 IP 音柱 TTS 播报（`tts` 为 Summary → 播报文案映射，未命中用默认文案） | +| `task_switch` | `enabled` / `api_base` / `start_command` / `stop_command` / `poll_interval` / `debounce` / `active_low` | true / `http://127.0.0.1` / 1 / 0 / 0.05 / 3 / false | IN2 数字输入控制任务启停（见下节） | | `alarm_forward` | `enabled` / `url` / `video_upload_url` / `timeout` | false / … / … / 10.0 | 告警转发外部开放平台 | > 示例：`mqtt_forward.uri = "mqtt://admin:Mixc%402021@192.168.1.6:1883"` —— 密码中特殊字符 @@ -340,6 +350,25 @@ tags = ["aibox_sophon_adapter", "sophon", "adapter"] --- +## IN2 任务启停开关（task_switch） + +监听盒子 IN2 数字输入（diwatch `SysfsBackend` 轮询 `gpio_status`，含抗抖）， +电平变化时通过盒子本地 HTTP 接口控制**全部**算法任务： + +- IN2 **有电**（电平 1）→ 启动全部任务（`ControlCommand = start_command`，默认 1） +- IN2 **断电**（电平 0）→ 停止全部任务（`ControlCommand = stop_command`，默认 0） + +控制流程：`POST {api_base}/api/alg_task_fetch`（查全部任务的 `AlgTaskSession`）→ +逐个 `POST {api_base}/api/alg_task_control`（启停控制，按返回 `Result.Code` 判定成败并记日志）。 + +行为约定： + +1. **只响应电平跳变**，不按插件启动时的初始电平动作（避免插件重启时误触发全量启停）。 +2. DI 轮询跑在独立线程，HTTP 控制调度回事件循环并加锁串行，快速抖动不会并发穿插。 +3. 找不到 IN2 状态文件（非盒子环境）时降级为不启动，仅输出警告，不影响插件其他功能。 + +--- + ## 开发指南 ### 新增 RPC 端点 diff --git a/aibox_sophon_adapter/__init__.py b/aibox_sophon_adapter/__init__.py index 3d6c9ba..33697ca 100644 --- a/aibox_sophon_adapter/__init__.py +++ b/aibox_sophon_adapter/__init__.py @@ -10,6 +10,7 @@ AI Box Sophon Adapter 插件 6. 视频上传接口 — 盒子告警时录制视频上传（FORM-DATA） 7. 告警上报接口 — 违规事件 JSON 报文字动解析入库 8. 告警查询接口 — 告警列表/详情查询 + 9. IN2 任务启停开关 — DI 输入有电启动全部任务，断电停止全部任务 心跳超时机制: - 设备按配置间隔（默认 5 秒）上报心跳 @@ -129,6 +130,9 @@ class AiboxSophonAdapterPlugin(Plugin): 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 @@ -174,6 +178,24 @@ class AiboxSophonAdapterPlugin(Plugin): "[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), + ) + 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" @@ -183,9 +205,10 @@ class AiboxSophonAdapterPlugin(Plugin): self.spawn_task(_check_offline_devices(check_interval)) async def on_cleanup(self): - """cleanup 时调用：关闭流控制器与外部 MQTT 客户端（db 与后台任务由基类处理）。""" + """cleanup 时调用：关闭流控制器、外部 MQTT 客户端与任务启停开关（db 与后台任务由基类处理）。""" from .global_var import ( stream_manager, mqtt_forward, mqtt_subscribe, internal_mqtt, + task_switch, ) if stream_manager is not None: @@ -200,6 +223,9 @@ class AiboxSophonAdapterPlugin(Plugin): if internal_mqtt is not None: await internal_mqtt.stop() + if task_switch is not None: + task_switch.stop() + logger.info("[aibox_sophon_adapter] 清理完成") diff --git a/aibox_sophon_adapter/diwatch/backends.py b/aibox_sophon_adapter/diwatch/backends.py index e8b8bff..283332c 100644 --- a/aibox_sophon_adapter/diwatch/backends.py +++ b/aibox_sophon_adapter/diwatch/backends.py @@ -14,7 +14,6 @@ import errno import glob import os -import select # noqa: F401 (留给上层使用) import struct import time @@ -42,9 +41,11 @@ class Backend: raise NotImplementedError def read(self): - """非阻塞读一次，返回 (level, ts) 或 None。 + """非阻塞读一次，返回本批事件列表 [(level, ts), ...]；无事件返回 []。 level 是原始电平 0/1，尚未做 active_low 翻转。 + 同一批数据里的多个事件必须全部返回，不能只留最后一个， + 否则快速抖动（一次 read 里按下+释放）会丢失中间跳变。 """ raise NotImplementedError @@ -125,15 +126,15 @@ class EventBackend(Backend): # ---- 读取 ---- def read(self): - """返回 (level, ts) 或 None；只汇报关心的 key。""" + """返回 [(level, ts), ...]；只汇报关心的 key，批内事件按序全部返回。""" if self.fd is None: - return None + return [] try: data = os.read(self.fd, EVENT_STRUCT.size * 64) except OSError: - return None + return [] self.buf += data - result = None + events = [] while len(self.buf) >= EVENT_STRUCT.size: chunk = self.buf[:EVENT_STRUCT.size] self.buf = self.buf[EVENT_STRUCT.size:] @@ -147,8 +148,8 @@ class EventBackend(Backend): if value not in (VALUE_RELEASE, VALUE_PRESS): continue ts = sec + usec / 1e6 - result = (value, ts) - return result + events.append((value, ts)) + return events # --------------------------------------------------------------------------- @@ -165,6 +166,7 @@ class SysfsBackend(Backend): self.interval = interval self._cand = None self._cand_n = 0 + self._reported = None # 上次上报的稳定值（跳变只报一次） @property def name(self): @@ -193,13 +195,14 @@ class SysfsBackend(Backend): # ---- 带抗抖的读取 ---- def read(self): - """返回 (level, ts) 或 None。 + """返回 [(level, ts)]；抗抖未确认或电平未变化时返回 []。 - level 为抗抖确认后的稳定值；None 表示还没确认跳变。 + 连续 debounce 次采样一致才确认稳定值；确认值与上次上报不同 + （真实跳变）才上报一次，避免稳态期间每个抗抖周期重复产出事件。 """ v = self.read_raw() if v is None: - return None + return [] ts = time.time() if self._cand is None or v == self._cand: @@ -209,10 +212,13 @@ class SysfsBackend(Backend): self._cand = v self._cand_n = 1 - if self._cand_n >= self.debounce: - self._cand_n = 0 # 复位，避免重复上报 - return (int(v), ts) - return None + if self._cand_n < self.debounce: + return [] + confirmed = int(v) + if self._reported == confirmed: + return [] + self._reported = confirmed + return [(confirmed, ts)] # --------------------------------------------------------------------------- diff --git a/aibox_sophon_adapter/diwatch/core.py b/aibox_sophon_adapter/diwatch/core.py index 36779c5..47780d4 100644 --- a/aibox_sophon_adapter/diwatch/core.py +++ b/aibox_sophon_adapter/diwatch/core.py @@ -161,9 +161,8 @@ class DIChannel: break timeout = int(min(remain, 0.5) * 1000) poller.poll(timeout) - got = self.backend.read() - if got: - raw, ts = got + # read() 返回本批全部事件，按序处理，快速抖动不丢中间跳变 + for raw, ts in self.backend.read(): if self.on_event: self.on_event(ts, raw) self._emit(self._to_level(raw), ts) @@ -173,13 +172,10 @@ class DIChannel: while not self._stop: if deadline and time.time() >= deadline: break - got = self.backend.read() - if got: - raw, ts = got - lvl = self._to_level(raw) + for raw, ts in self.backend.read(): if self.on_event: self.on_event(ts, raw) - self._emit(lvl, ts) + self._emit(self._to_level(raw), ts) time.sleep(interval) def request_stop(self): diff --git a/aibox_sophon_adapter/global_var.py b/aibox_sophon_adapter/global_var.py index 8d06706..c69b5c9 100644 --- a/aibox_sophon_adapter/global_var.py +++ b/aibox_sophon_adapter/global_var.py @@ -13,6 +13,7 @@ if TYPE_CHECKING: from chongming_mqtt.mqtt_forward import MQTTForwardClient from chongming_mqtt.mqtt_subscribe import MQTTSubscribeClient from chongming_mqtt.internal_mqtt import InternalMQTTClient + from .task_switch import In2TaskSwitch # 数据库实例 db: Database | None = None @@ -69,6 +70,9 @@ heartbeat_interval: float = 5.0 # ffmpeg 是否可用（on_init 时探测；不可用时跳过 RTSP 拉流，不阻断心跳） ffmpeg_available: bool = True +# IN2 任务启停开关控制器（DI 有电启动全部任务，断电停止全部任务） +task_switch: "In2TaskSwitch | None" = None + def set_global_var(name: str, value: Any) -> None: """ diff --git a/aibox_sophon_adapter/in1.py b/aibox_sophon_adapter/in1.py deleted file mode 100644 index 931056c..0000000 --- a/aibox_sophon_adapter/in1.py +++ /dev/null @@ -1,213 +0,0 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- -"""IN1 (12V 输入) 事件监听。 - -背景： - IN1 在硬件上接到某条 GPIO，但那条 GPIO 已被内核 gpio-keys 驱动占用， - /sys/class/gpio 导不出来。驱动把它映射到 /dev/input/event1， - 上报键码 KEY_UP (103)。 - - 本机光耦输入是反相的：施加 12V 时 GPIO 被拉低，驱动上报“释放”(0)； - 撤除 12V 时 GPIO 被上拉，驱动上报“按下”(1)。 - 因此默认 active_low=True，把语义翻回来，让“加 12V = 有电”。 - -本模块只做一件事：盯着 IN1，有电/断电时回调。 -依赖：仅标准库。 -""" - -import argparse -import datetime -import glob -import os -import select -import struct -import subprocess -import sys -import time - -# ---- input 事件协议 ---- -EV_KEY = 0x01 -EVENT_STRUCT = struct.Struct("llHHi") # aarch64 / x86_64 都是 24 字节 - -KEY_UP = 103 -VALUE_RELEASE = 0 -VALUE_PRESS = 1 -VALUE_REPEAT = 2 - -DEFAULT_DEVICE = "/dev/input/event1" - - -def now(): - return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] - - -def find_event_device(name="gpio-keys"): - """按设备名找 eventN，找不到返回 None。设备号变化时很有用。""" - for path in sorted(glob.glob("/dev/input/event*")): - ev = path.rsplit("/", 1)[-1] - namefile = "/sys/class/input/%s/device/name" % ev - try: - with open(namefile) as f: - if f.read().strip() == name: - return path - except OSError: - continue - return None - - -class IN1Listener: - """监听 IN1，有电/断电触发回调。 - - active_low=True（默认）：驱动上报“释放”(0) 视为有电，“按下”(1) 视为断电。 - 适配本机光耦反相电路。 - active_low=False： 按驱动原始语义，value=1 视为有电。 - - 用法： - def on_press(ts): print("IN1 有电", ts) - def on_release(ts): print("IN1 断电", ts) - - with IN1Listener(on_press=on_press, on_release=on_release) as lis: - lis.run() # 阻塞，Ctrl-C 退出 - # lis.run(duration=10) # 跑 10 秒后返回 - """ - - def __init__(self, device=DEFAULT_DEVICE, key=KEY_UP, - on_press=None, on_release=None, - on_event=None, log=None, - active_low=True): - self.device = device - self.key = key - self.on_press = on_press # 12V 接通 - self.on_release = on_release # 12V 撤除 - self.on_event = on_event # 原始事件 (ts, code, value)，可选 - self.log = log or (lambda msg: print(msg, flush=True)) - self.active_low = active_low - - self.fd = None - self.buf = b"" - - # ---- 生命周期 ---- - def open(self): - if self.fd is None: - self.fd = os.open(self.device, os.O_RDONLY | os.O_NONBLOCK) - - def close(self): - if self.fd is not None: - try: - os.close(self.fd) - finally: - self.fd = None - - def __enter__(self): - self.open() - return self - - def __exit__(self, *exc): - self.close() - - # ---- 事件处理 ---- - def _drain(self): - if self.fd is None: - return - try: - data = os.read(self.fd, EVENT_STRUCT.size * 64) - except OSError: - return - self.buf += data - while len(self.buf) >= EVENT_STRUCT.size: - chunk, self.buf = self.buf[:EVENT_STRUCT.size], self.buf[EVENT_STRUCT.size:] - sec, usec, etype, code, value = EVENT_STRUCT.unpack(chunk) - if etype != EV_KEY: - continue - ts = sec + usec / 1e6 - if self.on_event: - self.on_event(ts, code, value) - if code != self.key: - continue - - if value == VALUE_REPEAT: - self.log("%s IN1 重复" % now()) - continue - - # 硬件极性：active_low 时，value=0 表示 12V 接通 - if self.active_low: - powered = (value == VALUE_RELEASE) - else: - powered = (value == VALUE_PRESS) - - if powered: - self.log("%s IN1 有电" % now()) - if self.on_press: - self.on_press(ts) - else: - self.log("%s IN1 断电" % now()) - if self.on_release: - self.on_release(ts) - - # ---- 主循环 ---- - def run(self, duration=None): - """阻塞监听。duration 秒后返回；None = 一直跑到 Ctrl-C。""" - if self.fd is None: - self.open() - if self.fd is None: - return - poller = select.poll() - poller.register(self.fd, select.POLLIN) - - deadline = time.time() + duration if duration else None - while True: - timeout = 500 # ms - if deadline: - remain = deadline - time.time() - if remain <= 0: - return - timeout = int(min(remain, 0.5) * 1000) - poller.poll(timeout) - self._drain() - - -# ---- CLI 演示 ---- -def main(): - ap = argparse.ArgumentParser(description="监听 IN1 (12V 输入)") - ap.add_argument("-d", "--device", default=DEFAULT_DEVICE, - help="事件设备，默认 %(default)s；传 auto 自动按名字查找") - ap.add_argument("-k", "--key", type=lambda s: int(s, 0), default=KEY_UP, - help="键码，默认 103 (KEY_UP)") - ap.add_argument("-t", "--duration", type=float, default=0, - help="监听秒数，0 = 一直跑") - ap.add_argument("--press-command", help="IN1 有电时执行的 shell 命令") - ap.add_argument("--release-command", help="IN1 断电时执行的 shell 命令") - ap.add_argument("--active-low", dest="active_low", action="store_true", - default=True, - help="12V 接通上报为“有电”（默认，适配本机光耦反相）") - ap.add_argument("--no-active-low", dest="active_low", action="store_false", - help="不翻转，按驱动原始语义") - args = ap.parse_args() - - device = args.device - if device == "auto": - device = find_event_device("gpio-keys") or DEFAULT_DEVICE - - def run_cmd(cmd): - if cmd: - subprocess.Popen(cmd, shell=True) - - with IN1Listener( - device=device, - key=args.key, - active_low=args.active_low, - on_press=lambda ts: run_cmd(args.press_command), - on_release=lambda ts: run_cmd(args.release_command), - ) as lis: - print("监听 %s key=%d active_low=%s，Ctrl-C 退出" - % (device, args.key, args.active_low), flush=True) - try: - lis.run(duration=args.duration or None) - except KeyboardInterrupt: - print("\n退出。") - - return 0 - - -if __name__ == "__main__": - sys.exit(main()) \ No newline at end of file diff --git a/aibox_sophon_adapter/in11.py b/aibox_sophon_adapter/in11.py deleted file mode 100644 index f41da49..0000000 --- a/aibox_sophon_adapter/in11.py +++ /dev/null @@ -1,456 +0,0 @@ -#!/usr/bin/env python3 -""" -watch_in2.py — 监听数字输入 IN2 的电平变化（Python 版） - -原 Bash 脚本 watch_in2.sh 的功能等价实现： - - IN2 编号：Linux 全局 GPIO 483 - = gpiochip480（label "27010000.gpio"，bank-name "porta"）的第 3 线 - - 推导依据可用 --check 现场复算 - - 由于 483 被平台驱动占用，sysfs export 会 EBUSY， - 唯一可读通道是 gpio_status，不支持 poll/select，只能定时采样。 - - 该文件权限 0666，所以本脚本不需要 root。 - -用法: - ./watch_in2.py [轮询间隔秒] 默认 0.02 - ./watch_in2.py 0.005 5ms 细粒度 - ./watch_in2.py --check 只做自检：复算编号 + 打印当前电平，然后退出 - -停止: Ctrl-C 打印汇总 -""" - -import os -import sys -import time -import signal -import datetime -import argparse -from pathlib import Path - -# --------------------------------------------------------------------------- -# 常量 -# --------------------------------------------------------------------------- -IN_INDEX = 2 -GPIO_NUM = 483 -GPIOCHIP = 480 -DEBOUNCE = 2 # 连续多少次采样一致才确认跳变 - -DEFAULT_INTERVAL = 0.02 -LOG_DIR_ENV = "DI_LOG_DIR" -STATUS_FILE_ENV = "DI_STATUS_FILE" # 仅用于测试 - -# 候选状态文件路径 -STATUS_CANDIDATES = [ - f"/sys/bus/platform/devices/gpio-in-{IN_INDEX}/gpio_status", - f"/dev/tv_gpioin_{IN_INDEX}", -] -IN1_CANDIDATES = [ - "/sys/bus/platform/devices/gpio-in-1/gpio_status", - "/dev/tv_gpioin_1", -] - -DEVICETREE_BASE = Path("/sys/firmware/devicetree/base") - - -# --------------------------------------------------------------------------- -# 工具函数 -# --------------------------------------------------------------------------- -def usage() -> str: - """返回脚本头部注释作为用法说明。""" - return __doc__ or "" - - -def read_level_of(path: str): - """读取状态文件，返回 '0' / '1'，失败返回 None。""" - try: - with open(path, "r") as f: - v = f.read() - except (OSError, IOError): - return None - v = v.strip().strip("\t\r\n ") - if v in ("0", "1"): - return v - return None - - -def ts_short() -> str: - """HH:MM:SS.mmm""" - now = datetime.datetime.now() - return now.strftime("%H:%M:%S.") + f"{now.microsecond // 1000:03d}" - - -def fmt_dur(ms: int) -> str: - """毫秒格式化。""" - if ms < 1000: - return f"{ms}ms" - return f"{ms / 1000:.3f}s" - - -def probe_writable(path: Path) -> bool: - """检查目录是否可写，必要时创建。""" - try: - path.mkdir(parents=True, exist_ok=True) - except OSError: - return False - test_file = path / f".wtest.{os.getpid()}" - try: - with open(test_file, "w"): - pass - test_file.unlink() - return True - except OSError: - try: - test_file.unlink() - except OSError: - pass - return False - - -# --------------------------------------------------------------------------- -# 设备树复算（等价于 Bash 的 derive_gpio） -# --------------------------------------------------------------------------- -def derive_gpio(n: int): - """从设备树复算 IN 的全局 GPIO 号；失败返回 None。""" - f = DEVICETREE_BASE / f"gpio-in-{n}" / "control-gpios" - if not f.exists(): - return None - - try: - data = f.read_bytes() - except OSError: - return None - - if len(data) < 8: - return None - - # 前 4 字节 phandle，后 4 字节 pin（大端） - ph = int.from_bytes(data[0:4], "big") - pin = int.from_bytes(data[4:8], "big") - - # phandle -> gpio-controller 节点 - ctrl = None - for c in DEVICETREE_BASE.glob("gpio@*/gpio-controller@*"): - ph_file = c / "phandle" - if not ph_file.exists(): - continue - try: - cph = int.from_bytes(ph_file.read_bytes()[:4], "big") - except OSError: - continue - if cph == ph: - ctrl = c - break - if ctrl is None: - return None - - # gpio@<地址>/gpio-controller@N -> gpiochip label "<地址>.gpio" -> base - bank = ctrl.parent.name # gpio@27010000 - bank = bank[len("gpio@"):] if bank.startswith("gpio@") else bank - label = f"{bank}.gpio" - - base = None - gpio_class = Path("/sys/class/gpio") - if gpio_class.exists(): - for g in gpio_class.glob("gpiochip*"): - label_file = g / "label" - base_file = g / "base" - if not label_file.exists() or not base_file.exists(): - continue - try: - if label_file.read_text().strip() == label: - base = int(base_file.read_text().strip()) - break - except (OSError, ValueError): - continue - if base is None: - return None - - return base + pin - - -# --------------------------------------------------------------------------- -# 主监听器 -# --------------------------------------------------------------------------- -class Watcher: - def __init__(self, interval: float, dev: str, dev_in1: str, log_dir: Path): - self.interval = interval - self.dev = dev - self.dev_in1 = dev_in1 - self.log_dir = log_dir - self.log_path = None - self.log_fh = None - - self.state = None - self.count = 0 - self.ms_low = 0 - self.ms_high = 0 - self.state_since = 0 - - self._stop = False - - # ---------- 日志 ---------- - def open_log(self): - ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") - self.log_path = self.log_dir / f"in2_{ts}.log" - try: - self.log_fh = open(self.log_path, "a", encoding="utf-8") - except OSError: - self.log_fh = None - - def say(self, msg: str): - print(msg) - if self.log_fh: - try: - self.log_fh.write(msg + "\n") - self.log_fh.flush() - except OSError: - pass - - # ---------- 电平读取 ---------- - def read_level(self): - return read_level_of(self.dev) - - def peer(self) -> str: - if self.dev_in1: - v = read_level_of(self.dev_in1) - if v is not None: - return f"IN1={v}" - return "IN1=-" - - # ---------- 信号处理 ---------- - def request_stop(self, *_args): - self._stop = True - - # ---------- 汇总 ---------- - def print_summary(self): - now_ms = int(time.time() * 1000) - dur = now_ms - self.state_since - if self.state == "0": - self.ms_low += dur - elif self.state == "1": - self.ms_high += dur - - self.say("") - self.say("============================================================") - self.say(f"汇总 {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") - self.say(f" 跳变次数：{self.count}") - self.say(f" 低电平累计：{fmt_dur(self.ms_low)}") - self.say(f" 高电平累计：{fmt_dur(self.ms_high)}") - self.say(f" 结束电平：{self.state}") - if self.count == 0: - self.say("") - self.say(f" 整段时间内 IN{IN_INDEX} 没有任何变化。") - self.say(f" 若此时确实短接过 IN{IN_INDEX} 与 COM，说明激励没真正加到端子上；") - self.say(f" 请确认接的是 IN{IN_INDEX}（不是 IN1）且 COM 已共地，必要时上 12V 再试。") - self.say(f" 日志已保存：{self.log_path}") - self.say("============================================================") - - # ---------- 主循环 ---------- - def run(self): - # 首次采样 - state = self.read_level() - if state is None: - print(f"错误：读不出 {self.dev} 的有效电平（期望 0/1）。", file=sys.stderr) - try: - content = Path(self.dev).read_text()[:64] - except OSError: - content = "<无法读取>" - print(f" 内容：{content}", file=sys.stderr) - return 1 - - self.state = state - self.state_since = int(time.time() * 1000) - - self.open_log() - - self.say("============================================================") - self.say(f"IN{IN_INDEX} 监听开始 {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") - self.say(f" IN{IN_INDEX} = GPIO {GPIO_NUM} = gpiochip{GPIOCHIP} 第 {GPIO_NUM - GPIOCHIP} 线") - self.say(f" 状态文件：{self.dev}") - self.say(f" 轮询间隔：{self.interval}s 抗抖：连续 {DEBOUNCE} 次采样一致") - self.say(f" 日志：{self.log_path}") - self.say("============================================================") - self.say(f"初始电平：{self.state}") - self.say("") - self.say("把 IN1 与 IN2 短接约 2 秒再松开（实测这根接法才有效；单接 IN2-GND 无效）。") - self.say("下面每行会同时给出 IN1 和 IN2 的电平。Ctrl-C 结束。") - self.say("") - - cand = None - cand_n = 0 - - # 注册信号 - signal.signal(signal.SIGINT, self.request_stop) - signal.signal(signal.SIGTERM, self.request_stop) - - try: - while not self._stop: - v = self.read_level() - if v is None: - self.say(f"{ts_short()} 读取失败，跳过本次采样") - time.sleep(self.interval) - continue - - now_ms = int(time.time() * 1000) - - if v == self.state: - cand = None - cand_n = 0 - else: - if v == cand: - cand_n += 1 - else: - cand = v - cand_n = 1 - - if cand_n >= DEBOUNCE: - dur = now_ms - self.state_since - if self.state == "0": - self.ms_low += dur - else: - self.ms_high += dur - - self.count += 1 - self.say( - f"{ts_short()} 变化 IN{IN_INDEX}: {self.state} -> {v}" - f" [{self.peer()}] 上一状态持续 {fmt_dur(dur)}" - ) - - self.state = v - self.state_since = now_ms - cand = None - cand_n = 0 - - time.sleep(self.interval) - finally: - self.print_summary() - if self.log_fh: - self.log_fh.close() - - return 0 - - -# --------------------------------------------------------------------------- -# --check 模式 -# --------------------------------------------------------------------------- -def do_check(dev: str, interval: float): - print(f"=== IN{IN_INDEX} 编号自检 ===") - print() - print(f"状态文件：{dev}") - print(f"轮询间隔：{interval}s（本模式不轮询）") - print() - print("--- 从设备树现场复算 ---") - - ok = True - for n in (1, 2): - got = derive_gpio(n) - if got is None: - print(f" IN{n}：复算失败（设备树 gpio-in-{n} 或对应 gpiochip 不可读）") - ok = False - continue - expect = 481 + n # IN1 -> 482, IN2 -> 483 - if got == expect: - print(f" IN{n} -> GPIO {got} ✓ 与预期一致") - else: - print(f" IN{n} -> GPIO {got} ✗ 预期 {expect}，需人工核对") - ok = False - - print() - if ok: - print(f"结论：IN{IN_INDEX} = GPIO {GPIO_NUM}" - f"（gpiochip{GPIOCHIP} 第 {GPIO_NUM - GPIOCHIP} 线），推导成立。") - else: - print("结论：推导与实测不符，请把上面的输出贴出来再排查。") - - print() - print("--- 当前电平 ---") - v = read_level_of(dev) - if v is not None: - print(f" gpio-in-{IN_INDEX}/gpio_status = {v} （对应的全局 GPIO 为 {GPIO_NUM}）") - print(" 注：待机时读到 0 是正常的，激励到来时才会翻转。") - else: - print(f" 读取失败：{dev} 内容不是 0/1。") - - return 0 - - -# --------------------------------------------------------------------------- -# 入口 -# --------------------------------------------------------------------------- -def find_status_file() -> str: - """定位 IN2 状态文件。""" - # 测试用环境变量 - test_file = os.environ.get(STATUS_FILE_ENV) - if test_file and os.access(test_file, os.R_OK): - return test_file - - for c in STATUS_CANDIDATES: - if os.access(c, os.R_OK): - return c - return "" - - -def find_in1_file() -> str: - """定位 IN1 状态文件（可选）。""" - for c in IN1_CANDIDATES: - if os.access(c, os.R_OK): - return c - return "" - - -def main(argv=None): - parser = argparse.ArgumentParser( - description=usage(), - formatter_class=argparse.RawDescriptionHelpFormatter, - add_help=True, - ) - parser.add_argument( - "interval", nargs="?", default=str(DEFAULT_INTERVAL), - help=f"轮询间隔秒（默认 {DEFAULT_INTERVAL}）" - ) - parser.add_argument( - "--check", action="store_true", - help="只做自检：复算编号 + 打印当前电平，然后退出" - ) - args = parser.parse_args(argv) - - # 解析 interval - try: - interval = float(args.interval) - except ValueError: - print(f"错误：轮询间隔 '{args.interval}' 不是正数。", file=sys.stderr) - return 1 - if interval <= 0: - print(f"错误：轮询间隔 '{args.interval}' 不是正数。", file=sys.stderr) - return 1 - - # 定位状态文件 - dev = find_status_file() - if not dev: - print(f"错误：找不到 gpio-in-{IN_INDEX} 的状态文件。", file=sys.stderr) - print(f" 已尝试：{STATUS_CANDIDATES[0]}", file=sys.stderr) - print(f" {STATUS_CANDIDATES[1]}", file=sys.stderr) - print(" 排查：ls /sys/bus/platform/devices/ | grep gpio-in", file=sys.stderr) - return 1 - - # --check 模式 - if args.check: - return do_check(dev, interval) - - # 准备日志目录 - log_dir = Path(os.environ.get(LOG_DIR_ENV, str(Path.home() / "di_logs"))) - if not probe_writable(log_dir): - print(f"提示：{log_dir} 不可写（属主可能是 root），本次日志改写到 {Path.home()}", - file=sys.stderr) - print(f" 想恢复到 {log_dir}：sudo chown {os.environ.get('USER', '')} {log_dir}", - file=sys.stderr) - log_dir = Path.home() - - # IN1 一并盯着 - dev_in1 = find_in1_file() - - watcher = Watcher(interval, dev, dev_in1, log_dir) - return watcher.run() - - -if __name__ == "__main__": - sys.exit(main()) \ No newline at end of file diff --git a/aibox_sophon_adapter/router/board_ping.py b/aibox_sophon_adapter/router/board_ping.py index 34f6608..57bfbc0 100644 --- a/aibox_sophon_adapter/router/board_ping.py +++ b/aibox_sophon_adapter/router/board_ping.py @@ -1,7 +1,8 @@ """ AI Box Sophon Adapter — MQTT 心跳上报处理器 -监听 MQTT 主题 /board_ping（通过 MQTT Bridge 路由到 chongming/aibox_sophon_adapter/board_ping）， +监听 MQTT 主题 box/aibox_sophon/board_ping（通过 MQTT Bridge 路由到 +aibox_sophon_adapter.board_ping，勿用 chongming/ 保留命名空间）， 解析设备心跳数据并更新数据库记录。 """ import orjson @@ -116,9 +117,9 @@ async def board_ping(request: BoardPingRequest, worker_id: str = "") -> dict: if not success: logger.warning("[MQTTForward] 心跳转发失败") else: - logger.warning(f"[MQTTForward] 心跳已转发 ({len(ping_data)} keys)") + logger.debug(f"[MQTTForward] 心跳已转发 ({len(ping_data)} keys)") else: - logger.warning("[MQTTForward] mqtt_forward 客户端为 None，跳过转发") + logger.debug("[MQTTForward] mqtt_forward 客户端未启用，跳过转发") logger.debug( f"[Heartbeat] {board_id}: temp={request.BoardTemp}, " diff --git a/aibox_sophon_adapter/task_switch.py b/aibox_sophon_adapter/task_switch.py new file mode 100644 index 0000000..e5b840c --- /dev/null +++ b/aibox_sophon_adapter/task_switch.py @@ -0,0 +1,175 @@ +""" +AI Box Sophon Adapter — IN2 数字输入控制任务启停 + +监听盒子 IN2 DI 输入（复用 diwatch 的 DIChannel + SysfsBackend，sysfs 轮询）， +电平变化时通过盒子本地 HTTP 接口控制所有算法任务： + + - IN2 有电（电平 1）→ 启动全部任务（ControlCommand = start_command，默认 1） + - IN2 断电（电平 0）→ 停止全部任务（ControlCommand = stop_command，默认 0） + +盒子接口（POST，JSON）： + - /api/alg_task_fetch 查询全部任务：{} → {"Content": [{"AlgTaskSession": ...}], "Result": {"Code": 0}} + - /api/alg_task_control 控制单个任务：{"AlgTaskSession": ..., "ControlCommand": 0/1} + → {"Result": {"Code": 0, "Desc": "已停止"}} + +说明： + - 只响应电平跳变，不按启动时的初始电平动作（避免插件重启时误触发全量启停）。 + - DIChannel 阻塞轮询跑在独立线程，回调经 run_coroutine_threadsafe 调度回事件循环， + HTTP 控制在事件循环内串行执行（apply_lock 防止快速抖动导致并发穿插）。 +""" +import asyncio +import logging +import threading + +import httpx + +from .diwatch.backends import SysfsBackend +from .diwatch.config import IN2_INDEX, find_status_file +from .diwatch.core import DIChannel + +logger = logging.getLogger(__name__) + + +class In2TaskSwitch: + """IN2 数字输入 → 全部算法任务启停开关。""" + + def __init__( + self, + api_base: str = "http://127.0.0.1", + start_command: int = 1, + stop_command: int = 0, + poll_interval: float = 0.05, + debounce: int = 3, + active_low: bool = False, + timeout: float = 10.0, + ): + self.api_base = api_base.rstrip("/") + self.start_command = start_command + self.stop_command = stop_command + self.poll_interval = poll_interval + self.debounce = debounce + self.active_low = active_low + self.timeout = timeout + + self._channel: DIChannel | None = None + self._thread: threading.Thread | None = None + self._loop: asyncio.AbstractEventLoop | None = None + self._apply_lock = asyncio.Lock() + + # ---- 生命周期 ---- + def start(self, loop: asyncio.AbstractEventLoop) -> bool: + """启动监听线程；找不到 IN2 状态文件时返回 False（非盒子环境属正常）。""" + status = find_status_file(IN2_INDEX) + if not status: + logger.warning( + "[TaskSwitch] 找不到 IN2 状态文件，任务启停开关未启动" + "（非盒子环境属正常现象）" + ) + return False + + backend = SysfsBackend(status, debounce=self.debounce, interval=self.poll_interval) + self._channel = DIChannel( + "IN2", backend, + active_low=self.active_low, + on_change=self._on_level_change, + log=lambda m: logger.debug(f"[TaskSwitch] {m}"), + ) + self._loop = loop + self._thread = threading.Thread( + target=self._run_channel, name="in2-task-switch", daemon=True, + ) + self._thread.start() + logger.info( + f"[TaskSwitch] IN2 任务启停开关已启动: 状态文件={status}, " + f"api_base={self.api_base}, " + f"启动命令={self.start_command}, 停止命令={self.stop_command}" + ) + return True + + def stop(self) -> None: + if self._channel is not None: + self._channel.request_stop() + if self._thread is not None: + self._thread.join(timeout=2.0) + self._thread = None + logger.info("[TaskSwitch] IN2 任务启停开关已停止") + + def _run_channel(self) -> None: + try: + self._channel.run() + except Exception as e: + logger.error(f"[TaskSwitch] IN2 监听线程异常退出: {type(e).__name__}: {e}") + + # ---- DI 回调（轮询线程内执行）---- + def _on_level_change(self, old: int, new: int, ts: float) -> None: + logger.info(f"[TaskSwitch] IN2 电平变化: {old} -> {new}") + if self._loop is not None: + asyncio.run_coroutine_threadsafe(self.apply(new), self._loop) + + # ---- 任务控制（事件循环内执行）---- + async def apply(self, level: int) -> None: + """按目标电平控制全部任务：1=启动，0=停止。""" + async with self._apply_lock: + command = self.start_command if level == 1 else self.stop_command + action = "启动" if level == 1 else "停止" + + sessions = await self._fetch_task_sessions() + if not sessions: + logger.warning("[TaskSwitch] 未获取到任何任务，跳过控制") + return + + logger.info( + f"[TaskSwitch] IN2 {'有电' if level else '断电'}，" + f"{action}全部 {len(sessions)} 个任务: {sessions}" + ) + ok = 0 + failed = 0 + async with httpx.AsyncClient(timeout=self.timeout) as client: + for session in sessions: + try: + resp = await client.post( + f"{self.api_base}/api/alg_task_control", + json={"AlgTaskSession": session, "ControlCommand": command}, + ) + result = resp.json().get("Result", {}) + if resp.is_success and result.get("Code") == 0: + ok += 1 + logger.info( + f"[TaskSwitch] 任务 {session} {action}成功: " + f"{result.get('Desc', '')}" + ) + else: + failed += 1 + logger.warning( + f"[TaskSwitch] 任务 {session} {action}失败: " + f"Status={resp.status_code}, Body={resp.text[:200]}" + ) + except Exception as e: + failed += 1 + logger.error( + f"[TaskSwitch] 任务 {session} {action}异常: " + f"{type(e).__name__}: {e}" + ) + logger.info(f"[TaskSwitch] {action}完成: 成功 {ok}, 失败 {failed}") + + async def _fetch_task_sessions(self) -> list[str]: + """查询全部任务的 AlgTaskSession 列表；失败返回空列表。""" + try: + async with httpx.AsyncClient(timeout=self.timeout) as client: + resp = await client.post(f"{self.api_base}/api/alg_task_fetch", json={}) + if not resp.is_success: + logger.warning( + f"[TaskSwitch] 查询任务失败: Status={resp.status_code}, " + f"Body={resp.text[:200]}" + ) + return [] + data = resp.json() + result = data.get("Result", {}) + if result.get("Code") != 0: + logger.warning(f"[TaskSwitch] 查询任务返回错误: {result}") + return [] + content = data.get("Content") or [] + return [t.get("AlgTaskSession") for t in content if t.get("AlgTaskSession")] + except Exception as e: + logger.error(f"[TaskSwitch] 查询任务异常: {type(e).__name__}: {e}") + return []