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 # Use the default music file or the MUSIC_FILE environment variable
python3 wecker.py 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 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) ## Automating and Managing Alarms (GraphQL API)
@@ -83,6 +86,7 @@ query {
id id
cronExpression cronExpression
isEnabled isEnabled
style
} }
} }
``` ```
@@ -92,14 +96,17 @@ query {
mutation { mutation {
setAlarm( setAlarm(
cronExpression: "45 6 * * 1-5", cronExpression: "45 6 * * 1-5",
isEnabled: true isEnabled: true,
style: "simple"
) { ) {
id id
cronExpression cronExpression
isEnabled 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.)* *(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:** **Delete an alarm:**
@@ -112,10 +119,10 @@ mutation {
**Start the alarm immediately:** **Start the alarm immediately:**
```graphql ```graphql
mutation { 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:** **Stop the alarm immediately:**
```graphql ```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 ## How the Puzzle Works
1. **Ringing:** The music plays in an endless loop. 1. **Ringing:** The music plays in an endless loop.
2. **Start:** Press the arcade button once to start the puzzle. Wait 3 seconds. 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 import uuid
from crontab import CronTab, CronSlices from crontab import CronTab, CronSlices
from styles import LEGACY_STYLE
class CrontabManager: class CrontabManager:
COMMENT_PREFIX = "wecker-alarm:" COMMENT_PREFIX = "wecker-alarm:"
@@ -28,6 +31,7 @@ class CrontabManager:
"cron_expression": str(job.slices), "cron_expression": str(job.slices),
"command": job.command, "command": job.command,
"is_enabled": job.is_enabled(), "is_enabled": job.is_enabled(),
"style": _parse_style(job.command),
} }
) )
return alarms return alarms
@@ -64,3 +68,21 @@ class CrontabManager:
comment = f"{self.COMMENT_PREFIX}{alarm_id}" comment = f"{self.COMMENT_PREFIX}{alarm_id}"
cron.remove_all(comment=comment) cron.remove_all(comment=comment)
cron.write() 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 import strawberry
from typing import List, Optional from typing import List, Optional
import os import os
import shlex
import sys import sys
import time import time
import subprocess import subprocess
@@ -8,6 +9,7 @@ import signal
from pathlib import Path from pathlib import Path
from api.crontab_manager import CrontabManager from api.crontab_manager import CrontabManager
from common import PID_FILE from common import PID_FILE
from styles import get_style
def get_manager(): def get_manager():
@@ -42,7 +44,7 @@ def _project_root() -> Path:
return start.parent.parent 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.""" """Return the default shell command to run wecker.py from crontab."""
project_root = _project_root() project_root = _project_root()
python_exec = sys.executable python_exec = sys.executable
@@ -50,18 +52,18 @@ def _default_command() -> str:
return ( return (
f"cd {project_root} && " f"cd {project_root} && "
f"XDG_RUNTIME_DIR={runtime_dir} SDL_AUDIODRIVER=pulse " 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.""" """Start wecker.py without invoking a shell."""
project_root = _project_root() project_root = _project_root()
env = os.environ.copy() env = os.environ.copy()
env.setdefault("SDL_AUDIODRIVER", "pulse") env.setdefault("SDL_AUDIODRIVER", "pulse")
env.setdefault("XDG_RUNTIME_DIR", f"/run/user/{os.getuid()}") env.setdefault("XDG_RUNTIME_DIR", f"/run/user/{os.getuid()}")
return subprocess.Popen( return subprocess.Popen(
[sys.executable, str(project_root / "wecker.py")], [sys.executable, str(project_root / "wecker.py"), "--style", style],
cwd=project_root, cwd=project_root,
stdout=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
@@ -75,6 +77,7 @@ class Alarm:
id: str id: str
cron_expression: str cron_expression: str
is_enabled: bool is_enabled: bool
style: str
def _alarm_from_dict(data: dict) -> Alarm: def _alarm_from_dict(data: dict) -> Alarm:
@@ -83,6 +86,7 @@ def _alarm_from_dict(data: dict) -> Alarm:
id=data["id"], id=data["id"],
cron_expression=data["cron_expression"], cron_expression=data["cron_expression"],
is_enabled=data["is_enabled"], is_enabled=data["is_enabled"],
style=data["style"],
) )
@@ -118,8 +122,10 @@ class Mutation:
cron_expression: str, cron_expression: str,
is_enabled: bool = True, is_enabled: bool = True,
id: Optional[str] = None, id: Optional[str] = None,
style: str = "simple",
) -> Alarm: ) -> Alarm:
command = _default_command() get_style(style) # validate; raises ValueError on unknown style
command = _default_command(style)
manager = get_manager() manager = get_manager()
new_id = manager.set_alarm( new_id = manager.set_alarm(
@@ -132,6 +138,7 @@ class Mutation:
id=new_id, id=new_id,
cron_expression=cron_expression, cron_expression=cron_expression,
is_enabled=is_enabled, is_enabled=is_enabled,
style=style,
) )
@strawberry.field @strawberry.field
@@ -146,12 +153,13 @@ class Mutation:
return False return False
@strawberry.field @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. """Start the wecker alarm if it is not already ringing.
Returns True if started, False if already ringing.""" Returns True if started, False if already ringing."""
if is_wecker_ringing(): if is_wecker_ringing():
return False return False
_start_wecker_process() get_style(style) # validate; raises ValueError on unknown style
_start_wecker_process(style)
return True return True
@strawberry.field @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 from fastapi.testclient import TestClient
import tempfile import tempfile
import os import os
import pytest
import signal import signal
# We need to set the environment variable before importing the app # 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, ( assert api_schema.PID_FILE is common.PID_FILE, (
"api.schema.PID_FILE must reference common.PID_FILE, not redefine it" "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) manager = CrontabManager(tabfile=crontab_file)
with pytest.raises(ValueError, match="Invalid cron expression"): with pytest.raises(ValueError, match="Invalid cron expression"):
manager.set_alarm("test-id", "not-a-cron-expression", "cmd", True) 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 pytest
import sys from unittest.mock import patch
from unittest.mock import MagicMock, patch
# Mock hardware modules before importing wecker (tests run off the Pi). # Hardware/audio mocks live in tests/conftest.py (shared across all test files).
sys.modules["RPi"] = MagicMock() import common
sys.modules["RPi.GPIO"] = MagicMock() import wecker
sys.modules["pygame"] = MagicMock()
import common # noqa: E402
import wecker # noqa: E402
@pytest.fixture @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 pytest
import sys
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
# Mock RPi.GPIO and pygame before importing wecker import styles.blink
sys.modules["RPi"] = MagicMock() import styles.simple
sys.modules["RPi.GPIO"] = MagicMock() import wecker
sys.modules["pygame"] = MagicMock()
import wecker # noqa: E402
@pytest.fixture @pytest.fixture
@@ -20,7 +16,18 @@ def mock_gpio():
@pytest.fixture @pytest.fixture
def mock_pygame(): 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 yield mock
@@ -32,114 +39,78 @@ def test_set_led(mock_gpio):
mock_gpio.output.assert_called_with(wecker.LED_PIN, mock_gpio.HIGH) mock_gpio.output.assert_called_with(wecker.LED_PIN, mock_gpio.HIGH)
@patch("wecker.time.time") def test_setup_inits_mixer_and_gpio_not_music(mock_gpio, mock_pygame):
def test_run_alarm_start_to_wait(mock_time, mock_gpio, mock_pygame): wecker.setup()
# Mock pygame.mixer.get_init() to return True so music plays assert mock_pygame.mixer.pre_init.called
mock_pygame.mixer.get_init.return_value = True 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) wecker.run_alarm(test_mode=True)
assert mock_pygame.mixer.music.play.called 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): def test_run_alarm_simple_beeps_and_plays_no_music(mock_gpio, mock_pygame):
clock = wecker.AlarmClock() """The simple style beeps and never touches the music stream."""
wecker.run_alarm(style_name="simple", test_mode=True)
# Transition to ringing -> wait before blink assert mock_pygame.mixer.Sound.return_value.play.called # beep played
clock.update(100.0, True) # button press assert not mock_pygame.mixer.music.play.called # no music
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")
+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 os
import sys
import time
import fcntl import fcntl
import logging
import pygame
import RPi.GPIO as GPIO
# Use PulseAudio/PipeWire audio driver to avoid ALSA device-busy errors. # Use PulseAudio/PipeWire audio driver to avoid ALSA device-busy errors.
os.environ.setdefault("SDL_AUDIODRIVER", "pulse") os.environ.setdefault("SDL_AUDIODRIVER", "pulse")
from logging.handlers import RotatingFileHandler from logging.handlers import RotatingFileHandler
from common import PID_FILE 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. # Configure logging with rotation to avoid unbounded growth on the SD card.
logging.basicConfig( logging.basicConfig(
@@ -69,11 +70,12 @@ LED_PIN = 27
DEFAULT_MUSIC_FILE = "Laid Back - Sunshine Reggae.mp3" DEFAULT_MUSIC_FILE = "Laid Back - Sunshine Reggae.mp3"
def setup(music_file: str | None = None): def setup():
"""Initialize GPIO and audio hardware. Call once before running.""" """Initialize GPIO and the audio engine. Call once before running.
if music_file is None:
music_file = os.environ.get("MUSIC_FILE", DEFAULT_MUSIC_FILE)
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.setmode(GPIO.BCM)
GPIO.setup(LED_PIN, GPIO.OUT) GPIO.setup(LED_PIN, GPIO.OUT)
GPIO.setup(BUTTON_PIN, GPIO.IN, pull_up_down=GPIO.PUD_UP) 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.pre_init(frequency=44100, size=-16, channels=2, buffer=4096)
pygame.mixer.init() 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): def set_led(on):
"""Turns the LED on or off (LOW = ON, HIGH = OFF)""" """Turns the LED on or off (LOW = ON, HIGH = OFF)"""
@@ -111,122 +94,31 @@ def set_led(on):
GPIO.output(LED_PIN, GPIO.HIGH) GPIO.output(LED_PIN, GPIO.HIGH)
class AlarmClock: def run_alarm(
def __init__(self): style_name: str = LEGACY_STYLE,
self.state = STATE_RINGING test_mode: bool = False,
self.target_blinks = 0 music_file: str | None = None,
self.user_presses = 0 ):
self.last_interaction_time = 0 """Run the alarm clock with the given style until the style signals stop.
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): The runner owns only the audio engine lifecycle (mixer init/quit), button
button_just_pressed = False sampling, and cleanup; each style owns its own ringing tone (via pygame
if is_pressed and not self.button_was_pressed: directly) and LED behaviour.
self.button_was_pressed = True """
button_just_pressed = True style_cls = get_style(style_name) # validate before touching hardware
elif not is_pressed and self.button_was_pressed: if music_file is None:
self.button_was_pressed = False music_file = os.environ.get("MUSIC_FILE", DEFAULT_MUSIC_FILE)
if self.state == STATE_RINGING: setup()
if button_just_pressed: style = style_cls(set_led, music_file=music_file)
logging.info( logging.info("Alarm clock started.")
"Alarm button pressed! Puzzle started. Waiting 3 seconds..." try:
) style.start() # style kicks off its own tone + initial LED
self.state = STATE_WAIT_BEFORE_BLINK except Exception:
self.last_interaction_time = now logging.exception("Error starting alarm")
if "pytest" not in sys.modules:
elif self.state == STATE_WAIT_BEFORE_BLINK: sys.exit(1)
if now - self.last_interaction_time >= 3.0: raise
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()
try: try:
while True: while True:
@@ -237,7 +129,7 @@ def run_alarm(test_mode=False, music_file: str | None = None):
else: else:
is_pressed = False is_pressed = False
keep_running = clock.update(now, is_pressed) keep_running = style.update(now, is_pressed)
if not keep_running or test_mode: if not keep_running or test_mode:
break break
@@ -255,13 +147,19 @@ if __name__ == "__main__":
parser.add_argument( parser.add_argument(
"--music-file", "--music-file",
default=os.environ.get("MUSIC_FILE", DEFAULT_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() args = parser.parse_args()
ensure_single_instance() ensure_single_instance()
try: try:
run_alarm(music_file=args.music_file) run_alarm(style_name=args.style, music_file=args.music_file)
finally: finally:
pygame.mixer.quit() pygame.mixer.quit()
GPIO.cleanup() GPIO.cleanup()