diff --git a/styles/__init__.py b/styles/__init__.py new file mode 100644 index 0000000..bcc90bd --- /dev/null +++ b/styles/__init__.py @@ -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)}" + ) diff --git a/styles/base.py b/styles/base.py new file mode 100644 index 0000000..0d1ce80 --- /dev/null +++ b/styles/base.py @@ -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. + """ diff --git a/styles/blink.py b/styles/blink.py new file mode 100644 index 0000000..7d6d046 --- /dev/null +++ b/styles/blink.py @@ -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") diff --git a/styles/simple.py b/styles/simple.py new file mode 100644 index 0000000..d8bdd47 --- /dev/null +++ b/styles/simple.py @@ -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") diff --git a/tests/test_styles_blink.py b/tests/test_styles_blink.py new file mode 100644 index 0000000..0f304a4 --- /dev/null +++ b/tests/test_styles_blink.py @@ -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 diff --git a/tests/test_styles_registry.py b/tests/test_styles_registry.py new file mode 100644 index 0000000..c8210ef --- /dev/null +++ b/tests/test_styles_registry.py @@ -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") diff --git a/tests/test_styles_simple.py b/tests/test_styles_simple.py new file mode 100644 index 0000000..74f9bc1 --- /dev/null +++ b/tests/test_styles_simple.py @@ -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] diff --git a/tests/test_wecker.py b/tests/test_wecker.py index 53f12b3..851b16d 100644 --- a/tests/test_wecker.py +++ b/tests/test_wecker.py @@ -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): diff --git a/wecker.py b/wecker.py index 782b282..71632c4 100644 --- a/wecker.py +++ b/wecker.py @@ -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 run_alarm( + style_name: str = LEGACY_STYLE, + test_mode: bool = False, + music_file: str | None = None, +): + """Run the alarm clock with the given style until the style signals stop. - 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() - ): - 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) - - 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()