Compare commits
8
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
000b204215 | ||
|
|
3a7b6ac7a3 | ||
|
|
92eba762ce | ||
|
|
88a57dab6e | ||
|
|
670119f93e | ||
|
|
c50f296232 | ||
|
|
b3f3db6f13 | ||
|
|
7ee28d72b6 |
@@ -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.
|
||||||
|
|||||||
@@ -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
@@ -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
|
||||||
|
|||||||
@@ -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`
|
||||||
@@ -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)}"
|
||||||
|
)
|
||||||
@@ -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
@@ -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")
|
||||||
@@ -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")
|
||||||
@@ -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()
|
||||||
@@ -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()
|
||||||
|
|||||||
@@ -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"
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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
|
||||||
@@ -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")
|
||||||
@@ -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
@@ -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")
|
|
||||||
|
|||||||
@@ -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
|
|
||||||
self.button_was_pressed = False
|
|
||||||
self._blink_phase: int = 0
|
|
||||||
self._blink_phases: int = 0
|
|
||||||
self._blink_next_toggle: float | None = None
|
|
||||||
|
|
||||||
def update(self, now, is_pressed):
|
|
||||||
button_just_pressed = False
|
|
||||||
if is_pressed and not self.button_was_pressed:
|
|
||||||
self.button_was_pressed = True
|
|
||||||
button_just_pressed = True
|
|
||||||
elif not is_pressed and self.button_was_pressed:
|
|
||||||
self.button_was_pressed = False
|
|
||||||
|
|
||||||
if self.state == STATE_RINGING:
|
|
||||||
if button_just_pressed:
|
|
||||||
logging.info(
|
|
||||||
"Alarm button pressed! Puzzle started. Waiting 3 seconds..."
|
|
||||||
)
|
|
||||||
self.state = STATE_WAIT_BEFORE_BLINK
|
|
||||||
self.last_interaction_time = now
|
|
||||||
|
|
||||||
elif self.state == STATE_WAIT_BEFORE_BLINK:
|
|
||||||
if now - self.last_interaction_time >= 3.0:
|
|
||||||
self.target_blinks = random.randint(1, 7)
|
|
||||||
logging.info(f"Blinking {self.target_blinks} times...")
|
|
||||||
self.state = STATE_BLINKING
|
|
||||||
self._blink_phase = 0
|
|
||||||
self._blink_phases = self.target_blinks * 2
|
|
||||||
self._blink_next_toggle = now + BLINK_INTERVAL
|
|
||||||
set_led(True)
|
|
||||||
|
|
||||||
elif self.state == STATE_BLINKING:
|
|
||||||
# Non-blocking blink: toggle the LED at fixed intervals so the main
|
|
||||||
# loop keeps sampling the button and can be interrupted.
|
|
||||||
if self._blink_next_toggle is None:
|
|
||||||
# Safety for direct state assignment (e.g. tests).
|
|
||||||
self._blink_phase = 0
|
|
||||||
self._blink_phases = self.target_blinks * 2
|
|
||||||
self._blink_next_toggle = now + BLINK_INTERVAL
|
|
||||||
set_led(True)
|
|
||||||
|
|
||||||
while now >= self._blink_next_toggle:
|
|
||||||
self._blink_phase += 1
|
|
||||||
self._blink_next_toggle += BLINK_INTERVAL
|
|
||||||
if self._blink_phase >= self._blink_phases:
|
|
||||||
set_led(False)
|
|
||||||
self._blink_phase = 0
|
|
||||||
self._blink_phases = 0
|
|
||||||
self._blink_next_toggle = None
|
|
||||||
logging.info("Blinking finished. Waiting for input...")
|
|
||||||
self.state = STATE_WAIT_FOR_INPUT
|
|
||||||
self.user_presses = 0
|
|
||||||
self.last_interaction_time = now
|
|
||||||
if "GPIO" in globals() and hasattr(GPIO, "input"):
|
|
||||||
self.button_was_pressed = GPIO.input(BUTTON_PIN) == GPIO.LOW
|
|
||||||
break
|
|
||||||
else:
|
|
||||||
set_led(self._blink_phase % 2 == 0)
|
|
||||||
|
|
||||||
elif self.state == STATE_WAIT_FOR_INPUT:
|
|
||||||
set_led(is_pressed)
|
|
||||||
|
|
||||||
if button_just_pressed:
|
|
||||||
self.user_presses += 1
|
|
||||||
logging.info(f"Button pressed: {self.user_presses} times")
|
|
||||||
|
|
||||||
if is_pressed:
|
|
||||||
self.last_interaction_time = now
|
|
||||||
|
|
||||||
if not is_pressed and (now - self.last_interaction_time >= 3.0):
|
|
||||||
self.state = STATE_EVALUATING
|
|
||||||
|
|
||||||
elif self.state == STATE_EVALUATING:
|
|
||||||
logging.info(
|
|
||||||
f"Evaluation: Target={self.target_blinks}, Entered={self.user_presses}"
|
|
||||||
)
|
|
||||||
if self.user_presses == self.target_blinks:
|
|
||||||
logging.info("Puzzle solved correctly! Alarm clock is stopping.")
|
|
||||||
if (
|
|
||||||
"pygame" in globals()
|
|
||||||
and hasattr(pygame, "mixer")
|
|
||||||
and pygame.mixer.get_init()
|
|
||||||
):
|
):
|
||||||
pygame.mixer.music.stop()
|
"""Run the alarm clock with the given style until the style signals 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:
|
The runner owns only the audio engine lifecycle (mixer init/quit), button
|
||||||
if now - self.last_interaction_time >= 5.0:
|
sampling, and cleanup; each style owns its own ringing tone (via pygame
|
||||||
self.target_blinks = random.randint(1, 7)
|
directly) and LED behaviour.
|
||||||
logging.info(f"New attempt! Blinking {self.target_blinks} times...")
|
"""
|
||||||
self.state = STATE_BLINKING
|
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)
|
||||||
|
|
||||||
return True # Keep running
|
setup()
|
||||||
|
style = style_cls(set_led, music_file=music_file)
|
||||||
|
logging.info("Alarm clock started.")
|
||||||
def run_alarm(test_mode=False, music_file: str | None = None):
|
try:
|
||||||
setup(music_file=music_file)
|
style.start() # style kicks off its own tone + initial LED
|
||||||
logging.info("Alarm clock started. Music is playing in an endless loop.")
|
except Exception:
|
||||||
if "pygame" in globals() and hasattr(pygame, "mixer") and pygame.mixer.get_init():
|
logging.exception("Error starting alarm")
|
||||||
pygame.mixer.music.play(-1)
|
if "pytest" not in sys.modules:
|
||||||
|
sys.exit(1)
|
||||||
set_led(False) # LED off at start
|
raise
|
||||||
|
|
||||||
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()
|
||||||
|
|||||||
Reference in New Issue
Block a user