gurix/wecker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add pluggable alarm styles and migrate wecker.py

Browse Source 
Introduce a styles/ plugin package:
- styles/base.py: AlarmStyle ABC (injected set_led, update()->bool contract)
- styles/blink.py: BlinkStyle, the existing count-the-blinks puzzle moved
  out of wecker.py's AlarmClock
- styles/simple.py: SimpleStyle, press-once-to-stop (the new default style)
- styles/__init__.py: STYLES registry + get_style() validator + LEGACY_STYLE

wecker.py now resolves the style via the registry and owns only the shared
music/button/cleanup loop; the --style CLI arg defaults to blink so existing
cron entries keep their behaviour. Unknown styles raise before hardware init.
This commit is contained in:
gurix
2026-07-31 16:19:14 +02:00
parent 7ee28d72b6
commit b3f3db6f13
9 changed files with 395 additions and 212 deletions
Download Patch File Download Diff File Expand all files Collapse all files
styles/__init__.py
+23
View File
@@ -0,0 +1,23 @@
from styles.base import AlarmStyle as AlarmStyle
from styles.blink import BlinkStyle
from styles.simple import SimpleStyle
# Registry of available alarm styles. Add a line here when you add a style.
STYLES = {
"blink": BlinkStyle,
"simple": SimpleStyle,
}
# Fallback style for legacy cron entries that predate --style, so an upgrade
# never silently changes an existing alarm's behaviour.
LEGACY_STYLE = "blink"
def get_style(name: str):
"""Return the style class for ``name`` or raise ValueError."""
try:
return STYLES[name]
except KeyError:
raise ValueError(
f"Unknown alarm style: {name!r}. Known: {sorted(STYLES)}"
)
styles/base.py
+29
View File
@@ -0,0 +1,29 @@
from abc import ABC, abstractmethod
from typing import Callable
class AlarmStyle(ABC):
"""A pluggable alarm behaviour.
A style decides *when* the alarm stops and what the LED does while it
rings. It never touches hardware directly: ``set_led`` is injected by the
runner, so styles stay import-safe in the API process (no GPIO/pygame).
Lifecycle: construct once with ``set_led``, then call ``update`` every
tick. Return ``False`` from ``update`` to stop the alarm.
"""
def __init__(self, set_led: Callable[[bool], None]):
self.set_led = set_led
@abstractmethod
def update(self, now: float, is_pressed: bool) -> bool:
"""Advance the style by one tick.
Args:
now: current time (``time.time()`` from the runner).
is_pressed: whether the button is currently held down.
Returns:
True to keep ringing, False to stop.
"""
styles/blink.py
+148
View File
@@ -0,0 +1,148 @@
import logging
import random
from styles.base import AlarmStyle
# State Machine states for the puzzle flow.
STATE_RINGING = 0
STATE_WAIT_BEFORE_BLINK = 1
STATE_BLINKING = 2
STATE_WAIT_FOR_INPUT = 3
STATE_EVALUATING = 4
STATE_WAIT_BEFORE_RETRY = 5
BLINK_INTERVAL = 0.3
class BlinkStyle(AlarmStyle):
"""The count-the-blinks puzzle.
Press once to start, watch the LED blink 1-7 times, then press the button
exactly that many times. A wrong count starts a fresh sequence.
"""
def __init__(self, set_led):
super().__init__(set_led)
self.state = STATE_RINGING
self.target_blinks = 0
self.user_presses = 0
self.last_interaction_time = 0
self.button_was_pressed = False
self._blink_phase = 0
self._blink_phases = 0
self._blink_next_toggle: float | None = None
def update(self, now, is_pressed):
button_just_pressed = False
if is_pressed and not self.button_was_pressed:
self.button_was_pressed = True
button_just_pressed = True
elif not is_pressed and self.button_was_pressed:
self.button_was_pressed = False
if self.state == STATE_RINGING:
if button_just_pressed:
logging.info(
"Alarm button pressed! Puzzle started. Waiting 3 seconds..."
)
self.state = STATE_WAIT_BEFORE_BLINK
self.last_interaction_time = now
elif self.state == STATE_WAIT_BEFORE_BLINK:
if now - self.last_interaction_time >= 3.0:
self.target_blinks = random.randint(1, 7)
logging.info(f"Blinking {self.target_blinks} times...")
self.state = STATE_BLINKING
self._blink_phase = 0
self._blink_phases = self.target_blinks * 2
self._blink_next_toggle = now + BLINK_INTERVAL
self.set_led(True)
elif self.state == STATE_BLINKING:
# Non-blocking blink: toggle the LED at fixed intervals so the
# runner keeps sampling the button and can be interrupted.
if self._blink_next_toggle is None:
# Safety for direct state assignment (e.g. tests).
self._blink_phase = 0
self._blink_phases = self.target_blinks * 2
self._blink_next_toggle = now + BLINK_INTERVAL
self.set_led(True)
while now >= self._blink_next_toggle:
self._blink_phase += 1
self._blink_next_toggle += BLINK_INTERVAL
if self._blink_phase >= self._blink_phases:
self.set_led(False)
self._blink_phase = 0
self._blink_phases = 0
self._blink_next_toggle = None
logging.info("Blinking finished. Waiting for input...")
self.state = STATE_WAIT_FOR_INPUT
self.user_presses = 0
self.last_interaction_time = now
self.button_was_pressed = is_pressed
break
else:
self.set_led(self._blink_phase % 2 == 0)
elif self.state == STATE_WAIT_FOR_INPUT:
self.set_led(is_pressed)
if button_just_pressed:
self.user_presses += 1
logging.info(f"Button pressed: {self.user_presses} times")
if is_pressed:
self.last_interaction_time = now
if not is_pressed and (now - self.last_interaction_time >= 3.0):
self.state = STATE_EVALUATING
elif self.state == STATE_EVALUATING:
logging.info(
f"Evaluation: Target={self.target_blinks}, "
f"Entered={self.user_presses}"
)
if self.user_presses == self.target_blinks:
logging.info("Puzzle solved correctly! Alarm clock is stopping.")
self.set_led(False)
return False # Signal the runner to stop.
else:
logging.info("Incorrect input! Waiting before retrying...")
self.state = STATE_WAIT_BEFORE_RETRY
self.last_interaction_time = now
self.set_led(False)
elif self.state == STATE_WAIT_BEFORE_RETRY:
if now - self.last_interaction_time >= 5.0:
self.target_blinks = random.randint(1, 7)
logging.info(f"New attempt! Blinking {self.target_blinks} times...")
self.state = STATE_BLINKING
return True # Keep running
if __name__ == "__main__":
# Self-check: a correct blink sequence stops the alarm.
style = BlinkStyle(lambda on: None)
assert style.update(100.0, True) is True
assert style.state == STATE_WAIT_BEFORE_BLINK
import random as _random
_random.randint = lambda a, b: 3 # force 3 blinks
style.update(103.1, False)
assert style.state == STATE_BLINKING and style.target_blinks == 3
style.update(105.0, False)
assert style.state == STATE_WAIT_FOR_INPUT
for t in (105.5, 106.0, 106.5):
style.update(t, True)
style.update(t + 0.05, False)
assert style.user_presses == 3
style.update(109.7, False)
assert style.update(109.8, False) is False # solved -> stop
print("blink self-check ok")
styles/simple.py
+18
View File
@@ -0,0 +1,18 @@
from styles.base import AlarmStyle
class SimpleStyle(AlarmStyle):
"""Press the button once to stop the alarm."""
def update(self, now, is_pressed):
# LED solid on so the button is findable in the dark.
self.set_led(True)
return not is_pressed # first press -> False -> stop
if __name__ == "__main__":
# Self-check: not pressed keeps ringing; a press stops it.
style = SimpleStyle(lambda on: None)
assert style.update(0.0, False) is True
assert style.update(0.1, True) is False
print("simple self-check ok")
tests/test_styles_blink.py
+97
View File
@@ -0,0 +1,97 @@
from styles.blink import (
STATE_BLINKING,
STATE_EVALUATING,
STATE_WAIT_BEFORE_BLINK,
STATE_WAIT_BEFORE_RETRY,
STATE_WAIT_FOR_INPUT,
BlinkStyle,
)
def _led():
"""Return (calls list, set_led callable recording each call)."""
calls = []
return calls, lambda on: calls.append(on)
def test_press_starts_puzzle():
_, set_led = _led()
style = BlinkStyle(set_led)
assert style.update(100.0, True) is True
assert style.state == STATE_WAIT_BEFORE_BLINK
def test_blink_correct_sequence_stops(monkeypatch):
_, set_led = _led()
style = BlinkStyle(set_led)
style.update(100.0, True) # press -> start puzzle
assert style.state == STATE_WAIT_BEFORE_BLINK
monkeypatch.setattr("styles.blink.random.randint", lambda a, b: 3)
style.update(103.1, False) # 3s passed -> blinking, 3 blinks
assert style.state == STATE_BLINKING
assert style.target_blinks == 3
style.update(105.0, False) # run through the blink sequence
assert style.state == STATE_WAIT_FOR_INPUT
# Press exactly 3 times
for t in (105.5, 106.0, 106.5):
style.update(t, True)
style.update(t + 0.05, False)
assert style.user_presses == 3
style.update(109.7, False) # 3s idle -> evaluating
assert style.state == STATE_EVALUATING
assert style.update(109.8, False) is False # correct -> stop
def test_blink_incorrect_retries():
_, set_led = _led()
style = BlinkStyle(set_led)
style.state = STATE_WAIT_FOR_INPUT
style.target_blinks = 3
style.update(100.0, True) # one press
style.update(100.1, False)
style.last_interaction_time = 100.1
style.update(103.2, False) # -> evaluating
keep_running = style.update(103.3, False) # 1 != 3 -> retry
assert keep_running is True
assert style.state == STATE_WAIT_BEFORE_RETRY
def test_blinking_non_blocking():
calls, set_led = _led()
style = BlinkStyle(set_led)
style.state = STATE_BLINKING
style.target_blinks = 2
style.update(100.0, False) # start of blinking: LED on
assert style.state == STATE_BLINKING
assert calls[-1] is True
style.update(100.3, False) # mid-blink toggles by elapsed time
assert calls[-1] is False
# 2 blinks * 2 phases * 0.3s = 1.2s
style.update(101.2, False)
assert style.state == STATE_WAIT_FOR_INPUT
assert calls[-1] is False
def test_blinking_advances_time():
_, set_led = _led()
style = BlinkStyle(set_led)
style.state = STATE_BLINKING
style.target_blinks = 4
# 4 blinks * 2 phases * 0.3s = 2.4s
style.update(100.0, False)
style.update(102.4, False)
assert style.state == STATE_WAIT_FOR_INPUT
assert style.last_interaction_time == 102.4
tests/test_styles_registry.py
+20
View File
@@ -0,0 +1,20 @@
import pytest
from styles import STYLES, get_style
from styles.blink import BlinkStyle
from styles.simple import SimpleStyle
def test_registry_has_blink_and_simple():
assert STYLES["blink"] is BlinkStyle
assert STYLES["simple"] is SimpleStyle
def test_get_style_returns_class():
assert get_style("blink") is BlinkStyle
assert get_style("simple") is SimpleStyle
def test_get_style_unknown_raises():
with pytest.raises(ValueError, match="Unknown alarm style"):
get_style("nope")
tests/test_styles_simple.py
+23
View File
@@ -0,0 +1,23 @@
from styles.simple import SimpleStyle
def test_simple_keeps_ringing_when_not_pressed():
calls = []
style = SimpleStyle(lambda on: calls.append(on))
assert style.update(0.0, False) is True
assert calls[-1] is True # LED solid on while ringing
def test_simple_stops_on_first_press():
calls = []
style = SimpleStyle(lambda on: calls.append(on))
assert style.update(0.0, True) is False # press -> stop
def test_simple_led_off_on_release_after_press():
# After a press the style stops; verify it never turns the LED off itself
# (the runner's cleanup handles that). It only ever asserts LED on.
calls = []
style = SimpleStyle(lambda on: calls.append(on))
style.update(0.0, False)
assert calls == [True]
tests/test_wecker.py
+17 -89
View File
@@ -33,104 +33,32 @@ def test_set_led(mock_gpio):
@patch("wecker.time.time")
def test_run_alarm_start_to_wait(mock_time, mock_gpio, mock_pygame):
# Mock pygame.mixer.get_init() to return True so music plays
def test_run_alarm_plays_music(mock_time, mock_gpio, mock_pygame):
mock_pygame.mixer.get_init.return_value = True
# Just run it in test mode, it should execute the loop once and exit
wecker.run_alarm(test_mode=True)
assert mock_pygame.mixer.music.play.called
def test_state_machine_evaluation(mock_gpio, mock_pygame):
clock = wecker.AlarmClock()
# Transition to ringing -> wait before blink
clock.update(100.0, True) # button press
assert clock.state == wecker.STATE_WAIT_BEFORE_BLINK
# Wait 3 seconds -> blinking
with patch("wecker.random.randint", return_value=3):
clock.update(103.1, False)
assert clock.state == wecker.STATE_BLINKING
assert clock.target_blinks == 3
# Advance time past the blink sequence (3 blinks * 2 phases * 0.3s = 1.8s)
clock.update(105.0, False)
assert clock.state == wecker.STATE_WAIT_FOR_INPUT
# Press 1
clock.update(105.5, True)
clock.update(105.6, False)
# Press 2
clock.update(106.0, True)
clock.update(106.1, False)
# Press 3
clock.update(106.5, True)
clock.update(106.6, False)
assert clock.user_presses == 3
# Wait 3 seconds to evaluate
clock.update(109.7, False) # Triggers state change
keep_running = clock.update(109.8, False) # Triggers evaluation
# It should evaluate, see it's correct, and return False (stop running)
assert clock.state == wecker.STATE_EVALUATING
assert not keep_running
def test_run_alarm_unknown_style_raises(mock_gpio, mock_pygame):
"""An unknown style is rejected before any hardware is touched."""
with pytest.raises(ValueError, match="Unknown alarm style"):
wecker.run_alarm(style_name="nope", test_mode=True)
mock_pygame.mixer.init.assert_not_called()
def test_state_machine_incorrect(mock_gpio, mock_pygame):
clock = wecker.AlarmClock()
clock.state = wecker.STATE_WAIT_FOR_INPUT
clock.target_blinks = 3
def test_run_alarm_uses_selected_style(mock_gpio, mock_pygame):
"""run_alarm looks up the style by name and drives the returned style."""
mock_pygame.mixer.get_init.return_value = True
fake = MagicMock()
fake.update.return_value = True # keep ringing
fake_style_cls = MagicMock(return_value=fake)
# Only press once
clock.update(100.0, True)
clock.update(100.1, False)
with patch("wecker.get_style", return_value=fake_style_cls) as get_style:
wecker.run_alarm(style_name="whatever", test_mode=True)
# Wait to evaluate
clock.last_interaction_time = 100.1
clock.update(103.2, False) # triggers eval state
keep_running = clock.update(103.3, False) # evals to incorrect
# Evaluated incorrectly, should wait before retry
assert keep_running
assert clock.state == wecker.STATE_WAIT_BEFORE_RETRY
def test_blinking_is_non_blocking(mock_gpio, mock_pygame):
clock = wecker.AlarmClock()
clock.state = wecker.STATE_BLINKING
clock.target_blinks = 2
# Start of blinking: LED on
clock.update(100.0, False)
assert clock.state == wecker.STATE_BLINKING
mock_gpio.output.assert_called_with(wecker.LED_PIN, mock_gpio.LOW)
# Mid-blink: LED toggles based on elapsed time
clock.update(100.3, False)
mock_gpio.output.assert_called_with(wecker.LED_PIN, mock_gpio.HIGH)
# After sequence completes (2 blinks * 2 phases * 0.3s = 1.2s)
clock.update(101.2, False)
assert clock.state == wecker.STATE_WAIT_FOR_INPUT
mock_gpio.output.assert_called_with(wecker.LED_PIN, mock_gpio.HIGH)
def test_blinking_updates_time_correctly(mock_gpio, mock_pygame):
clock = wecker.AlarmClock()
clock.state = wecker.STATE_BLINKING
clock.target_blinks = 4
# Start blinking, then advance to the end of the sequence.
# 4 blinks * 2 phases * 0.3s = 2.4s
clock.update(100.0, False)
clock.update(102.4, False)
assert clock.state == wecker.STATE_WAIT_FOR_INPUT
assert clock.last_interaction_time == 102.4
get_style.assert_called_once_with("whatever")
fake_style_cls.assert_called_once_with(wecker.set_led)
fake.update.assert_called()
def test_setup_uses_env_music_file(mock_gpio, mock_pygame, monkeypatch):
wecker.py
+19 -122
View File
@@ -2,7 +2,6 @@ import RPi.GPIO as GPIO
import time
import logging
import pygame
import random
import sys
import os
import fcntl
@@ -12,6 +11,7 @@ os.environ.setdefault("SDL_AUDIODRIVER", "pulse")
from logging.handlers import RotatingFileHandler
from common import PID_FILE
from styles import STYLES, LEGACY_STYLE, get_style
# Configure logging with rotation to avoid unbounded growth on the SD card.
logging.basicConfig(
@@ -91,16 +91,6 @@ def setup(music_file: str | None = None):
if "pytest" not in sys.modules:
sys.exit(1)
# State Machine states for the flow
STATE_RINGING = 0
STATE_WAIT_BEFORE_BLINK = 1
STATE_BLINKING = 2
STATE_WAIT_FOR_INPUT = 3
STATE_EVALUATING = 4
STATE_WAIT_BEFORE_RETRY = 5
BLINK_INTERVAL = 0.3
def set_led(on):
"""Turns the LED on or off (LOW = ON, HIGH = OFF)"""
@@ -111,122 +101,23 @@ def set_led(on):
GPIO.output(LED_PIN, GPIO.HIGH)
class AlarmClock:
def __init__(self):
self.state = STATE_RINGING
self.target_blinks = 0
self.user_presses = 0
self.last_interaction_time = 0
self.button_was_pressed = False
self._blink_phase: int = 0
self._blink_phases: int = 0
self._blink_next_toggle: float | None = None
def update(self, now, is_pressed):
button_just_pressed = False
if is_pressed and not self.button_was_pressed:
self.button_was_pressed = True
button_just_pressed = True
elif not is_pressed and self.button_was_pressed:
self.button_was_pressed = False
if self.state == STATE_RINGING:
if button_just_pressed:
logging.info(
"Alarm button pressed! Puzzle started. Waiting 3 seconds..."
)
self.state = STATE_WAIT_BEFORE_BLINK
self.last_interaction_time = now
elif self.state == STATE_WAIT_BEFORE_BLINK:
if now - self.last_interaction_time >= 3.0:
self.target_blinks = random.randint(1, 7)
logging.info(f"Blinking {self.target_blinks} times...")
self.state = STATE_BLINKING
self._blink_phase = 0
self._blink_phases = self.target_blinks * 2
self._blink_next_toggle = now + BLINK_INTERVAL
set_led(True)
elif self.state == STATE_BLINKING:
# Non-blocking blink: toggle the LED at fixed intervals so the main
# loop keeps sampling the button and can be interrupted.
if self._blink_next_toggle is None:
# Safety for direct state assignment (e.g. tests).
self._blink_phase = 0
self._blink_phases = self.target_blinks * 2
self._blink_next_toggle = now + BLINK_INTERVAL
set_led(True)
while now >= self._blink_next_toggle:
self._blink_phase += 1
self._blink_next_toggle += BLINK_INTERVAL
if self._blink_phase >= self._blink_phases:
set_led(False)
self._blink_phase = 0
self._blink_phases = 0
self._blink_next_toggle = None
logging.info("Blinking finished. Waiting for input...")
self.state = STATE_WAIT_FOR_INPUT
self.user_presses = 0
self.last_interaction_time = now
if "GPIO" in globals() and hasattr(GPIO, "input"):
self.button_was_pressed = GPIO.input(BUTTON_PIN) == GPIO.LOW
break
else:
set_led(self._blink_phase % 2 == 0)
elif self.state == STATE_WAIT_FOR_INPUT:
set_led(is_pressed)
if button_just_pressed:
self.user_presses += 1
logging.info(f"Button pressed: {self.user_presses} times")
if is_pressed:
self.last_interaction_time = now
if not is_pressed and (now - self.last_interaction_time >= 3.0):
self.state = STATE_EVALUATING
elif self.state == STATE_EVALUATING:
logging.info(
f"Evaluation: Target={self.target_blinks}, Entered={self.user_presses}"
)
if self.user_presses == self.target_blinks:
logging.info("Puzzle solved correctly! Alarm clock is stopping.")
if (
"pygame" in globals()
and hasattr(pygame, "mixer")
and pygame.mixer.get_init()
def run_alarm(
style_name: str = LEGACY_STYLE,
test_mode: bool = False,
music_file: str | None = None,
):
pygame.mixer.music.stop()
set_led(False)
return False # Indicate we should stop running
else:
logging.info("Incorrect input! Waiting 3 seconds before retrying...")
self.state = STATE_WAIT_BEFORE_RETRY
self.last_interaction_time = time.time()
set_led(False)
"""Run the alarm clock with the given style until the style signals stop.
elif self.state == STATE_WAIT_BEFORE_RETRY:
if now - self.last_interaction_time >= 5.0:
self.target_blinks = random.randint(1, 7)
logging.info(f"New attempt! Blinking {self.target_blinks} times...")
self.state = STATE_BLINKING
return True # Keep running
def run_alarm(test_mode=False, music_file: str | None = None):
The runner owns everything shared (music loop, button sampling, cleanup);
the style only decides when to stop and what the LED does.
"""
style = get_style(style_name)(set_led)
setup(music_file=music_file)
logging.info("Alarm clock started. Music is playing in an endless loop.")
if "pygame" in globals() and hasattr(pygame, "mixer") and pygame.mixer.get_init():
pygame.mixer.music.play(-1)
set_led(False) # LED off at start
clock = AlarmClock()
set_led(False) # LED off at start; the style drives it from here.
try:
while True:
@@ -237,7 +128,7 @@ def run_alarm(test_mode=False, music_file: str | None = None):
else:
is_pressed = False
keep_running = clock.update(now, is_pressed)
keep_running = style.update(now, is_pressed)
if not keep_running or test_mode:
break
@@ -257,11 +148,17 @@ if __name__ == "__main__":
default=os.environ.get("MUSIC_FILE", DEFAULT_MUSIC_FILE),
help="Path to the MP3 file to play (default: $MUSIC_FILE or default track).",
)
parser.add_argument(
"--style",
default=LEGACY_STYLE,
choices=sorted(STYLES),
help="Alarm style to run (default: %(default)s).",
)
args = parser.parse_args()
ensure_single_instance()
try:
run_alarm(music_file=args.music_file)
run_alarm(style_name=args.style, music_file=args.music_file)
finally:
pygame.mixer.quit()
GPIO.cleanup()