45 lines
1.4 KiB
Python
45 lines
1.4 KiB
Python
import time
|
|
import threading
|
|
import grovepi
|
|
from sensors.lock import grove_lock
|
|
from shared import config
|
|
|
|
BUZZER_PIN = 8 # Must be a PWM pin on GrovePi (D3, D5, D6, D8 support analogWrite)
|
|
|
|
# Initialize pin mode
|
|
try:
|
|
with grove_lock:
|
|
grovepi.pinMode(BUZZER_PIN, "OUTPUT")
|
|
except Exception as e:
|
|
print(f"[BUZZER] Init error: {e}")
|
|
|
|
def _siren_worker(beats_nb: int, delay: float):
|
|
if not config.BUZZER_ACTIVATED:
|
|
return
|
|
|
|
with grove_lock:
|
|
try:
|
|
for _ in range(beats_nb):
|
|
# 1. Send single PWM write to start sound (Duty cycle ~128/255)
|
|
grovepi.analogWrite(BUZZER_PIN, 128)
|
|
time.sleep(delay)
|
|
|
|
# 2. Send single PWM write to kill sound
|
|
grovepi.analogWrite(BUZZER_PIN, 0)
|
|
time.sleep(delay)
|
|
except Exception as e:
|
|
print(f"[BUZZER] Error during siren execution: {e}")
|
|
finally:
|
|
# Absolute safety reset
|
|
try:
|
|
grovepi.analogWrite(BUZZER_PIN, 0)
|
|
grovepi.digitalWrite(BUZZER_PIN, 0)
|
|
except Exception:
|
|
pass
|
|
|
|
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) |