Compare commits

Author SHA1 Message Date
gurix 000b204215 refactor: keep style-specific config off the base class via **kwargs
The base AlarmStyle contract carried music_file, forcing SimpleStyle to
accept a parameter it never uses. Move style-specific config to the
subclass: the base __init__ takes only set_led (**kwargs swallows the rest);
each style declares the keyword args it actually uses (BlinkStyle: music_file;
SimpleStyle: none). The runner passes config as keyword args and each style
keeps only what it needs, so the base contract never grows as styles are
added — a future 'talk' style adds talk_file to its own signature, not to the
base. README plugin contract updated.
2026-08-02 22:09:46 +02:00
gurix 3a7b6ac7a3 refactor: styles own their ringing tone directly (Pattern C)
Drop the injected AlarmSound Protocol and the wecker.Sound class. Each style
now owns its tone by using pygame directly; the runner owns only the audio
engine lifecycle (mixer init/quit), button sampling, and cleanup. This is the
seam for future styles that handle their own tone — a 'talk' style would just
import pygame and play speech, with no shared interface to extend.

- styles/base.py: AlarmStyle.__init__(set_led, music_file) + start() + update()
- styles/blink.py: start() loads+plays music via pygame.mixer.music
- styles/simple.py: owns its beep — synthesises a square-wave buffer in
  module (stdlib array+math) and plays it via pygame.mixer.Sound
- wecker.py: setup() brings up the mixer only; run_alarm constructs the style
  and guards start() with a clean log+exit on failure

Tests: a shared tests/conftest.py stubs RPi.GPIO/pygame in sys.modules before
any SUT import (order-independent, removes duplicated inline mocking); the
wecker mock_pygame fixture patches one fresh pygame mock into wecker + both
style modules so assertions see the same calls.

README: tone-ownership and plugin-contract updated.
2026-08-02 21:50:00 +02:00
gurix 92eba762ce feat: styles own their ringing tone (simple beeps, blink plays music)
Move audio ownership out of the shared runner and into each style via an
injected Sound capability (play_music(path) / play_beep()). The runner now
only owns button sampling and cleanup; the style kicks off its own tone in
start() and drives the LED.

- simple: a repeating square-wave beep (ordinary alarm-clock tone), no music
  file required. The beep is synthesised in memory as signed-16-bit stereo
  PCM (stdlib array+math) and played via pygame.mixer.Sound — no shipped
  audio asset, no new dependency.
- blink: unchanged behaviour — music on an endless loop from --music-file /
  MUSIC_FILE / the default track, via pygame.mixer.music.
- AlarmStyle contract gains sound + music_file in __init__ and a start()
  lifecycle hook; AlarmSound Protocol documents the audio seam for future
  styles that handle their own tone.
