271 lines
10 KiB
Python
271 lines
10 KiB
Python
from time import time
|
|
from astral import LocationInfo
|
|
from astral.sun import sun
|
|
from datetime import date, datetime, timedelta
|
|
from zoneinfo import ZoneInfo
|
|
from typing import Callable
|
|
import requests
|
|
import subprocess
|
|
|
|
from grove.gpio import GPIO
|
|
from grove.display.jhd1802 import JHD1802
|
|
|
|
shelly_server = "https://shelly-237-eu.shelly.cloud"
|
|
shelly_token = "M2M5YTYxdWlkEFD82A2301693D62637E530B4CDDC0DC6E2CDD8F01BE5E5102000B92638A2DA74019E6A81E6D17E0"
|
|
|
|
shelly_deviceId = "8cbfea9fd6d0"
|
|
|
|
def setShellyPlugState(deviceId: str, state: bool) -> dict:
|
|
"""
|
|
Set the state of a Shelly plug/switch device.
|
|
|
|
Args:
|
|
deviceId: The Shelly device id
|
|
state: True to turn on, False to turn off
|
|
|
|
Returns:
|
|
Response from the API as a dictionary
|
|
"""
|
|
url = f"{shelly_server}/v2/devices/api/set/switch"
|
|
params = {"auth_key": shelly_token}
|
|
payload = {
|
|
"id": deviceId,
|
|
"on": state
|
|
}
|
|
|
|
response = requests.post(url, params=params, json=payload)
|
|
|
|
if response.status_code == 200:
|
|
return {"success": True}
|
|
else:
|
|
return response.json()
|
|
|
|
city = LocationInfo("Seinäjoki", "Finland", "Europe/Helsinki", 62.7900, 22.8400)
|
|
s = sun(city.observer, date=date.today())
|
|
print(f"City: {city.name}, {city.region}")
|
|
print(f"Timezone: {city.timezone}")
|
|
print(f"Latitude: {city.latitude:.6f}; Longitude: {city.longitude:.6f}")
|
|
|
|
def isSunUp() -> bool:
|
|
now = datetime.now(city.tzinfo)
|
|
return s["sunrise"] <= now <= s["sunset"]
|
|
|
|
def playAudio(filePath: str) -> None:
|
|
"""
|
|
Play an audio file through the Raspberry Pi's standard aux output.
|
|
|
|
Args:
|
|
filePath: Path to the audio file (.wav, .mp3, .ogg, etc.)
|
|
"""
|
|
if filePath.endswith(".wav"):
|
|
subprocess.Popen(["aplay", filePath], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
else:
|
|
# Use mpg123 for mp3, or ffplay/mpv as fallback for other formats
|
|
subprocess.Popen(["mpg123", "-q", filePath], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
|
|
class WakeUpAction:
|
|
offsetSeconds: int # how many seconds before alarm time to trigger
|
|
action: Callable[[], None]
|
|
triggered: bool = False
|
|
|
|
def __init__(self, offsetSeconds: int, action: Callable[[], None]):
|
|
self.offsetSeconds = offsetSeconds
|
|
self.action = action
|
|
self.triggered = False
|
|
|
|
|
|
# sunset and sunrise
|
|
class AlarmClock:
|
|
wakeupActions: list[WakeUpAction] = []
|
|
ledGpio: GPIO
|
|
buttonGpio: GPIO
|
|
relayGpio: GPIO
|
|
lastKnownButtonState: bool = False
|
|
lcd: JHD1802
|
|
lastButtonPressTime: float = 0.0
|
|
lastButtonStateChangeTime: float = 0.0
|
|
debounceDelay: float = 0.05 # 50ms debounce delay
|
|
longPressThreshold: float = 1.0 # 1 second for long press
|
|
buttonPressStartTime: float = 0.0
|
|
configMode: str = "normal" # modes: normal, set_hour, set_minute
|
|
alarmTime: datetime = datetime.now(ZoneInfo("Europe/Helsinki")).replace(hour=9, minute=32, second=0, microsecond=0)
|
|
actionsResetForToday: bool = False
|
|
def __init__(self, pins: dict = {}):
|
|
self.ledGpio = GPIO(pins.get("led", 5), GPIO.OUT)
|
|
self.buttonGpio = GPIO(pins.get("button", 6), GPIO.IN)
|
|
self.relayGpio = GPIO(pins.get("relay", 16), GPIO.OUT)
|
|
self.lcd = JHD1802()
|
|
self.wakeupActions = []
|
|
self.setLcdText("AlarmClock Init")
|
|
|
|
def addWakeUpAction(self, offsetSeconds: int, action: Callable[[], None]):
|
|
"""Add a wakeup action to be triggered offsetSeconds before alarm time"""
|
|
self.wakeupActions.append(WakeUpAction(offsetSeconds, action))
|
|
|
|
def resetWakeUpActions(self):
|
|
"""Reset all wakeup actions so they can trigger again"""
|
|
for action in self.wakeupActions:
|
|
action.triggered = False
|
|
self.actionsResetForToday = True
|
|
print("Wakeup actions reset")
|
|
|
|
def checkWakeUpActions(self):
|
|
"""Check and trigger any wakeup actions that are due"""
|
|
if not self.wakeupActions:
|
|
return
|
|
|
|
now = datetime.now(ZoneInfo(city.timezone))
|
|
today_alarm = now.replace(
|
|
hour=self.alarmTime.hour,
|
|
minute=self.alarmTime.minute,
|
|
second=0,
|
|
microsecond=0
|
|
)
|
|
|
|
# Find the earliest action (largest offset)
|
|
max_offset = max(action.offsetSeconds for action in self.wakeupActions)
|
|
first_trigger_time = today_alarm - timedelta(seconds=max_offset)
|
|
reset_time = first_trigger_time - timedelta(seconds=10)
|
|
alarm_window_end = today_alarm + timedelta(minutes=1)
|
|
|
|
# Reset actions 10 seconds before the first one should trigger
|
|
if now >= reset_time and now < first_trigger_time and not self.actionsResetForToday:
|
|
self.resetWakeUpActions()
|
|
|
|
# After alarm window ends, allow reset to happen again tomorrow
|
|
if now >= alarm_window_end:
|
|
self.actionsResetForToday = False
|
|
|
|
for wakeupAction in self.wakeupActions:
|
|
if wakeupAction.triggered:
|
|
continue
|
|
trigger_time = today_alarm - timedelta(seconds=wakeupAction.offsetSeconds)
|
|
if now >= trigger_time and now < today_alarm + timedelta(minutes=1):
|
|
print(f"Triggering wakeup action (offset: {wakeupAction.offsetSeconds}s)")
|
|
wakeupAction.action()
|
|
wakeupAction.triggered = True
|
|
|
|
|
|
def start(self):
|
|
print("AlarmClock started")
|
|
while True:
|
|
self.loop()
|
|
|
|
def setRelayState(self, state: bool):
|
|
self.relayGpio.write(1 if state else 0)
|
|
|
|
def loop(self):
|
|
currentButtonState = not self.buttonGpio.read()
|
|
currentTime = time()
|
|
|
|
if (self.configMode == "normal"):
|
|
currentMinute = datetime.now(ZoneInfo(city.timezone)).minute
|
|
currentHour = datetime.now(ZoneInfo(city.timezone)).hour
|
|
currrentSecond = datetime.now(ZoneInfo(city.timezone)).second
|
|
|
|
self.setLcdText(f"Time {currentHour:02d}:{currentMinute:02d}:{currrentSecond:02d}\nAlarm {self.alarmTime.hour:02d}:{self.alarmTime.minute:02d}")
|
|
|
|
# Check and trigger wakeup actions
|
|
self.checkWakeUpActions()
|
|
|
|
# Only process button state change if debounce delay has passed
|
|
if currentButtonState != self.lastKnownButtonState:
|
|
if (currentTime - self.lastButtonStateChangeTime) >= self.debounceDelay:
|
|
self.lastKnownButtonState = currentButtonState
|
|
self.lastButtonStateChangeTime = currentTime
|
|
self.onButtonPress("button", currentButtonState)
|
|
|
|
# Check for long press while button is held
|
|
if currentButtonState and self.buttonPressStartTime > 0:
|
|
pressDuration = currentTime - self.buttonPressStartTime
|
|
if pressDuration >= self.longPressThreshold:
|
|
self.onLongPress()
|
|
self.buttonPressStartTime = 0.0 # Reset to prevent repeated triggers
|
|
|
|
def onButtonPress(self, button: str, state: bool):
|
|
print(f"Button '{button}' pressed state: {state}")
|
|
currentTime = time()
|
|
|
|
if button == "button":
|
|
if state: # Button pressed down
|
|
self.buttonPressStartTime = currentTime
|
|
else: # Button released
|
|
if self.buttonPressStartTime > 0:
|
|
pressDuration = currentTime - self.buttonPressStartTime
|
|
self.buttonPressStartTime = 0.0
|
|
|
|
# Only handle as short press if it wasn't a long press
|
|
if pressDuration < self.longPressThreshold:
|
|
self.onShortPress()
|
|
|
|
def onShortPress(self):
|
|
"""Handle short button press based on current mode"""
|
|
if self.configMode == "normal":
|
|
# Normal mode: toggle relay
|
|
self.setRelayState(not self.relayGpio.read())
|
|
setShellyPlugState(shelly_deviceId, self.relayGpio.read() == 0)
|
|
elif self.configMode == "set_hour":
|
|
# Increment hour
|
|
self.alarmTime = self.alarmTime.replace(hour=(self.alarmTime.hour + 1) % 24)
|
|
self.setLcdText(f"Set Hour: {self.alarmTime.hour:02d}")
|
|
elif self.configMode == "set_minute":
|
|
# Increment minute
|
|
self.alarmTime = self.alarmTime.replace(minute=(self.alarmTime.minute + 1) % 60)
|
|
self.setLcdText(f"Set Min: {self.alarmTime.minute:02d}")
|
|
|
|
def onLongPress(self):
|
|
"""Handle long button press - cycle through configuration modes"""
|
|
if self.configMode == "normal":
|
|
self.configMode = "set_hour"
|
|
self.setLcdText(f"Set Hour: {self.alarmTime.hour:02d}")
|
|
print(f"Entering hour setting mode")
|
|
elif self.configMode == "set_hour":
|
|
self.configMode = "set_minute"
|
|
self.setLcdText(f"Set Min: {self.alarmTime.minute:02d}")
|
|
print(f"Switching to minute setting mode")
|
|
elif self.configMode == "set_minute":
|
|
self.configMode = "normal"
|
|
self.setLcdText(f"Saved: {self.alarmTime.hour:02d}:{self.alarmTime.minute:02d}")
|
|
print(f"Alarm time saved: {self.alarmTime.hour:02d}:{self.alarmTime.minute:02d}")
|
|
|
|
lastLcdText: str = ""
|
|
def setLcdText(self, text: str):
|
|
if text == self.lastLcdText:
|
|
return
|
|
self.lastLcdText = text
|
|
self.lcd.clear()
|
|
rows = text.split("\n")
|
|
for i, row in enumerate(rows):
|
|
if i == 0:
|
|
self.lcd.setCursor(0, 0)
|
|
self.lcd.write(f"{(" "+row):<17}")
|
|
elif i == 1:
|
|
self.lcd.setCursor(1, 0)
|
|
self.lcd.write(row)
|
|
#print(f"LCD: {text}")
|
|
|
|
|
|
|
|
|
|
|
|
pins = {
|
|
"button": 6,
|
|
"led": 5,
|
|
"buzzer": 12,
|
|
}
|
|
alarm_clock = AlarmClock(pins)
|
|
|
|
# Add wakeup actions (offsetSeconds = seconds before alarm to trigger)
|
|
alarm_clock.addWakeUpAction(120, lambda: print("2 minutes to alarm!")) # 2 minutes before alarm
|
|
alarm_clock.addWakeUpAction(90, lambda: setShellyPlugState(shelly_deviceId, True)) # activate plug 90s before alarm
|
|
alarm_clock.addWakeUpAction(30, lambda: print("30 seconds to alarm!")) # 30 seconds before alarm
|
|
alarm_clock.addWakeUpAction(0, lambda: print("Alarm triggered!")) # At alarm time
|
|
alarm_clock.addWakeUpAction(0, lambda: playAudio("./boxing_bell_multiple.wav")) # Play sound at alarm time
|
|
|
|
alarm_clock.start()
|
|
|
|
|
|
|
|
|
|
|