modify: DI 输入监听
This commit is contained in:
@@ -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<n> 的全局 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())
|
||||
Reference in New Issue
Block a user