201 lines
5.9 KiB
Python
201 lines
5.9 KiB
Python
#!/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)
|
||
# read() 返回本批全部事件，按序处理，快速抖动不丢中间跳变
|
||
for raw, ts in self.backend.read():
|
||
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
|
||
for raw, ts in self.backend.read():
|
||
if self.on_event:
|
||
self.on_event(ts, raw)
|
||
self._emit(self._to_level(raw), 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,
|
||
}
|