diff --git a/aibox_sophon_adapter/diwatch/__init__.py b/aibox_sophon_adapter/diwatch/__init__.py new file mode 100644 index 0000000..30003ce --- /dev/null +++ b/aibox_sophon_adapter/diwatch/__init__.py @@ -0,0 +1,36 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""diwatch — 统一 DI 输入监听库，适配 SE5 / SE9。 + +对外主要 API： + + from diwatch import DIChannel, EventBackend, SysfsBackend, make_in1_backend + + # IN1（自动选后端） + ch = DIChannel("IN1", backend=make_in1_backend(), active_low=True) + ch.on_press = lambda ts: print("有电", ts) + ch.on_release = lambda ts: print("断电", ts) + ch.run() + + # IN2（sysfs 轮询） + ch2 = DIChannel("IN2", + backend=SysfsBackend("/sys/bus/platform/devices/gpio-in-2/gpio_status"), + active_low=False) + ch2.on_change = lambda old, new, ts: print(f"{old}->{new}", ts) + ch2.run(duration=10) +""" + +from .core import DIChannel, fmt_dur, now_str +from .backends import ( + EventBackend, SysfsBackend, make_backend, + make_in1_backend, find_status_file, +) +from .devicetree import derive_gpio + +__all__ = [ + "DIChannel", "fmt_dur", "now_str", + "EventBackend", "SysfsBackend", "make_backend", + "make_in1_backend", "find_status_file", "derive_gpio", +] + +__version__ = "1.1.0" diff --git a/aibox_sophon_adapter/diwatch/__main__.py b/aibox_sophon_adapter/diwatch/__main__.py new file mode 100644 index 0000000..ac6a947 --- /dev/null +++ b/aibox_sophon_adapter/diwatch/__main__.py @@ -0,0 +1,6 @@ +#!/usr/bin/env python3 +import sys +from .cli import main + +if __name__ == "__main__": + sys.exit(main()) diff --git a/aibox_sophon_adapter/diwatch/backends.py b/aibox_sophon_adapter/diwatch/backends.py new file mode 100644 index 0000000..e8b8bff --- /dev/null +++ b/aibox_sophon_adapter/diwatch/backends.py @@ -0,0 +1,268 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""DI 通道的底层读取后端。 + +每个后端只负责：open / close / read。 +电平语义（0=低，1=高）由上层 DIChannel 统一处理。 + +后端类型： + - EventBackend：/dev/input/eventN，事件驱动，可 poll。SE5 的 IN1。 + - SysfsBackend：/sys/.../gpio_status 或 /dev/tv_gpioin_N，轮询 + 抗抖。 + SE9 的 IN1、所有设备的 IN2。 +""" + +import errno +import glob +import os +import select # noqa: F401 (留给上层使用) +import struct +import time + +from .config import ( + EV_KEY, VALUE_RELEASE, VALUE_PRESS, VALUE_REPEAT, + DEFAULT_EVENT_DEVICE, GPIO_KEYS_NAME, + DEBOUNCE, DEFAULT_INTERVAL, + find_status_file, +) + +EVENT_STRUCT = struct.Struct("llHHi") # aarch64 / x86_64 都是 24 字节 + + +# --------------------------------------------------------------------------- +# 基类 +# --------------------------------------------------------------------------- +class Backend: + """后端接口。""" + can_poll = False + + def open(self): + raise NotImplementedError + + def close(self): + raise NotImplementedError + + def read(self): + """非阻塞读一次，返回 (level, ts) 或 None。 + + level 是原始电平 0/1，尚未做 active_low 翻转。 + """ + raise NotImplementedError + + @property + def name(self): + raise NotImplementedError + + +# --------------------------------------------------------------------------- +# /dev/input/eventN 后端（SE5 的 IN1） +# --------------------------------------------------------------------------- +class EventBackend(Backend): + """基于 input 事件设备，事件驱动。 + + 原始 value 语义： + VALUE_PRESS(1) -> 硬件“按下” + VALUE_RELEASE(0) -> 硬件“释放” + 本后端返回原始 value，极性翻转留给 DIChannel。 + """ + + can_poll = True + + def __init__(self, device=DEFAULT_EVENT_DEVICE, key=None, + find_by_name=GPIO_KEYS_NAME): + self.device = device + self.key = key + self.find_by_name = find_by_name + self.fd = None + self.buf = b"" + + # ---- 自动查找 ---- + @staticmethod + def find_event_device(name=GPIO_KEYS_NAME): + """按设备名找 eventN，找不到返回 None。""" + for path in sorted(glob.glob("/dev/input/event*")): + ev = path.rsplit("/", 1)[-1] + namefile = f"/sys/class/input/{ev}/device/name" + try: + with open(namefile) as f: + if f.read().strip() == name: + return path + except OSError: + continue + return None + + def resolve_device(self): + if not self.device or self.device == "auto": + return self.find_event_device(self.find_by_name) or DEFAULT_EVENT_DEVICE + return self.device + + @property + def name(self): + dev = self.device or "(auto)" + return f"event:{dev}" + + # ---- 生命周期 ---- + def open(self): + if self.fd is None: + dev = self.resolve_device() + if not dev or not os.path.exists(dev): + raise FileNotFoundError( + errno.ENOENT, + f"event 设备不存在：{dev}", + dev, + ) + self.device = dev + self.fd = os.open(dev, 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 fileno(self): + return self.fd + + # ---- 读取 ---- + def read(self): + """返回 (level, ts) 或 None；只汇报关心的 key。""" + if self.fd is None: + return None + try: + data = os.read(self.fd, EVENT_STRUCT.size * 64) + except OSError: + return None + self.buf += data + result = None + while len(self.buf) >= EVENT_STRUCT.size: + chunk = self.buf[:EVENT_STRUCT.size] + self.buf = self.buf[EVENT_STRUCT.size:] + sec, usec, etype, code, value = EVENT_STRUCT.unpack(chunk) + if etype != EV_KEY: + continue + if self.key is not None and code != self.key: + continue + if value == VALUE_REPEAT: + continue + if value not in (VALUE_RELEASE, VALUE_PRESS): + continue + ts = sec + usec / 1e6 + result = (value, ts) + return result + + +# --------------------------------------------------------------------------- +# sysfs 状态文件后端（SE9 的 IN1、所有 IN2） +# --------------------------------------------------------------------------- +class SysfsBackend(Backend): + """基于 /sys/.../gpio_status 或 /dev/tv_gpioin_N，轮询采样 + 抗抖。""" + + can_poll = False + + def __init__(self, path, debounce=DEBOUNCE, interval=DEFAULT_INTERVAL): + self.path = path + self.debounce = max(1, int(debounce)) + self.interval = interval + self._cand = None + self._cand_n = 0 + + @property + def name(self): + return f"sysfs:{self.path}" + + def open(self): + if not os.access(self.path, os.R_OK): + raise FileNotFoundError( + errno.ENOENT, + f"状态文件不可读：{self.path}", + self.path, + ) + + def close(self): + pass + + # ---- 原始电平 ---- + def read_raw(self): + try: + with open(self.path, "r") as f: + v = f.read() + except OSError: + return None + v = v.strip().strip("\t\r\n ") + return v if v in ("0", "1") else None + + # ---- 带抗抖的读取 ---- + def read(self): + """返回 (level, ts) 或 None。 + + level 为抗抖确认后的稳定值；None 表示还没确认跳变。 + """ + v = self.read_raw() + if v is None: + return None + ts = time.time() + + if self._cand is None or v == self._cand: + self._cand = v + self._cand_n += 1 + else: + self._cand = v + self._cand_n = 1 + + if self._cand_n >= self.debounce: + self._cand_n = 0 # 复位，避免重复上报 + return (int(v), ts) + return None + + +# --------------------------------------------------------------------------- +# 工厂 +# --------------------------------------------------------------------------- +def make_backend(kind: str, **kwargs): + """kind = 'event' | 'sysfs'。""" + if kind == "event": + return EventBackend(**kwargs) + if kind == "sysfs": + return SysfsBackend(**kwargs) + raise ValueError(f"未知后端类型：{kind}") + + +def make_in1_backend(device=None, key=None, in_index=1): + """自动为 IN1 选择后端，适配 SE5 / SE9。 + + 优先级： + 1) 显式 device（非 auto） -> EventBackend + 2) 自动查找 gpio-keys 的 event 设备 -> EventBackend + 3) 查找 gpio-in- 的状态文件 -> SysfsBackend + 4) 都找不到 -> 返回 EventBackend（open 时抛出可诊断的错误） + """ + # 1) 显式指定 + if device and device != "auto": + return EventBackend(device, key=key) + + # 2) 自动查找 gpio-keys + ev = EventBackend.find_event_device(GPIO_KEYS_NAME) + if ev: + return EventBackend(ev, key=key) + + # 3) 回退到 sysfs 状态文件 + status = find_status_file(in_index) + if status: + return SysfsBackend(status) + + # 4) 都找不到，返回一个 open 时会报错的 EventBackend + return EventBackend(DEFAULT_EVENT_DEVICE, key=key) + +def find_event_device(name=GPIO_KEYS_NAME): + """模块级别别名，等价于 EventBackend.find_event_device.""" + return EventBackend.find_event_device(name) + +__all__ = [ + "Backend", + "EventBackend", + "SysfsBackend", + "make_backend", + "make_in1_backend", + "find_status_file", + "find_event_device", +] diff --git a/aibox_sophon_adapter/diwatch/cli.py b/aibox_sophon_adapter/diwatch/cli.py new file mode 100644 index 0000000..6b5f284 --- /dev/null +++ b/aibox_sophon_adapter/diwatch/cli.py @@ -0,0 +1,342 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""diwatch 统一命令行入口，适配 SE5 / SE9。 + +用法示例： + # IN1（自动选后端：SE5 走 event，SE9 走 sysfs） + python3 -m diwatch in1 + python3 -m diwatch in1 -t 10 + python3 -m diwatch in1 --press-command 'logger "IN1 on"' \ + --release-command 'logger "IN1 off"' + python3 -m diwatch in1 -d /dev/input/event0 # 手动指定 event + python3 -m diwatch in1 --no-active-low # 不翻转极性 + + # IN2（sysfs 轮询） + python3 -m diwatch in2 + python3 -m diwatch in2 0.005 + python3 -m diwatch in2 --check + + # 同时监听 + python3 -m diwatch both +""" + +import argparse +import datetime +import os +import signal +import sys +import time +from pathlib import Path + +from .backends import ( + EventBackend, SysfsBackend, + make_in1_backend, find_status_file, find_event_device, +) +from .core import DIChannel, fmt_dur +from .devicetree import derive_gpio +from .config import ( + DEFAULT_EVENT_DEVICE, KEY_UP, DEFAULT_INTERVAL, + IN1_INDEX, IN2_INDEX, IN2_GPIO_NUM, IN2_GPIOCHIP, + STATUS_CANDIDATES, +) + + +# --------------------------------------------------------------------------- +# 构造通道 +# --------------------------------------------------------------------------- +def build_in1(args, log): + backend = make_in1_backend( + device=args.device, + key=args.key, + in_index=IN1_INDEX, + ) + ch = DIChannel( + "IN1", backend, + active_low=args.active_low, + log=log, + ) + if args.press_command: + ch.on_press = lambda ts: run_cmd(args.press_command) + if args.release_command: + ch.on_release = lambda ts: run_cmd(args.release_command) + return ch + + +def build_in2(args, log): + dev = find_status_file(IN2_INDEX) + if not dev: + return None + backend = SysfsBackend( + dev, + interval=args.interval or DEFAULT_INTERVAL, + ) + return DIChannel("IN2", backend, active_low=False, log=log) + + +# --------------------------------------------------------------------------- +# 子命令 +# --------------------------------------------------------------------------- +def cmd_in1(args): + log = make_logger("IN1") + ch = build_in1(args, log) + + install_signal(ch) + + # 手动 open，便于给出友好错误 + try: + ch.open() + except FileNotFoundError as e: + _print_open_error("IN1", e) + return 1 + except OSError as e: + print(f"错误：打开 IN1 失败：{e}", file=sys.stderr) + return 1 + + try: + print(f"监听 {ch.backend.name} key={args.key} " + f"active_low={args.active_low}，Ctrl-C 退出", flush=True) + ch.run(duration=args.duration or None) + except KeyboardInterrupt: + pass + finally: + ch.close() + print_summary([ch]) + return 0 + + +def cmd_in2(args): + log = make_logger("IN2") + ch = build_in2(args, log) + if ch is None: + print(f"错误：找不到 IN{IN2_INDEX} 状态文件。已尝试：", file=sys.stderr) + for c in STATUS_CANDIDATES[IN2_INDEX]: + print(f" {c}", file=sys.stderr) + print("排查：ls /sys/bus/platform/devices/ | grep gpio-in", + file=sys.stderr) + return 1 + + if args.check: + return do_check_in2(ch) + + install_signal(ch) + + try: + ch.open() + except FileNotFoundError as e: + _print_open_error("IN2", e) + return 1 + except OSError as e: + print(f"错误：打开 IN2 失败：{e}", file=sys.stderr) + return 1 + + try: + print("=" * 60) + print(f"IN{IN2_INDEX} 监听开始 {datetime.datetime.now():%Y-%m-%d %H:%M:%S}") + print(f" IN{IN2_INDEX} = GPIO {IN2_GPIO_NUM} = " + f"gpiochip{IN2_GPIOCHIP} 第 {IN2_GPIO_NUM - IN2_GPIOCHIP} 线") + print(f" 状态文件：{ch.backend.path}") + print(f" 轮询间隔：{ch.backend.interval}s " + f"抗抖：连续 {ch.backend.debounce} 次采样一致") + print("=" * 60) + ch.run(duration=args.duration or None) + except KeyboardInterrupt: + pass + finally: + ch.close() + print_summary([ch]) + return 0 + + +def cmd_both(args): + log = make_logger("BOTH") + ch1 = build_in1(args, log) + ch2 = build_in2(args, log) + if ch2 is None: + print(f"错误：找不到 IN{IN2_INDEX} 状态文件。", file=sys.stderr) + return 1 + + install_signal(ch1, ch2) + + # 逐个 open，便于报错 + for name, ch in (("IN1", ch1), ("IN2", ch2)): + try: + ch.open() + except FileNotFoundError as e: + _print_open_error(name, e) + return 1 + except OSError as e: + print(f"错误：打开 {name} 失败：{e}", file=sys.stderr) + return 1 + + try: + print("=" * 60) + print(f"同时监听 IN1 + IN2 {datetime.datetime.now():%Y-%m-%d %H:%M:%S}") + print(f" IN1 -> {ch1.backend.name}") + print(f" IN2 -> {ch2.backend.name}") + print("=" * 60) + + import threading + duration = args.duration or None + t1 = threading.Thread(target=ch1.run, kwargs={"duration": duration}) + t2 = threading.Thread(target=ch2.run, kwargs={"duration": duration}) + t1.start(); t2.start() + while t1.is_alive() or t2.is_alive(): + t1.join(0.5); t2.join(0.5) + except KeyboardInterrupt: + ch1.request_stop(); ch2.request_stop() + finally: + ch1.close(); ch2.close() + print_summary([ch1, ch2]) + return 0 + + +# --------------------------------------------------------------------------- +# 辅助 +# --------------------------------------------------------------------------- +def make_logger(tag): + def _log(msg): + print(f"[{tag}] {msg}", flush=True) + return _log + + +def run_cmd(cmd): + import subprocess + subprocess.Popen(cmd, shell=True) + + +def install_signal(*channels): + def _handler(*_a): + for c in channels: + c.request_stop() + signal.signal(signal.SIGINT, _handler) + signal.signal(signal.SIGTERM, _handler) + + +def _print_open_error(name, exc): + print(f"错误：{name} 不可用：{getattr(exc, 'filename', exc)}", + file=sys.stderr) + print("", file=sys.stderr) + print("排查步骤：", file=sys.stderr) + print(" 1) 查看 event 设备及其名字：", file=sys.stderr) + print(" for f in /dev/input/event*; do \\", file=sys.stderr) + print(" echo \"$f -> $(cat /sys/class/input/$(basename $f)/device/name 2>/dev/null)\"; \\", + file=sys.stderr) + print(" done", file=sys.stderr) + print(" 2) 查看 sysfs 状态文件：", file=sys.stderr) + print(" ls /sys/bus/platform/devices/ | grep gpio-in", file=sys.stderr) + print(" 3) 手动指定 event 设备：", file=sys.stderr) + print(" python3 -m diwatch in1 -d /dev/input/eventN", file=sys.stderr) + print(" 4) 若 SE9 无 /dev/input/，本工具会自动回退到 " + "gpio-in-1/gpio_status", file=sys.stderr) + + +def print_summary(channels): + print() + print("=" * 60) + print(f"汇总 {datetime.datetime.now():%Y-%m-%d %H:%M:%S}") + for ch in channels: + s = ch.summary() + print(f" [{s['name']}] 跳变 {s['count']} 次 " + f"低 {fmt_dur(s['ms_low'])} 高 {fmt_dur(s['ms_high'])} " + f"结束电平 {s['level']}") + print("=" * 60) + + +# --------------------------------------------------------------------------- +# --check for IN2 +# --------------------------------------------------------------------------- +def do_check_in2(ch): + print(f"=== IN{IN2_INDEX} 编号自检 ===") + print() + print(f"状态文件：{ch.backend.path}") + print() + print("--- 从设备树现场复算 ---") + ok = True + for n in (1, 2): + got = derive_gpio(n) + if got is None: + print(f" IN{n}：复算失败（设备树不可读或节点缺失）") + ok = False + continue + gpio_num, label, pin = got + expect = 481 + n + if gpio_num == expect: + print(f" IN{n} -> GPIO {gpio_num} ✓ 与预期一致") + else: + print(f" IN{n} -> GPIO {gpio_num} ✗ 预期 {expect}") + ok = False + print() + if ok: + print(f"结论：IN{IN2_INDEX} = GPIO {IN2_GPIO_NUM}，推导成立。") + else: + print("结论：推导与实测不符，请人工核对。") + print() + print("--- 当前电平 ---") + raw = ch.backend.read_raw() + if raw is not None: + print(f" gpio_status = {raw} （全局 GPIO {IN2_GPIO_NUM}）") + print(" 注：待机时读到 0 是正常的。") + else: + print(f" 读取失败：{ch.backend.path} 内容不是 0/1。") + return 0 + + +# --------------------------------------------------------------------------- +# 入口 +# --------------------------------------------------------------------------- +def build_parser(): + p = argparse.ArgumentParser( + prog="diwatch", + description="统一 DI 输入监听（适配 SE5 / SE9）", + ) + sub = p.add_subparsers(dest="cmd", required=True) + + # ---- in1 ---- + p1 = sub.add_parser("in1", help="监听 IN1（自动适配 event / sysfs）") + p1.add_argument("-d", "--device", default="auto", + help="事件设备路径；auto=自动查找（默认）。" + "SE5 用 /dev/input/eventN，SE9 自动回退到 sysfs。") + p1.add_argument("-k", "--key", type=lambda s: int(s, 0), default=KEY_UP, + help="键码，默认 103 (KEY_UP)") + p1.add_argument("-t", "--duration", type=float, default=0, + help="监听秒数，0=一直跑") + p1.add_argument("--press-command", help="有电时执行的 shell 命令") + p1.add_argument("--release-command", help="断电时执行的 shell 命令") + p1.add_argument("--active-low", dest="active_low", action="store_true", + default=True, help="12V 接通上报为“有电”（默认）") + p1.add_argument("--no-active-low", dest="active_low", action="store_false", + help="不翻转，按驱动/文件原始语义") + p1.set_defaults(func=cmd_in1, interval=None) + + # ---- in2 ---- + p2 = sub.add_parser("in2", help="监听 IN2（sysfs gpio_status）") + p2.add_argument("interval", nargs="?", type=float, default=DEFAULT_INTERVAL, + help=f"轮询间隔秒（默认 {DEFAULT_INTERVAL}）") + p2.add_argument("--check", action="store_true", + help="只自检：复算编号 + 打印当前电平") + p2.add_argument("-t", "--duration", type=float, default=0, + help="监听秒数，0=一直跑") + p2.set_defaults(func=cmd_in2) + + # ---- both ---- + p3 = sub.add_parser("both", help="同时监听 IN1 + IN2") + p3.add_argument("-d", "--device", default="auto", + help="IN1 事件设备路径；auto=自动（默认）") + p3.add_argument("-k", "--key", type=lambda s: int(s, 0), default=KEY_UP) + p3.add_argument("--active-low", dest="active_low", action="store_true", + default=True) + p3.add_argument("--no-active-low", dest="active_low", action="store_false") + p3.add_argument("-t", "--duration", type=float, default=0) + p3.add_argument("--interval", type=float, default=DEFAULT_INTERVAL, + help="IN2 轮询间隔秒") + p3.set_defaults(func=cmd_both) + return p + + +def main(argv=None): + args = build_parser().parse_args(argv) + return args.func(args) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/aibox_sophon_adapter/diwatch/config.py b/aibox_sophon_adapter/diwatch/config.py new file mode 100644 index 0000000..7cd9a07 --- /dev/null +++ b/aibox_sophon_adapter/diwatch/config.py @@ -0,0 +1,63 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""全局常量与默认值。""" + +import os + +# ---- input 事件协议（IN1/event 后端用）---- +EV_KEY = 0x01 +KEY_UP = 103 +VALUE_RELEASE = 0 +VALUE_PRESS = 1 +VALUE_REPEAT = 2 + +DEFAULT_EVENT_DEVICE = "/dev/input/event1" +GPIO_KEYS_NAME = "gpio-keys" + +# ---- IN 通道编号 ---- +IN1_INDEX = 1 +IN2_INDEX = 2 + +# IN2 硬件信息（用于 --check 复算） +IN2_GPIO_NUM = 483 +IN2_GPIOCHIP = 480 + +# ---- sysfs 状态文件候选路径 ---- +# 按 in_index 分组，每个通道可以有多个候选路径，按顺序尝试 +STATUS_CANDIDATES = { + 1: [ + "/sys/bus/platform/devices/gpio-in-1/gpio_status", + "/dev/tv_gpioin_1", + # SE9 若有其它路径，加到这里 + ], + 2: [ + "/sys/bus/platform/devices/gpio-in-2/gpio_status", + "/dev/tv_gpioin_2", + ], +} + +STATUS_FILE_ENV = "DI_STATUS_FILE" # 测试用覆盖 +LOG_DIR_ENV = "DI_LOG_DIR" + +DEVICETREE_BASE = "/sys/firmware/devicetree/base" + +DEBOUNCE = 2 # 连续多少次采样一致才确认跳变 +DEFAULT_INTERVAL = 0.02 # 轮询间隔秒 + + +def candidate_status_files(in_index: int): + """返回某通道的所有候选状态文件路径（含环境变量覆盖）。""" + paths = [] + env = os.environ.get(STATUS_FILE_ENV) + if env: + paths.append(env) + paths.extend(STATUS_CANDIDATES.get(in_index, [])) + return paths + + +def find_status_file(in_index: int): + """找到第一个可读的状态文件；找不到返回 None。""" + for p in candidate_status_files(in_index): + if os.access(p, os.R_OK): + return p + return None diff --git a/aibox_sophon_adapter/diwatch/core.py b/aibox_sophon_adapter/diwatch/core.py new file mode 100644 index 0000000..36779c5 --- /dev/null +++ b/aibox_sophon_adapter/diwatch/core.py @@ -0,0 +1,204 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""DIChannel：统一的数字输入通道。 + +把不同后端（event / sysfs）归一成同一套 API： + + ch = DIChannel("IN1", backend=make_in1_backend(), active_low=True) + ch.on_change = lambda old, new, ts: ... + ch.run() # 阻塞 + ch.run(duration=10) # 跑 10 秒 + +电平语义： + 对外统一用 0/1，0 = 低/断电，1 = 高/有电。 + active_low=True 时，把底层原始值取反（适配光耦反相）。 +""" + +import datetime +import select +import time + + +def now_str(): + return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] + + +def fmt_dur(ms): + if ms < 1000: + return f"{ms}ms" + return f"{ms / 1000:.3f}s" + + +class DIChannel: + """单个数字输入通道。""" + + def __init__(self, name, backend, active_low=False, + on_change=None, on_press=None, on_release=None, + on_event=None, log=None): + """ + name : 通道名，如 "IN1" + backend : Backend 实例 + active_low : True 时把原始值取反（0->1, 1->0） + on_change : (old, new, ts) -> None + on_press : (ts) -> None # new=1 + on_release : (ts) -> None # new=0 + on_event : (ts, raw) -> None # 原始事件 + log : (msg) -> None + """ + self.name = name + self.backend = backend + self.active_low = active_low + self.on_change = on_change + self.on_press = on_press + self.on_release = on_release + self.on_event = on_event + self.log = log or (lambda m: print(m, flush=True)) + + self._level = None + self._level_since = None + self._stop = False + self._count = 0 + self._ms_low = 0 + self._ms_high = 0 + + # ---- 上下文管理 ---- + def __enter__(self): + self.open() + return self + + def __exit__(self, *exc): + self.close() + + def open(self): + self.backend.open() + + def close(self): + self.backend.close() + + # ---- 电平换算 ---- + def _to_level(self, raw): + if raw is None: + return None + v = 1 if raw else 0 + if self.active_low: + v = 1 - v + return v + + # ---- 回调分发 ---- + def _emit(self, new, ts): + old = self._level + if old == new: + return + self._level = new + now_ms = int(time.time() * 1000) + if self._level_since is not None: + dur = now_ms - self._level_since + if old == 0: + self._ms_low += dur + elif old == 1: + self._ms_high += dur + self._level_since = now_ms + self._count += 1 + + self.log(f"{now_str()} {self.name} {'有电' if new else '断电'}") + + if self.on_change: + self.on_change(old, new, ts) + if new == 1 and self.on_press: + self.on_press(ts) + elif new == 0 and self.on_release: + self.on_release(ts) + + # ---- 单次读取（用于初始化）---- + def read_now(self): + """立即读一次并返回归一化后的电平；失败返回 None。 + + 仅对支持 read_raw 的后端（sysfs）有效；event 后端返回 None。 + """ + raw = None + if hasattr(self.backend, "read_raw"): + raw = self.backend.read_raw() + if raw is None: + return None + lvl = self._to_level(int(raw)) + self._level = lvl + self._level_since = int(time.time() * 1000) + return lvl + + # ---- 主循环 ---- + def run(self, duration=None): + """阻塞监听。duration 秒后返回；None = 一直跑到 Ctrl-C。""" + self.open() + + # 初始化当前电平 + if self._level is None: + lvl = self.read_now() + if lvl is not None: + self.log(f"{self.name} 初始电平：{lvl}") + else: + # event 后端：置为 0，等第一个事件再校正 + self._level = 0 + self._level_since = int(time.time() * 1000) + + deadline = time.time() + duration if duration else None + + try: + if self.backend.can_poll: + self._run_poll(deadline) + else: + self._run_sample(deadline) + finally: + self.close() + + def _run_poll(self, deadline): + poller = select.poll() + poller.register(self.backend.fileno(), select.POLLIN) + while not self._stop: + timeout = 500 + if deadline: + remain = deadline - time.time() + if remain <= 0: + break + timeout = int(min(remain, 0.5) * 1000) + poller.poll(timeout) + got = self.backend.read() + if got: + raw, ts = got + if self.on_event: + self.on_event(ts, raw) + self._emit(self._to_level(raw), ts) + + def _run_sample(self, deadline): + interval = getattr(self.backend, "interval", 0.02) + 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) + if self.on_event: + self.on_event(ts, raw) + self._emit(lvl, ts) + time.sleep(interval) + + def request_stop(self): + self._stop = True + + # ---- 汇总 ---- + def summary(self): + now_ms = int(time.time() * 1000) + if self._level_since is not None: + dur = now_ms - self._level_since + if self._level == 0: + self._ms_low += dur + elif self._level == 1: + self._ms_high += dur + self._level_since = now_ms + return { + "name": self.name, + "count": self._count, + "ms_low": self._ms_low, + "ms_high": self._ms_high, + "level": self._level, + } diff --git a/aibox_sophon_adapter/diwatch/devicetree.py b/aibox_sophon_adapter/diwatch/devicetree.py new file mode 100644 index 0000000..389396b --- /dev/null +++ b/aibox_sophon_adapter/diwatch/devicetree.py @@ -0,0 +1,66 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""从设备树复算 gpio-in-N 对应的全局 GPIO 号。""" + +from pathlib import Path + +from .config import DEVICETREE_BASE + + +def _read_be32(path: Path, offset: int = 0): + try: + data = path.read_bytes() + except OSError: + return None + if len(data) < offset + 4: + return None + return int.from_bytes(data[offset:offset + 4], "big") + + +def derive_gpio(n: int, base_dir: str = DEVICETREE_BASE): + """从设备树复算 IN 的全局 GPIO 号。 + + 返回 (gpio_num, label, pin) 或 None。 + """ + base = Path(base_dir) + f = 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 + + ph = int.from_bytes(data[0:4], "big") + pin = int.from_bytes(data[4:8], "big") + + # phandle -> gpio-controller 节点 + ctrl = None + for c in base.glob("gpio@*/gpio-controller@*"): + cph = _read_be32(c / "phandle") + if cph == ph: + ctrl = c + break + if ctrl is None: + return None + + bank = ctrl.parent.name + if bank.startswith("gpio@"): + bank = bank[len("gpio@"):] + label = f"{bank}.gpio" + + # label -> gpiochip base + gpio_class = Path("/sys/class/gpio") + if not gpio_class.exists(): + return None + for g in gpio_class.glob("gpiochip*"): + try: + if (g / "label").read_text().strip() == label: + chip_base = int((g / "base").read_text().strip()) + return chip_base + pin, label, pin + except (OSError, ValueError): + continue + return None diff --git a/aibox_sophon_adapter/in1.py b/aibox_sophon_adapter/in1.py new file mode 100644 index 0000000..931056c --- /dev/null +++ b/aibox_sophon_adapter/in1.py @@ -0,0 +1,213 @@ +#!/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 new file mode 100644 index 0000000..f41da49 --- /dev/null +++ b/aibox_sophon_adapter/in11.py @@ -0,0 +1,456 @@ +#!/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