refactor: make blink_led non-blocking in the state machine
This commit is contained in:
+37
-30
@@ -32,15 +32,6 @@ def test_set_led(mock_gpio):
|
||||
mock_gpio.output.assert_called_with(wecker.LED_PIN, mock_gpio.HIGH)
|
||||
|
||||
|
||||
@patch("wecker.time.sleep")
|
||||
def test_blink_led(mock_sleep, mock_gpio):
|
||||
wecker.blink_led(2)
|
||||
# 2 blinks = 4 sleep calls, 2 set_led(True), 2 set_led(False)
|
||||
assert mock_sleep.call_count == 4
|
||||
# GPIO output called 4 times total (on, off, on, off)
|
||||
assert mock_gpio.output.call_count == 4
|
||||
|
||||
|
||||
@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
|
||||
@@ -59,33 +50,30 @@ def test_state_machine_evaluation(mock_gpio, mock_pygame):
|
||||
assert clock.state == wecker.STATE_WAIT_BEFORE_BLINK
|
||||
|
||||
# Wait 3 seconds -> blinking
|
||||
with patch("wecker.blink_led"):
|
||||
with patch("wecker.random.randint", return_value=3):
|
||||
clock.update(103.1, False)
|
||||
assert clock.state == wecker.STATE_BLINKING
|
||||
assert clock.target_blinks == 3
|
||||
|
||||
with patch("wecker.blink_led"):
|
||||
clock.update(103.2, False)
|
||||
# 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
|
||||
assert clock.target_blinks >= 1
|
||||
|
||||
# Set the user presses to be correct
|
||||
clock.target_blinks = 3
|
||||
|
||||
# Press 1
|
||||
clock.update(104.0, True)
|
||||
clock.update(104.1, False)
|
||||
clock.update(105.5, True)
|
||||
clock.update(105.6, False)
|
||||
# Press 2
|
||||
clock.update(104.5, True)
|
||||
clock.update(104.6, False)
|
||||
clock.update(106.0, True)
|
||||
clock.update(106.1, False)
|
||||
# Press 3
|
||||
clock.update(105.0, True)
|
||||
clock.update(105.1, False)
|
||||
clock.update(106.5, True)
|
||||
clock.update(106.6, False)
|
||||
|
||||
assert clock.user_presses == 3
|
||||
|
||||
# Wait 3 seconds to evaluate
|
||||
clock.update(108.2, False) # Triggers state change
|
||||
keep_running = clock.update(108.3, False) # Triggers evaluation
|
||||
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
|
||||
@@ -111,19 +99,38 @@ def test_state_machine_incorrect(mock_gpio, mock_pygame):
|
||||
assert clock.state == wecker.STATE_WAIT_BEFORE_RETRY
|
||||
|
||||
|
||||
@patch("wecker.time.time")
|
||||
def test_blinking_updates_time_correctly(mock_time, mock_gpio, mock_pygame):
|
||||
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
|
||||
|
||||
mock_time.return_value = 200.0
|
||||
|
||||
with patch("wecker.blink_led"):
|
||||
# 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 == 200.0
|
||||
assert clock.last_interaction_time == 102.4
|
||||
|
||||
|
||||
def test_setup_uses_env_music_file(mock_gpio, mock_pygame, monkeypatch):
|
||||
|
||||
@@ -96,11 +96,7 @@ STATE_WAIT_FOR_INPUT = 3
|
||||
STATE_EVALUATING = 4
|
||||
STATE_WAIT_BEFORE_RETRY = 5
|
||||
|
||||
state = STATE_RINGING
|
||||
target_blinks = 0
|
||||
user_presses = 0
|
||||
last_interaction_time = 0
|
||||
button_was_pressed = False
|
||||
BLINK_INTERVAL = 0.3
|
||||
|
||||
|
||||
def set_led(on):
|
||||
@@ -112,15 +108,6 @@ def set_led(on):
|
||||
GPIO.output(LED_PIN, GPIO.HIGH)
|
||||
|
||||
|
||||
def blink_led(times):
|
||||
"""Blinks the LED a specific number of times (blocking)"""
|
||||
for _ in range(times):
|
||||
set_led(True)
|
||||
time.sleep(0.3) # LED on for 300ms
|
||||
set_led(False)
|
||||
time.sleep(0.3) # LED off for 300ms
|
||||
|
||||
|
||||
class AlarmClock:
|
||||
def __init__(self):
|
||||
self.state = STATE_RINGING
|
||||
@@ -128,6 +115,9 @@ class AlarmClock:
|
||||
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
|
||||
@@ -150,19 +140,38 @@ class AlarmClock:
|
||||
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:
|
||||
# Move the blink_led out of the update loop for testability,
|
||||
# or just call it directly. Here we call it.
|
||||
blink_led(self.target_blinks)
|
||||
# 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 = (
|
||||
time.time()
|
||||
) # Use time.time() to account for blocking blink_led
|
||||
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)
|
||||
|
||||
Reference in New Issue
Block a user