75 lines
2.4 KiB
Python
75 lines
2.4 KiB
Python
import grovepi
|
|
import time
|
|
import threading
|
|
from sensors.lock import grove_lock
|
|
from shared.logging import log
|
|
|
|
button = 2
|
|
grovepi.pinMode(button, "INPUT")
|
|
|
|
button_callback = None
|
|
|
|
def read_button_state():
|
|
# Increase timeout slightly so the button thread can wait for long I2C sensor reads to finish
|
|
if not grove_lock.acquire(timeout=0.2):
|
|
return None
|
|
try:
|
|
val = grovepi.digitalRead(button)
|
|
# Force strict binary output (0 or 1)
|
|
return 1 if val == 1 else 0
|
|
except Exception as e:
|
|
log(f"BTN Error: {e}")
|
|
return None
|
|
finally:
|
|
grove_lock.release()
|
|
|
|
def monitor_button():
|
|
last_stable_state = 0
|
|
candidate_state = 0
|
|
consecutive_count = 0
|
|
|
|
# Require 3 consecutive identical reads (~60ms) to confirm a valid state change
|
|
REQUIRED_CONSECUTIVE_READS = 3
|
|
|
|
# Minimum time gap (in seconds) between allowed button triggers (cooldown)
|
|
DEBOUNCE_COOLDOWN = 0.3
|
|
last_trigger_time = 0
|
|
|
|
while True:
|
|
current_state = read_button_state()
|
|
|
|
if current_state is not None:
|
|
if current_state == candidate_state:
|
|
consecutive_count += 1
|
|
else:
|
|
candidate_state = current_state
|
|
consecutive_count = 1
|
|
|
|
# State is confirmed stable across multiple reads
|
|
if consecutive_count >= REQUIRED_CONSECUTIVE_READS:
|
|
now = time.time()
|
|
|
|
# Rising edge detection (0 -> 1 transition) with cooldown timer
|
|
if candidate_state == 1 and last_stable_state == 0:
|
|
if (now - last_trigger_time) > DEBOUNCE_COOLDOWN:
|
|
last_trigger_time = now
|
|
if button_callback:
|
|
try:
|
|
button_callback()
|
|
except Exception as e:
|
|
log(f"[Button] Callback exception: {e}")
|
|
|
|
last_stable_state = candidate_state
|
|
|
|
time.sleep(0.02) # 20ms poll interval
|
|
else:
|
|
# Lock was busy or read failed; reset candidate counter to reject noisy spikes
|
|
consecutive_count = 0
|
|
time.sleep(0.01)
|
|
|
|
def start_button_monitoring_thread():
|
|
threading.Thread(target=monitor_button, daemon=True).start()
|
|
|
|
def set_callback(callback):
|
|
global button_callback
|
|
button_callback = callback |