487 lines
18 KiB
Python
487 lines
18 KiB
Python
#!/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/SE9；SE5 无 gpio-in-2 时回退 gpio-keys event）
|
||
python3 -m diwatch in2
|
||
python3 -m diwatch in2 0.005
|
||
python3 -m diwatch in2 --check
|
||
python3 -m diwatch in2 -k 108 # 指定 event 键码
|
||
python3 -m diwatch detect # 扫描全部 event 设备，查 IN1/IN2 实际键码
|
||
|
||
# 同时监听
|
||
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, EVENT_STRUCT,
|
||
make_in1_backend, make_in2_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, DEFAULT_IN2_KEY,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 构造通道
|
||
# ---------------------------------------------------------------------------
|
||
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):
|
||
backend = make_in2_backend(
|
||
device=getattr(args, "device", None),
|
||
key=getattr(args, "key", None),
|
||
in_index=IN2_INDEX,
|
||
interval=args.interval or DEFAULT_INTERVAL,
|
||
)
|
||
if backend is None:
|
||
return None
|
||
ch = DIChannel(
|
||
"IN2", backend,
|
||
active_low=getattr(args, "active_low", False),
|
||
log=log,
|
||
)
|
||
if getattr(args, "press_command", None):
|
||
ch.on_press = lambda ts: run_cmd(args.press_command)
|
||
if getattr(args, "release_command", None):
|
||
ch.on_release = lambda ts: run_cmd(args.release_command)
|
||
return ch
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 子命令
|
||
# ---------------------------------------------------------------------------
|
||
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" sysfs: {c}", file=sys.stderr)
|
||
print(f" event: 自动查找 gpio-keys（默认键码 {DEFAULT_IN2_KEY}）", file=sys.stderr)
|
||
print("排查：ls /sys/bus/platform/devices/ | grep gpio-in", file=sys.stderr)
|
||
print(" ls -l /dev/input/event*", file=sys.stderr)
|
||
print(f" python3 -m diwatch detect # 查 IN2 实际键码", 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)
|
||
print("若为 Permission denied，把当前用户加入 input 组："
|
||
"sudo usermod -aG input $USER 后重新登录", file=sys.stderr)
|
||
return 1
|
||
|
||
try:
|
||
print("=" * 60)
|
||
print(f"IN{IN2_INDEX} 监听开始 {datetime.datetime.now():%Y-%m-%d %H:%M:%S}")
|
||
if isinstance(ch.backend, SysfsBackend):
|
||
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} 次采样一致")
|
||
else:
|
||
print(f" 后端：{ch.backend.name} 键码：{ch.backend.key}")
|
||
print(" 提示：若操作 IN2 无事件输出，运行 python3 -m diwatch detect 查实际键码")
|
||
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()
|
||
be = ch.backend
|
||
if not isinstance(be, SysfsBackend):
|
||
print(f"后端：{be.name} 键码：{be.key}")
|
||
print("event 后端无初始电平可读；操作一次 IN2 应出现 value=1（按下/有电）与 0（释放/断电）。")
|
||
print(f"若无任何输出，说明键码不是 {be.key}，用 python3 -m diwatch detect 查实际键码。")
|
||
return 0
|
||
|
||
print(f"状态文件：{be.path}")
|
||
print()
|
||
print("--- 从设备树现场复算 ---")
|
||
ok = True
|
||
for n in (1, 2):
|
||
got = derive_gpio(n)
|
||
if got is None:
|
||
print(f" IN{n}：复算失败（设备树不可读或节点缺失，SE5 无此节点属正常）")
|
||
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("结论：推导与实测不符（SE9 正常成立；SE5 走 event 后端，此节可忽略）。")
|
||
print()
|
||
print("--- 当前电平 ---")
|
||
raw = be.read_raw()
|
||
if raw is not None:
|
||
print(f" gpio_status = {raw} （全局 GPIO {IN2_GPIO_NUM}）")
|
||
print(" 注：待机时读到 0 是正常的。")
|
||
else:
|
||
print(f" 读取失败：{be.path} 内容不是 0/1。")
|
||
return 0
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# detect：扫描全部 event 设备，实时打印按键事件（确认 IN1/IN2 键码）
|
||
# ---------------------------------------------------------------------------
|
||
KEY_NAMES = {
|
||
1: "KEY_ESC", 28: "KEY_ENTER", 103: "KEY_UP", 108: "KEY_DOWN",
|
||
105: "KEY_LEFT", 106: "KEY_RIGHT", 114: "KEY_VOLUMEDOWN",
|
||
115: "KEY_VOLUMEUP", 116: "KEY_POWER",
|
||
}
|
||
|
||
|
||
def cmd_detect(args):
|
||
"""扫描可读的 /dev/input/event* 设备，监听 duration 秒并打印所有 EV_KEY 事件。"""
|
||
import glob
|
||
import select
|
||
|
||
devices = sorted(glob.glob("/dev/input/event*"))
|
||
if not devices:
|
||
print("错误：/dev/input/ 下没有 event 设备", file=sys.stderr)
|
||
return 1
|
||
|
||
names = {} # fd -> (device, 名称)
|
||
poller = select.poll()
|
||
for dev in devices:
|
||
try:
|
||
fd = os.open(dev, os.O_RDONLY | os.O_NONBLOCK)
|
||
except PermissionError:
|
||
print(f" 跳过 {dev}（权限不足；用 sudo 运行，或把用户加入 input 组）",
|
||
file=sys.stderr)
|
||
continue
|
||
except OSError as e:
|
||
print(f" 跳过 {dev}（{e}）", file=sys.stderr)
|
||
continue
|
||
name = ""
|
||
try:
|
||
with open(f"/sys/class/input/{os.path.basename(dev)}/device/name") as f:
|
||
name = f.read().strip()
|
||
except OSError:
|
||
pass
|
||
names[fd] = (dev, name)
|
||
poller.register(fd, select.POLLIN)
|
||
|
||
if not names:
|
||
print("错误：没有可读的 event 设备（Permission denied 时用 sudo 运行）", file=sys.stderr)
|
||
return 1
|
||
|
||
duration = args.duration or 5.0
|
||
print("=" * 60)
|
||
print(f"监听 {len(names)} 个 event 设备 {duration:g} 秒，请现在操作 IN1 / IN2 ...")
|
||
for fd, (dev, name) in names.items():
|
||
print(f" {dev} [{name}]")
|
||
print("=" * 60)
|
||
|
||
deadline = time.time() + duration
|
||
count = 0
|
||
try:
|
||
while True:
|
||
remain_ms = int((deadline - time.time()) * 1000)
|
||
if remain_ms <= 0:
|
||
break
|
||
if not poller.poll(min(remain_ms, 200)):
|
||
continue
|
||
for fd in list(names):
|
||
try:
|
||
data = os.read(fd, EVENT_STRUCT.size * 64)
|
||
except OSError:
|
||
continue
|
||
if not data:
|
||
continue
|
||
dev, _name = names[fd]
|
||
for off in range(0, len(data) - EVENT_STRUCT.size + 1, EVENT_STRUCT.size):
|
||
_sec, _usec, etype, code, value = EVENT_STRUCT.unpack_from(data, off)
|
||
if etype != 0x01: # 只看 EV_KEY
|
||
continue
|
||
if value == 2: # 长按 repeat 太吵
|
||
continue
|
||
count += 1
|
||
key = KEY_NAMES.get(code, "?")
|
||
val = "按下" if value == 1 else "释放"
|
||
print(f" {datetime.datetime.now():%H:%M:%S.%f} {dev} "
|
||
f"code={code:<4} ({key}) {val}")
|
||
except KeyboardInterrupt:
|
||
pass
|
||
finally:
|
||
for fd in names:
|
||
try:
|
||
os.close(fd)
|
||
except OSError:
|
||
pass
|
||
|
||
print()
|
||
if count == 0:
|
||
print("未捕获任何按键事件。若已操作 IN2 仍无输出，说明 SE5 的 IN2 不走 event 设备，")
|
||
print("请检查：ls /sys/bus/platform/devices/ | grep -i gpio；"
|
||
"sudo cat /sys/kernel/debug/gpio")
|
||
else:
|
||
print(f"共捕获 {count} 个按键事件。IN2 对应的 code 即为 -k/--key（或 task_switch.key）应填的键码。")
|
||
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/SE9，回退 gpio-keys event/SE5）")
|
||
p2.add_argument("interval", nargs="?", type=float, default=DEFAULT_INTERVAL,
|
||
help=f"sysfs 轮询间隔秒（默认 {DEFAULT_INTERVAL}）")
|
||
p2.add_argument("--check", action="store_true",
|
||
help="只自检：复算编号 + 打印当前电平")
|
||
p2.add_argument("-d", "--device", default=None,
|
||
help="显式指定 event 设备（回退 event 模式用）；默认自动选择")
|
||
p2.add_argument("-k", "--key", type=lambda s: int(s, 0), default=None,
|
||
help=f"event 模式键码（默认 {DEFAULT_IN2_KEY}，用 detect 查实际键码）")
|
||
p2.add_argument("--active-low", dest="active_low", action="store_true",
|
||
default=False, help="电平取反（默认不取反）")
|
||
p2.add_argument("--no-active-low", dest="active_low", action="store_false",
|
||
help="不翻转（默认）")
|
||
p2.add_argument("--press-command", help="有电时执行的 shell 命令")
|
||
p2.add_argument("--release-command", help="断电时执行的 shell 命令")
|
||
p2.add_argument("-t", "--duration", type=float, default=0,
|
||
help="监听秒数，0=一直跑")
|
||
p2.set_defaults(func=cmd_in2)
|
||
|
||
# ---- detect ----
|
||
pd = sub.add_parser("detect",
|
||
help="扫描全部 event 设备并打印按键事件（确认 IN1/IN2 键码）")
|
||
pd.add_argument("-t", "--duration", type=float, default=5.0,
|
||
help="监听秒数（默认 5）")
|
||
pd.set_defaults(func=cmd_detect)
|
||
|
||
# ---- 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())
|