54 lines
1.8 KiB
Python
54 lines
1.8 KiB
Python
import time
|
|
import threading
|
|
import grovepi
|
|
from sensors.lock import safe_grove_access
|
|
from shared import config
|
|
|
|
BUZZER_PIN = 8 # Must be a PWM pin on GrovePi (D3, D5, D6, D8 support analogWrite)
|
|
|
|
# Initialize pin mode safely
|
|
try:
|
|
with safe_grove_access(timeout=1.0) as acquired:
|
|
if acquired:
|
|
grovepi.pinMode(BUZZER_PIN, "OUTPUT")
|
|
except Exception as e:
|
|
print(f"[BUZZER] Init error: {e}")
|
|
|
|
def _stop_sound():
|
|
"""Helper to ensure sound is silenced safely under lock."""
|
|
with safe_grove_access(timeout=1.0) as acquired:
|
|
if acquired:
|
|
try:
|
|
grovepi.analogWrite(BUZZER_PIN, 0)
|
|
grovepi.digitalWrite(BUZZER_PIN, 0)
|
|
except Exception:
|
|
pass
|
|
|
|
def _siren_worker(beats_nb: int, delay: float):
|
|
if not config.BUZZER_ACTIVATED:
|
|
return
|
|
|
|
try:
|
|
for _ in range(beats_nb):
|
|
# 1. Turn sound ON (only acquire lock briefly for the I2C write)
|
|
with safe_grove_access(timeout=1.0) as acquired:
|
|
if acquired:
|
|
grovepi.analogWrite(BUZZER_PIN, 128)
|
|
time.sleep(delay) # Sleep WITHOUT holding the lock!
|
|
|
|
# 2. Turn sound OFF
|
|
with safe_grove_access(timeout=1.0) as acquired:
|
|
if acquired:
|
|
grovepi.analogWrite(BUZZER_PIN, 0)
|
|
time.sleep(delay) # Sleep WITHOUT holding the lock!
|
|
except Exception as e:
|
|
print(f"[BUZZER] Error during siren execution: {e}")
|
|
finally:
|
|
_stop_sound()
|
|
|
|
def buzzer_siren(beats_nb: int = 3, delay: float = 0.2, async_run: bool = True):
|
|
if async_run:
|
|
thread = threading.Thread(target=_siren_worker, args=(beats_nb, delay), daemon=True)
|
|
thread.start()
|
|
else:
|
|
_siren_worker(beats_nb, delay) |