- setup() no longer loads music (that is the style's job now).

README updated: per-style ringing tone, the --music-file note (blink only),
and the extended plugin contract.
2026-08-02 16:29:40 +02:00
gurix 88a57dab6e refactor: drop unused AlarmStyle re-export from styles package
Nothing imports AlarmStyle via the styles namespace (blink/simple both
import from styles.base directly), so the redundant-alias re-export was
dead weight that only existed to silence ruff F401.
2026-08-02 10:01:46 +02:00
gurix 670119f93e docs: document alarm styles and per-alarm style selection
Add an Alarm Styles section (simple vs blink), show --style in manual usage,
and update the GraphQL examples (setAlarm/getAlarms/startRinging) to include
the style field/argument. Note backward compat for legacy cron entries.
2026-07-31 16:20:50 +02:00
gurix c50f296232 feat: pass alarm style through crontab manager and GraphQL API
- crontab_manager.get_alarms now parses --style from the cron command
  (defaults to 'blink' for legacy entries without --style)
- GraphQL Alarm type gains a 'style' field
- setAlarm accepts a 'style' arg (default 'simple') and validates it
  against the styles registry; unknown styles raise a GraphQL error
- _default_command embeds --style <name> into the cron command
- startRinging accepts 'style' (default 'simple') and passes --style to
  the spawned wecker.py

The crontab remains the single source of truth; the style simply rides
in the cron command alongside --music-file.
2026-07-31 16:20:16 +02:00
gurix b3f3db6f13 feat: add pluggable alarm styles and migrate wecker.py
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.
2026-07-31 16:19:14 +02:00
gurix 7ee28d72b6 docs: plan for pluggable alarm styles 2026-07-31 13:41:17 +02:00
17 changed files with 1012 additions and 281 deletions
+27 -5
View File
@@ -33,10 +33,13 @@ Please refer to the [arcade-button-wiring.md](arcade-button-wiring.md) file for
# Use the default music file or the MUSIC_FILE environment variable
python3 wecker.py
# Or pass a custom music file directly
# Or pass a custom music file (used by the blink style; ignored by simple)
python3 wecker.py --music-file /path/to/your/alarm.mp3
# Choose an alarm style (default: blink; see Alarm Styles below)
python3 wecker.py --style simple
```
The music file is resolved in this order: `--music-file` argument, `MUSIC_FILE` environment variable, default `Laid Back - Sunshine Reggae.mp3`.
`--style` selects the alarm behaviour and defaults to `blink` (so existing cron entries keep working unchanged). Each style owns its own ringing tone: `blink` plays a music file on loop (resolved from `--music-file`, then the `MUSIC_FILE` env var, then the default `Laid Back - Sunshine Reggae.mp3`); `simple` ignores the music file and beeps.
## Automating and Managing Alarms (GraphQL API)
@@ -83,6 +86,7 @@ query {
id
cronExpression
isEnabled
style
}
}
```
@@ -92,14 +96,17 @@ query {
mutation {
setAlarm(
cronExpression: "45 6 * * 1-5",
isEnabled: true
isEnabled: true,
style: "simple"
) {
id
cronExpression
isEnabled
style
}
}
```
`style` is optional and defaults to `"simple"` (see [Alarm Styles](#alarm-styles)).
*(Note: `command` is generated automatically by the API and is **not** accepted as an input parameter on `setAlarm`, to prevent command injection into the system crontab.)*
**Delete an alarm:**
@@ -112,10 +119,10 @@ mutation {
**Start the alarm immediately:**
```graphql
mutation {
startRinging
startRinging(style: "blink")
}
```
Returns `true` if the alarm started, `false` if it was already ringing (ignored).
Returns `true` if the alarm started, `false` if it was already ringing (ignored). `style` is optional and defaults to `"simple"`.
**Stop the alarm immediately:**
```graphql
@@ -134,6 +141,21 @@ Returns `true` if the alarm was stopped, `false` if it wasn't ringing.
}
```
## Alarm Styles
Each alarm has a **style** that decides how you turn it off. Set it per alarm via the `style` field of `setAlarm` (or the `style` argument of `startRinging`). Available styles:
| Style | Ringing tone | How to stop |
|---|---|---|
| `simple` | A repeating, ordinary-alarm-clock **beep** (no music file needed). | Press the button once. The LED stays solid on so the button is findable in the dark. **Default for new alarms.** |
| `blink` | A music file on an endless loop (from `--music-file` / `MUSIC_FILE` / the default track). | The memory-and-attention puzzle described in [How the Puzzle Works](#how-the-puzzle-works). |
**The style owns its ringing tone.** The runner brings up the audio engine (the pygame mixer) and hands each style the button + LED; the style then produces its own tone directly — `blink` plays a music file, `simple` synthesises a beep, a future `talk` style would play speech. Styles use pygame directly (it's the audio engine, not hardware); the runner still owns mixer init and cleanup. This direct ownership is the seam for future styles that handle their own tone.
Adding a style is a plugin-style drop-in: add a class under `styles/` implementing the `AlarmStyle` contract (`__init__(set_led, **kwargs)` + `start()` + `update(now, is_pressed) -> bool`) and register one line in `styles/__init__.py`. Only `set_led` is universal; declare any style-specific config as your own keyword args (e.g. `BlinkStyle` takes `music_file`, `SimpleStyle` takes none) — the runner passes config as keyword args and each style keeps only what it uses.
**Backward compatibility:** existing cron entries created before this feature have no `--style` flag and keep running the `blink` puzzle, so an upgrade never silently changes an alarm. To switch an existing alarm to `simple`, re-save it with `setAlarm(id: ..., cronExpression: ..., style: "simple")`.
## How the Puzzle Works
1. **Ringing:** The music plays in an endless loop.
2. **Start:** Press the arcade button once to start the puzzle. Wait 3 seconds.
+22
View File
@@ -1,6 +1,9 @@
import shlex
import uuid
from crontab import CronTab, CronSlices
from styles import LEGACY_STYLE
class CrontabManager:
COMMENT_PREFIX = "wecker-alarm:"
@@ -28,6 +31,7 @@ class CrontabManager:
"cron_expression": str(job.slices),
"command": job.command,
"is_enabled": job.is_enabled(),
"style": _parse_style(job.command),
}
)
return alarms
@@ -64,3 +68,21 @@ class CrontabManager:
comment = f"{self.COMMENT_PREFIX}{alarm_id}"
cron.remove_all(comment=comment)
cron.write()
def _parse_style(command: str) -> str:
"""Extract the --style value from a cron command.
Returns LEGACY_STYLE ('blink') for entries that predate --style, so an
upgrade never silently changes an existing alarm's behaviour.
"""
try:
tokens = shlex.split(command)
except ValueError:
return LEGACY_STYLE
for i, tok in enumerate(tokens):
if tok == "--style" and i + 1 < len(tokens):
return tokens[i + 1]
if tok.startswith("--style="):
return tok.split("=", 1)[1]
return LEGACY_STYLE
+15 -7
View File
@@ -1,6 +1,7 @@
import strawberry
from typing import List, Optional
import os
import shlex
import sys
import time
import subprocess
@@ -8,6 +9,7 @@ import signal
from pathlib import Path
from api.crontab_manager import CrontabManager
from common import PID_FILE
from styles import get_style
def get_manager():
@@ -42,7 +44,7 @@ def _project_root() -> Path:
return start.parent.parent
def _default_command() -> str:
def _default_command(style: str) -> str:
"""Return the default shell command to run wecker.py from crontab."""
project_root = _project_root()
python_exec = sys.executable
@@ -50,18 +52,18 @@ def _default_command() -> str:
return (
f"cd {project_root} && "
f"XDG_RUNTIME_DIR={runtime_dir} SDL_AUDIODRIVER=pulse "
f"{python_exec} wecker.py >> wecker.log 2>&1"
f"{python_exec} wecker.py --style {shlex.quote(style)} >> wecker.log 2>&1"
)
def _start_wecker_process() -> subprocess.Popen:
def _start_wecker_process(style: str) -> subprocess.Popen:
"""Start wecker.py without invoking a shell."""
project_root = _project_root()
env = os.environ.copy()
env.setdefault("SDL_AUDIODRIVER", "pulse")
env.setdefault("XDG_RUNTIME_DIR", f"/run/user/{os.getuid()}")
return subprocess.Popen(
[sys.executable, str(project_root / "wecker.py")],
[sys.executable, str(project_root / "wecker.py"), "--style", style],
cwd=project_root,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
@@ -75,6 +77,7 @@ class Alarm:
id: str
cron_expression: str
is_enabled: bool
style: str
def _alarm_from_dict(data: dict) -> Alarm:
@@ -83,6 +86,7 @@ def _alarm_from_dict(data: dict) -> Alarm:
id=data["id"],
cron_expression=data["cron_expression"],
is_enabled=data["is_enabled"],
style=data["style"],
)
@@ -118,8 +122,10 @@ class Mutation:
cron_expression: str,
is_enabled: bool = True,
id: Optional[str] = None,
style: str = "simple",
) -> Alarm:
command = _default_command()
get_style(style) # validate; raises ValueError on unknown style
command = _default_command(style)
manager = get_manager()
new_id = manager.set_alarm(
@@ -132,6 +138,7 @@ class Mutation:
id=new_id,
cron_expression=cron_expression,
is_enabled=is_enabled,
style=style,
)
@strawberry.field
@@ -146,12 +153,13 @@ class Mutation:
return False
@strawberry.field
def start_ringing(self) -> bool:
def start_ringing(self, style: str = "simple") -> bool:
"""Start the wecker alarm if it is not already ringing.
Returns True if started, False if already ringing."""
if is_wecker_ringing():
return False
_start_wecker_process()
get_style(style) # validate; raises ValueError on unknown style
_start_wecker_process(style)
return True
@strawberry.field
+165
View File
@@ -0,0 +1,165 @@
# Plan: Pluggable Alarm Styles
Branch: `feature/alarm-styles`
Status: **plan only — no implementation yet.** Review and approve before coding.
## Goal
1. Support **multiple alarm styles**, selectable **per alarm**.
2. Make styles **pluggable** (a "plugin-like" system) so a new style is a drop-in.
3. Ship **two** styles:
- `simple` — press the button once → alarm stops. *(the "default" style; not implemented yet)*
- `blink` — the current count-the-blinks puzzle. *(already implemented, lives in `AlarmClock`)*
4. Style travels with each alarm (cron is still the single source of truth).
## Current state (what changes, what stays)
| Concern | Today | After |
|---|---|---|
| Scheduler | OS `cron`; crontab entries are alarms | **unchanged** |
| Source of truth | crontab (`wecker-alarm:<uuid>` comments) | **unchanged** |
| Per-alarm runnable | command `... wecker.py [--music-file X]` | command `... wecker.py --style <name> [--music-file X]` |
| Ring process | `wecker.py` runs `AlarmClock` state machine | `wecker.py` looks up style in a registry and runs it |
| Styles | one, hardcoded in `wecker.py` | a `styles/` package; `AlarmClock``BlinkStyle` |
| GraphQL `Alarm` type | `id`, `cronExpression`, `isEnabled` | + `style: str` |
| `setAlarm` | `cronExpression`, `isEnabled`, `id?` | + `style: str = "simple"` |
| `startRinging` | spawns `wecker.py` | spawns `wecker.py --style <name>` |
| PID file handoff | `wecker.pid` | **unchanged** |
No database, no in-process scheduler, no separate alarm store. The crontab stays the store; this feature only adds *which style each entry runs*.
## Design decisions (and the lazy choice for each)
### 1. Where the style lives per alarm → **in the cron command**
`--style <name>` rides in the command, exactly like `--music-file` already does. The command *is* "how to run this alarm", so the style belongs there.
- **Rejected:** storing style in the comment (`wecker-alarm:<uuid>:blink`). Works, but couples identity (id) with behaviour (style) and changes the comment format. The command is the right place.
- **Parsing back:** `get_alarms` already reads `job.command`; it scans the tokens for `--style <name>`. Since `_default_command()` is the only writer, parsing is reliable.
### 2. The plugin system → **`styles/` package + explicit registry** (stdlib only)
```
styles/
__init__.py # STYLES = {"blink": BlinkStyle, "simple": SimpleStyle}; get_style(name)
base.py # AlarmStyle: __init__(set_led), update(now, is_pressed) -> bool
blink.py # BlinkStyle (refactored out of wecker.py AlarmClock)
simple.py # SimpleStyle (new)
```
Adding a style = write a module, add one line to `styles/__init__.py`. This is the classic minimal Python plugin pattern.
- **Rejected:** `importlib.metadata` entry points (true external plugins). Overkill for a Pi alarm clock with two styles; the comment format isn't even a packaging distribution. **Upgrade path if needed:** auto-discover modules in `styles/` via `pkgutil.iter_modules` so a new file needs no registry edit. Not now.
### 3. The style interface → **one callable in, one bool out**
```python
class AlarmStyle:
def __init__(self, set_led: Callable[[bool], None]): ...
def update(self, now: float, is_pressed: bool) -> bool: ... # False = stop
```
That is the whole contract. `run_alarm()` owns everything shared:
- starts the music loop *before* the loop,
- samples the button each tick,
- calls `style.update(now, is_pressed)`,
- on `False` (or `test_mode`) stops the music and cleans up.
Styles only decide **when to stop** and **what the LED does**. They never import hardware directly — `set_led` is injected, so:
- no circular import (`styles/` doesn't touch `RPi.GPIO`/`pygame`),
- the API process can `import styles` purely to validate style names, with no hardware loaded.
- **Rejected:** a full `Hardware`/`Outputs` abstraction layer. One passed callable is enough; if a future style needs volume/pwm, promote `set_led` to a small object then. YAGNI.
- Music stop moves out of the style (today `AlarmClock` calls `pygame.mixer.music.stop()` itself) into `run_alarm`'s cleanup. Behaviour-equivalent; the `finally` already does `pygame.mixer.quit()`.
### 4. Defaults & backward compatibility
Two distinct "default" situations — kept separate on purpose:
| Situation | Result | Why |
|---|---|---|
| **Legacy** cron entry with **no** `--style` (existing alarms after upgrade) | `blink` | Preserve current behaviour; an upgrade must not silently change every alarm. `argparse` default for `--style` = `"blink"`. |
| **New API call** `setAlarm(...)` with `style` omitted | `simple` | The user's stated "default style". Makes the new simple behaviour the obvious choice going forward. |
| `startRinging` with `style` omitted | `simple` | Consistent with `setAlarm`. |
> **Decision point (please confirm):** I recommend `simple` as the API default for *new* alarms while keeping `blink` as the fallback for *legacy* cron commands. Alternative: default new alarms to `blink` too, so "no style" always means blink everywhere. Say the word and I'll flip it.
### 5. Validation
- **`wecker.py`**: unknown `--style` → log error and exit non-zero. Never ring silently on a typo.
- **`setAlarm`**: `style` not in `STYLES` → raise → GraphQL error (fail at the boundary, like invalid cron expressions today).
- **`startRinging`**: same.
## TDD / file-by-file work plan
Per AGENTS.md: tests first, ruff green, no untested commits.
### `styles/base.py` (new)
- `AlarmStyle` ABC: `__init__(self, set_led)`, abstract `update(now, is_pressed) -> bool`.
- `__main__` self-check: instantiate a fake style, assert `update` contract.
### `styles/blink.py` (new) — refactor of `wecker.AlarmClock`
- Move the six states (`STATE_RINGING``STATE_WAIT_BEFORE_RETRY`), `BLINK_INTERVAL`, and the non-blocking blink logic verbatim into `BlinkStyle(set_led).update`.
- `run_alarm` handles music; `BlinkStyle` no longer calls `pygame.mixer.music.stop()` (returns `False` instead).
### `styles/simple.py` (new)
```python
class SimpleStyle(AlarmStyle):
def update(self, now, is_pressed):
self.set_led(True) # LED solid on so the button is findable in the dark
return not is_pressed # first press → False → stop
```
> **Decision point:** LED behaviour for `simple` — solid-on while ringing (proposed), or mirror the button press? Minor; propose solid-on.
### `styles/__init__.py` (new)
```python
from styles.base import AlarmStyle
from styles.blink import BlinkStyle
from styles.simple import SimpleStyle
STYLES = {"blink": BlinkStyle, "simple": SimpleStyle}
def get_style(name: str):
try:
return STYLES[name]
except KeyError:
raise ValueError(f"Unknown alarm style: {name!r}. Known: {sorted(STYLES)}")
```
### `wecker.py` (modified)
- argparse: add `--style` (default `"blink"` for legacy compat).
- `setup()` unchanged (GPIO + mixer + load music).
- `run_alarm(style_name)`: `style = get_style(style_name)(set_led)`; loop calls `style.update`; on exit stop music.
- Remove `AlarmClock` and the `STATE_*` constants (moved to `styles/blink.py`).
- `__main__`: `run_alarm(style_name=args.style, music_file=args.music_file)`.
### `api/crontab_manager.py` (modified)
- `get_alarms()` dict gains a `"style"` key, parsed from `job.command` (`--style <name>` → value; absent → `"blink"`).
### `api/schema.py` (modified)
- `Alarm` type: + `style: str`; `_alarm_from_dict` maps it.
- `_default_command(style)`: append `--style {shlex.quote(style)}` to the command.
- `setAlarm(...)`: + `style: str = "simple"`; validate via `styles.get_style` (or check `STYLES`); pass to manager.
- `startRinging(style: str = "simple")`: validate; pass `--style {style}` to the `Popen` argv.
- `_start_wecker_process(style)`: add `--style` to the arg list.
### Tests
- `tests/test_styles_blink.py`: port `test_wecker.py`'s state-machine tests to `BlinkStyle(mock_set_led)`.
- `tests/test_styles_simple.py`: press → `update` returns `False`; no press → `True`.
- `tests/test_styles_registry.py`: `get_style("blink"/"simple")` resolve; unknown raises `ValueError`.
- `tests/test_wecker.py`: rewrite for new entry point — `--style` arg, unknown style errors, `run_alarm` dispatches to style. Drop the moved state-machine tests.
- `tests/test_crontab.py`: `set_alarm` writes `--style` into command; `get_alarms` returns parsed `style`; legacy command (no `--style`) → `"blink"`.
- `tests/test_api.py`: `setAlarm` accepts + returns `style`; rejects unknown style; `getAlarms` includes `style`; `startRinging` passes `--style` to `Popen`.
- Every new module gets a `demo()`/`__main__` self-check per ponytail (blink logic, simple logic, registry).
### `README.md` (modified, per AGENTS.md)
- New **"Alarm Styles"** section: describe `simple` vs `blink`.
- Update GraphQL examples: `setAlarm(..., style: "simple")`, `getAlarms { ... style }`, `startRinging(style: "blink")`.
- Note backward compat: existing alarms keep blinking until re-saved.
## Migration / rollout
1. Deploy code. Existing cron entries (no `--style`) still run `blink` — no behaviour change.
2. Existing alarms keep blinking. To switch one to simple: `setAlarm(id=..., cronExpression=..., style: "simple")` (re-writes the command).
3. New alarms default to `simple`.
## Out of scope (deliberately)
- External/plugin packages (entry points) — add if a third style comes from outside this repo.
- Per-style configuration knobs (e.g. blink count range) — hardcode sensible defaults now; add a config arg when a style actually needs tuning.
- A `Hardware` abstraction layer — one injected `set_led` callable is enough.
- Auto-discovery of style modules — explicit registry; switch to `pkgutil.iter_modules` only if the count grows.
## Suggested commit sequence
1. `feat: add styles package with AlarmStyle base, registry, and blink+simple` (with tests)
2. `refactor: move AlarmClock state machine into styles.blink`
3. `feat: pass --style through wecker.py, crontab, and GraphQL API` (with tests)
4. `docs: document alarm styles in README`
+22
View File
@@ -0,0 +1,22 @@
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)}"
)
+42
View File
@@ -0,0 +1,42 @@
from abc import ABC, abstractmethod
from typing import Callable
class AlarmStyle(ABC):
"""A pluggable alarm behaviour.
A style owns *its own ringing tone*: it uses the audio engine (pygame)
directly and decides when the alarm stops and what the LED does. It never
touches hardware directly: ``set_led`` is injected by the runner, and the
runner owns the audio engine lifecycle (mixer init/quit). Styles stay
import-safe in the API process — ``import pygame`` does not initialise
audio, so the API can import styles just to validate names.
Lifecycle: construct once with ``set_led`` (plus any style-specific
keyword config the runner passes); call ``start()`` to begin ringing (kick
off the tone and initial LED); then call ``update`` every tick. Return
``False`` from ``update`` to stop the alarm.
Only ``set_led`` is universal. Style-specific config (e.g. ``music_file``)
belongs on the subclass, not here, so the base contract never grows as
styles are added; the runner passes config as keyword arguments and each
style declares only the ones it uses (extra kwargs are swallowed).
"""
def __init__(self, set_led: Callable[[bool], None], **kwargs):
self.set_led = set_led
def start(self) -> None:
"""Begin ringing. Override to start the style's tone / initial 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.
"""
+160
View File
@@ -0,0 +1,160 @@
import logging
import random
import pygame
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, music_file=None, **kwargs):
super().__init__(set_led, **kwargs)
self.music_file = music_file
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 start(self):
# The puzzle rings with music on an endless loop; the LED stays off
# until the button is pressed to start the blink sequence.
pygame.mixer.music.load(self.music_file)
pygame.mixer.music.set_volume(1.0)
pygame.mixer.music.play(-1)
self.set_led(False)
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.
# (Does not call start(), which needs the audio engine.)
style = BlinkStyle(lambda on: None, "x.mp3")
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")
+82
View File
@@ -0,0 +1,82 @@
import array
import math
import pygame
from styles.base import AlarmStyle
# Time between beep retriggers; a short beep followed by this gap gives the
# classic alarm-clock "beep ... beep ... beep".
BEEP_INTERVAL = 0.5
def _make_beep_buffer(
freq: int = 1000, duration: float = 0.15, rate: int = 44100
) -> bytes:
"""Synthesize a short square-wave beep as signed-16-bit stereo PCM.
Matches the mixer format used by wecker.setup() (rate, -16, 2 channels) so
it feeds straight into ``pygame.mixer.Sound``. No audio file, no extra dep.
"""
n = int(rate * duration)
buf = array.array("h")
for i in range(n):
v = 32767 if math.sin(2 * math.pi * freq * (i / rate)) >= 0 else -32767
buf.append(v)
buf.append(v) # duplicate to stereo
return buf.tobytes()
_BEEP_BUFFER = _make_beep_buffer()
class SimpleStyle(AlarmStyle):
"""Press the button once to stop. Rings with a repeating beep.
This style plays no music file; it owns its tone via a synthesised square
wave played through ``pygame.mixer.Sound``.
"""
def __init__(self, set_led, **kwargs):
super().__init__(set_led, **kwargs)
self._next_beep = 0.0
self._beep = None
def start(self):
# LED solid on so the button is findable in the dark.
self.set_led(True)
self._beep = pygame.mixer.Sound(_BEEP_BUFFER)
def update(self, now, is_pressed):
self.set_led(True)
if now >= self._next_beep:
self._beep.play()
self._next_beep = now + BEEP_INTERVAL
return not is_pressed # first press -> False -> stop
if __name__ == "__main__":
# Self-check: the timer beeps on the interval and a press stops it.
# Bypasses start() (which needs the audio engine) by stubbing the beep.
class _Beep:
def __init__(self):
self.plays = 0
def play(self):
self.plays += 1
style = SimpleStyle(lambda on: None)
style._beep = _Beep()
assert style.update(0.0, False) is True
assert style._beep.plays == 1 # first tick beeps
assert style.update(0.4, False) is True # before interval -> no beep
assert style._beep.plays == 1
assert style.update(0.5, False) is True # at interval -> beep
assert style._beep.plays == 2
assert style.update(0.6, True) is False # press -> stop
print("simple self-check ok")
+14
View File
@@ -0,0 +1,14 @@
"""Shared test setup: stub hardware/audio modules before any SUT import.
wecker and the alarm styles import RPi.GPIO (hardware) and pygame (audio
engine). The tests run off the Pi and without a real audio device, so these
are replaced with MagicMocks in sys.modules before wecker/styles are imported.
Loaded by pytest before any test module in this directory is collected.
"""
import sys
from unittest.mock import MagicMock
sys.modules["RPi"] = MagicMock()
sys.modules["RPi.GPIO"] = MagicMock()
sys.modules["pygame"] = MagicMock()
+105
View File
@@ -2,6 +2,7 @@ from unittest.mock import patch, MagicMock
from fastapi.testclient import TestClient
import tempfile
import os
import pytest
import signal
# We need to set the environment variable before importing the app
@@ -403,3 +404,107 @@ def test_pid_file_same_shared_constant_in_api():
assert api_schema.PID_FILE is common.PID_FILE, (
"api.schema.PID_FILE must reference common.PID_FILE, not redefine it"
)
def _cleanup_alarm(alarm_id):
headers = {"X-API-Key": "test-secret"}
client.post(
"/graphql",
json={"query": f'mutation {{ deleteAlarm(id: "{alarm_id}") }}'},
headers=headers,
)
def test_set_alarm_accepts_and_returns_style():
headers = {"X-API-Key": "test-secret"}
mutation = '''
mutation {
setAlarm(cronExpression: "0 9 * * *", style: "blink") {
id
style
}
}
'''
res = client.post("/graphql", json={"query": mutation}, headers=headers)
assert res.status_code == 200
alarm = res.json()["data"]["setAlarm"]
assert alarm["style"] == "blink"
_cleanup_alarm(alarm["id"])
def test_set_alarm_default_style_is_simple():
headers = {"X-API-Key": "test-secret"}
mutation = '''
mutation {
setAlarm(cronExpression: "0 9 * * *") { id style }
}
'''
res = client.post("/graphql", json={"query": mutation}, headers=headers)
assert res.status_code == 200
alarm = res.json()["data"]["setAlarm"]
assert alarm["style"] == "simple"
_cleanup_alarm(alarm["id"])
def test_set_alarm_rejects_unknown_style():
headers = {"X-API-Key": "test-secret"}
mutation = '''
mutation {
setAlarm(cronExpression: "0 9 * * *", style: "nope") { id }
}
'''
res = client.post("/graphql", json={"query": mutation}, headers=headers)
assert res.status_code == 200
assert "errors" in res.json()
assert "Unknown alarm style" in str(res.json()["errors"])
def test_get_alarms_includes_style():
headers = {"X-API-Key": "test-secret"}
set_mut = '''
mutation { setAlarm(cronExpression: "0 9 * * *", style: "blink") { id } }
'''
res = client.post("/graphql", json={"query": set_mut}, headers=headers)
alarm_id = res.json()["data"]["setAlarm"]["id"]
res = client.post(
"/graphql", json={"query": "{ getAlarms { id style } }"}, headers=headers
)
alarms = res.json()["data"]["getAlarms"]
assert any(a["id"] == alarm_id and a["style"] == "blink" for a in alarms)
_cleanup_alarm(alarm_id)
def test_start_ringing_passes_style():
from api.schema import Mutation
mutation = Mutation()
with patch("api.schema.is_wecker_ringing", return_value=False), \
patch("subprocess.Popen") as mock_popen:
mock_popen.return_value.pid = 9999
assert mutation.start_ringing(style="blink") is True
argv = mock_popen.call_args[0][0]
assert "--style" in argv and "blink" in argv
def test_start_ringing_default_style_is_simple():
from api.schema import Mutation
mutation = Mutation()
with patch("api.schema.is_wecker_ringing", return_value=False), \
patch("subprocess.Popen") as mock_popen:
mock_popen.return_value.pid = 9999
mutation.start_ringing()
argv = mock_popen.call_args[0][0]
assert "simple" in argv
def test_start_ringing_rejects_unknown_style():
from api.schema import Mutation
mutation = Mutation()
with patch("api.schema.is_wecker_ringing", return_value=False), \
patch("subprocess.Popen") as mock_popen:
with pytest.raises(ValueError, match="Unknown alarm style"):
mutation.start_ringing(style="nope")
mock_popen.assert_not_called()
+35
View File
@@ -72,3 +72,38 @@ def test_set_alarm_rejects_invalid_cron_expression(crontab_file):
manager = CrontabManager(tabfile=crontab_file)
with pytest.raises(ValueError, match="Invalid cron expression"):
manager.set_alarm("test-id", "not-a-cron-expression", "cmd", True)
def test_get_alarms_parses_style(crontab_file):
manager = CrontabManager(tabfile=crontab_file)
manager.set_alarm(
alarm_id="style-id",
cron_expression="30 7 * * *",
command="python wecker.py --style simple",
is_enabled=True,
)
alarms = manager.get_alarms()
assert len(alarms) == 1
assert alarms[0]["style"] == "simple"
def test_get_alarms_parses_eq_style(crontab_file):
manager = CrontabManager(tabfile=crontab_file)
manager.set_alarm(
alarm_id="style-eq-id",
cron_expression="30 7 * * *",
command="python wecker.py --style=blink",
is_enabled=True,
)
assert manager.get_alarms()[0]["style"] == "blink"
def test_get_alarms_legacy_command_defaults_to_blink(crontab_file):
manager = CrontabManager(tabfile=crontab_file)
manager.set_alarm(
alarm_id="legacy-id",
cron_expression="30 7 * * *",
command="python wecker.py",
is_enabled=True,
)
assert manager.get_alarms()[0]["style"] == "blink"
+4 -9
View File
@@ -1,14 +1,9 @@
import pytest
import sys
from unittest.mock import MagicMock, patch
from unittest.mock import patch
# Mock hardware modules before importing wecker (tests run off the Pi).
sys.modules["RPi"] = MagicMock()
sys.modules["RPi.GPIO"] = MagicMock()
sys.modules["pygame"] = MagicMock()
import common # noqa: E402
import wecker # noqa: E402
# Hardware/audio mocks live in tests/conftest.py (shared across all test files).
import common
import wecker
@pytest.fixture
+105
View File
@@ -0,0 +1,105 @@
from unittest.mock import MagicMock, patch
from styles.blink import (
STATE_BLINKING,
STATE_EVALUATING,
STATE_WAIT_BEFORE_BLINK,
STATE_WAIT_BEFORE_RETRY,
STATE_WAIT_FOR_INPUT,
BlinkStyle,
)
def _make_blink(music_file="music.mp3"):
"""Return (style, set_led calls) for a BlinkStyle ready to update."""
calls = []
style = BlinkStyle(lambda on: calls.append(on), music_file)
return style, calls
def test_start_plays_music_and_led_off():
style, calls = _make_blink("track.mp3")
with patch("styles.blink.pygame", MagicMock()) as pg:
style.start()
pg.mixer.music.load.assert_called_with("track.mp3")
pg.mixer.music.set_volume.assert_called_with(1.0)
pg.mixer.music.play.assert_called_with(-1)
assert calls == [False] # LED off while music plays until a press starts the puzzle
def test_press_starts_puzzle():
style, _ = _make_blink()
assert style.update(100.0, True) is True
assert style.state == STATE_WAIT_BEFORE_BLINK
def test_blink_correct_sequence_stops(monkeypatch):
style, _ = _make_blink()
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():
style, _ = _make_blink()
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():
style, calls = _make_blink()
style.set_led = calls.append
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():
style, _ = _make_blink()
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
+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")
+65
View File
@@ -0,0 +1,65 @@
from unittest.mock import MagicMock, patch
from styles.simple import SimpleStyle
def _make_style():
calls = []
return SimpleStyle(lambda on: calls.append(on)), calls
def test_simple_ignores_unrelated_config():
# SimpleStyle must swallow style-specific kwargs it doesn't use (e.g.
# music_file belongs to blink), so the runner can pass config uniformly.
style, _ = _make_style()
style = SimpleStyle(lambda on: None, music_file="ignored.mp3")
with patch("styles.simple.pygame", MagicMock()):
style.start()
assert style.update(0.0, False) is True
def test_simple_keeps_ringing_when_not_pressed():
style, _ = _make_style()
with patch("styles.simple.pygame", MagicMock()):
style.start()
assert style.update(0.0, False) is True
def test_simple_stops_on_first_press():
style, _ = _make_style()
with patch("styles.simple.pygame", MagicMock()):
style.start()
assert style.update(0.0, True) is False # press -> stop
def test_simple_led_solid_on():
style, calls = _make_style()
with patch("styles.simple.pygame", MagicMock()):
style.start()
style.update(0.0, False)
assert calls == [True, True] # start sets LED on; update keeps it on
def test_simple_beeps_on_interval():
style, _ = _make_style()
with patch("styles.simple.pygame", MagicMock()) as pg:
style.start()
style.update(100.0, False) # first tick -> beep
assert pg.mixer.Sound.return_value.play.called
pg.mixer.Sound.return_value.play.reset_mock()
style.update(100.4, False) # before interval -> no beep
assert not pg.mixer.Sound.return_value.play.called
style.update(100.5, False) # at interval -> beep
assert pg.mixer.Sound.return_value.play.called
def test_simple_never_plays_music():
style, _ = _make_style()
with patch("styles.simple.pygame", MagicMock()) as pg:
style.start()
style.update(0.0, False)
style.update(0.0, True) # press -> stop
assert not pg.mixer.music.play.called
assert not pg.mixer.music.load.called
+85 -114
View File
@@ -1,13 +1,9 @@
import pytest
import sys
from unittest.mock import MagicMock, patch
# Mock RPi.GPIO and pygame before importing wecker
sys.modules["RPi"] = MagicMock()
sys.modules["RPi.GPIO"] = MagicMock()
sys.modules["pygame"] = MagicMock()
import wecker # noqa: E402
import styles.blink
import styles.simple
import wecker
@pytest.fixture
@@ -20,7 +16,18 @@ def mock_gpio():
@pytest.fixture
def mock_pygame():
with patch.object(wecker, "pygame") as mock:
"""One fresh pygame mock shared by wecker and both styles.
Under Pattern C the styles use pygame directly, so a single mock must be
patched into wecker (setup) and the style modules (start) for assertions
to see the same calls.
"""
mock = MagicMock()
with (
patch.object(wecker, "pygame", mock),
patch.object(styles.blink, "pygame", mock),
patch.object(styles.simple, "pygame", mock),
):
yield mock
@@ -32,114 +39,78 @@ def test_set_led(mock_gpio):
mock_gpio.output.assert_called_with(wecker.LED_PIN, mock_gpio.HIGH)
@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
mock_pygame.mixer.get_init.return_value = True
def test_setup_inits_mixer_and_gpio_not_music(mock_gpio, mock_pygame):
wecker.setup()
assert mock_pygame.mixer.pre_init.called
assert mock_pygame.mixer.init.called
assert mock_gpio.setmode.called
assert mock_gpio.setup.called
# setup() no longer loads music — that is now the style's job.
assert not mock_pygame.mixer.music.load.called
# Just run it in test mode, it should execute the loop once and exit
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_run_alarm_uses_selected_style(mock_gpio, mock_pygame):
"""run_alarm looks up the style, injects set_led + music_file, calls start()."""
fake = MagicMock()
fake.update.return_value = True # keep ringing
fake_style_cls = MagicMock(return_value=fake)
with patch("wecker.get_style", return_value=fake_style_cls) as get_style:
wecker.run_alarm(style_name="whatever", test_mode=True)
get_style.assert_called_once_with("whatever")
assert fake_style_cls.call_args.args[0] is wecker.set_led
assert isinstance(fake_style_cls.call_args.kwargs["music_file"], str)
fake.start.assert_called_once()
fake.update.assert_called()
def test_run_alarm_default_music_file(mock_gpio, mock_pygame, monkeypatch):
monkeypatch.delenv("MUSIC_FILE", raising=False)
fake = MagicMock()
fake.update.return_value = True
fake_style_cls = MagicMock(return_value=fake)
with patch("wecker.get_style", return_value=fake_style_cls):
wecker.run_alarm(test_mode=True)
assert fake_style_cls.call_args.kwargs["music_file"] == wecker.DEFAULT_MUSIC_FILE
def test_run_alarm_env_music_file(mock_gpio, mock_pygame, monkeypatch):
monkeypatch.setenv("MUSIC_FILE", "/env/track.mp3")
fake = MagicMock()
fake.update.return_value = True
fake_style_cls = MagicMock(return_value=fake)
with patch("wecker.get_style", return_value=fake_style_cls):
wecker.run_alarm(test_mode=True)
assert fake_style_cls.call_args.kwargs["music_file"] == "/env/track.mp3"
def test_run_alarm_music_file_arg_overrides_env(mock_gpio, mock_pygame, monkeypatch):
monkeypatch.setenv("MUSIC_FILE", "/env/track.mp3")
fake = MagicMock()
fake.update.return_value = True
fake_style_cls = MagicMock(return_value=fake)
with patch("wecker.get_style", return_value=fake_style_cls):
wecker.run_alarm(music_file="/arg/track.mp3", test_mode=True)
assert fake_style_cls.call_args.kwargs["music_file"] == "/arg/track.mp3"
def test_run_alarm_blink_plays_music(mock_gpio, mock_pygame):
"""The default (blink) style plays music on an endless loop."""
wecker.run_alarm(test_mode=True)
assert mock_pygame.mixer.music.play.called
assert mock_pygame.mixer.music.play.call_args.args[0] == -1 # endless loop
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_state_machine_incorrect(mock_gpio, mock_pygame):
clock = wecker.AlarmClock()
clock.state = wecker.STATE_WAIT_FOR_INPUT
clock.target_blinks = 3
# Only press once
clock.update(100.0, True)
clock.update(100.1, False)
# 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
def test_setup_uses_env_music_file(mock_gpio, mock_pygame, monkeypatch):
monkeypatch.setenv("MUSIC_FILE", "/custom/track.mp3")
wecker.setup()
mock_pygame.mixer.music.load.assert_called_with("/custom/track.mp3")
def test_setup_music_file_argument_overrides_env(mock_gpio, mock_pygame, monkeypatch):
monkeypatch.setenv("MUSIC_FILE", "/env/track.mp3")
wecker.setup(music_file="/arg/track.mp3")
mock_pygame.mixer.music.load.assert_called_with("/arg/track.mp3")
def test_run_alarm_simple_beeps_and_plays_no_music(mock_gpio, mock_pygame):
"""The simple style beeps and never touches the music stream."""
wecker.run_alarm(style_name="simple", test_mode=True)
assert mock_pygame.mixer.Sound.return_value.play.called # beep played
assert not mock_pygame.mixer.music.play.called # no music
+44 -146
View File
@@ -1,17 +1,18 @@
import RPi.GPIO as GPIO
import time
import logging
import pygame
import random
import sys
import os
import sys
import time
import fcntl
import logging
import pygame
import RPi.GPIO as GPIO
# Use PulseAudio/PipeWire audio driver to avoid ALSA device-busy errors.
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(
@@ -69,11 +70,12 @@ LED_PIN = 27
DEFAULT_MUSIC_FILE = "Laid Back - Sunshine Reggae.mp3"
def setup(music_file: str | None = None):
"""Initialize GPIO and audio hardware. Call once before running."""
if music_file is None:
music_file = os.environ.get("MUSIC_FILE", DEFAULT_MUSIC_FILE)
def setup():
"""Initialize GPIO and the audio engine. Call once before running.
Each style owns its own ringing tone via pygame directly; this only brings
up the mixer and the GPIO pins.
"""
GPIO.setmode(GPIO.BCM)
GPIO.setup(LED_PIN, GPIO.OUT)
GPIO.setup(BUTTON_PIN, GPIO.IN, pull_up_down=GPIO.PUD_UP)
@@ -82,25 +84,6 @@ def setup(music_file: str | None = None):
pygame.mixer.pre_init(frequency=44100, size=-16, channels=2, buffer=4096)
pygame.mixer.init()
try:
pygame.mixer.music.load(music_file)
pygame.mixer.music.set_volume(1.0)
except Exception as e:
logging.error(f"Error loading audio file: {e}")
# Don't exit in test mode
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 +94,31 @@ 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
The runner owns only the audio engine lifecycle (mixer init/quit), button
sampling, and cleanup; each style owns its own ringing tone (via pygame
directly) and LED behaviour.
"""
style_cls = get_style(style_name) # validate before touching hardware
if music_file is None:
music_file = os.environ.get("MUSIC_FILE", DEFAULT_MUSIC_FILE)
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):
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()
setup()
style = style_cls(set_led, music_file=music_file)
logging.info("Alarm clock started.")
try:
style.start() # style kicks off its own tone + initial LED
except Exception:
logging.exception("Error starting alarm")
if "pytest" not in sys.modules:
sys.exit(1)
raise
try:
while True:
@@ -237,7 +129,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
@@ -255,13 +147,19 @@ if __name__ == "__main__":
parser.add_argument(
"--music-file",
default=os.environ.get("MUSIC_FILE", DEFAULT_MUSIC_FILE),
help="Path to the MP3 file to play (default: $MUSIC_FILE or default track).",
help="Music file for the blink style (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()