69 lines
2.0 KiB
Python
69 lines
2.0 KiB
Python
import grovepi
|
|
import time
|
|
import threading
|
|
from sensors.lock import safe_grove_access
|
|
from shared.logging import log
|
|
|
|
button = 2
|
|
grovepi.pinMode(button, "INPUT")
|
|
|
|
button_callback = None
|
|
|
|
def read_button_state():
|
|
# Attempt to acquire the lock with a 0.2s timeout
|
|
with safe_grove_access(timeout=0.2) as acquired:
|
|
if not acquired:
|
|
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
|
|
|
|
def monitor_button():
|
|
last_stable_state = 0
|
|
candidate_state = 0
|
|
consecutive_count = 0
|
|
|
|
REQUIRED_CONSECUTIVE_READS = 3
|
|
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
|
|
|
|
if consecutive_count >= REQUIRED_CONSECUTIVE_READS:
|
|
now = time.time()
|
|
|
|
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:
|
|
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 |