100 lines
2.7 KiB
Python
Executable File
100 lines
2.7 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
import time
|
|
import signal
|
|
import sys
|
|
from collections import deque
|
|
|
|
import revpimodio2
|
|
|
|
|
|
RUNNING = True
|
|
|
|
|
|
def handle_signal(signum, frame):
|
|
global RUNNING
|
|
RUNNING = False
|
|
|
|
|
|
signal.signal(signal.SIGTERM, handle_signal)
|
|
signal.signal(signal.SIGINT, handle_signal)
|
|
|
|
|
|
def main():
|
|
rpi = revpimodio2.RevPiModIO(autorefresh=True)
|
|
|
|
auto_io = getattr(rpi.io, "I_1")
|
|
cycle_io = getattr(rpi.io, "I_2")
|
|
|
|
cycle_pulse_count = 0
|
|
previous_cycle = bool(cycle_io.value)
|
|
last_edge_ts = 0.0
|
|
last_cycle_ts = None
|
|
|
|
debounce_s = 0.05
|
|
pulse_window_s = 60.0
|
|
recent_pulses = deque()
|
|
|
|
print("[START] DI probe")
|
|
print("I_1 = machine_auto_signal")
|
|
print("I_2 = cycle_pulse")
|
|
print("Pulsa CTRL+C para salir")
|
|
|
|
try:
|
|
last_print = 0.0
|
|
|
|
while RUNNING:
|
|
now = time.monotonic()
|
|
|
|
machine_auto_signal = bool(auto_io.value)
|
|
current_cycle = bool(cycle_io.value)
|
|
|
|
# Flanco ascendente en I_2
|
|
if current_cycle and not previous_cycle:
|
|
if now - last_edge_ts >= debounce_s:
|
|
cycle_pulse_count += 1
|
|
last_edge_ts = now
|
|
last_cycle_ts = now
|
|
recent_pulses.append(now)
|
|
|
|
previous_cycle = current_cycle
|
|
|
|
# Limpiar pulsos fuera de la ventana
|
|
while recent_pulses and now - recent_pulses[0] > pulse_window_s:
|
|
recent_pulses.popleft()
|
|
|
|
cycle_rate_ppm = len(recent_pulses) * (60.0 / pulse_window_s)
|
|
|
|
if last_cycle_ts is None:
|
|
last_cycle_age_s = None
|
|
else:
|
|
last_cycle_age_s = round(now - last_cycle_ts, 1)
|
|
|
|
# Para la demo, machine_running se asocia al interruptor I_1.
|
|
# Más adelante podemos hacerlo depender también de pulsos recientes.
|
|
machine_running = machine_auto_signal
|
|
|
|
if now - last_print >= 0.5:
|
|
last_print = now
|
|
print(
|
|
"auto={auto} running={running} cycle_input={cycle} "
|
|
"pulse_count={count} cycle_rate_ppm={rate:.1f} last_cycle_age_s={age}".format(
|
|
auto=int(machine_auto_signal),
|
|
running=int(machine_running),
|
|
cycle=int(current_cycle),
|
|
count=cycle_pulse_count,
|
|
rate=cycle_rate_ppm,
|
|
age=last_cycle_age_s,
|
|
),
|
|
flush=True,
|
|
)
|
|
|
|
time.sleep(0.02)
|
|
|
|
finally:
|
|
rpi.exit()
|
|
print("[STOP] DI probe")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|