gurix/wecker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: make command parameter optional in set_alarm mutation

Browse Source
This commit is contained in:
Markus Graf
2026-05-11 15:05:29 +02:00
parent 5b5b845e41
commit ff79224560
10 changed files with 152 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Laid Back - Sunshine Reggae.mp3
BIN
View File
Binary file not shown.
README.md
+2 -1
View File
@@ -74,15 +74,16 @@ query {
```
**Set an alarm:**
*(Note: `command` is optional. If omitted, the API will automatically figure out the correct command to start `wecker.py`)*
```graphql
mutation {
setAlarm(
cronExpression: "45 6 * * 1-5",
command: "cd /home/pi/workspace/wecker && /usr/bin/python3 wecker.py > wecker.log 2>&1",
isEnabled: true
) {
id
cronExpression
command
}
}
```
api.pid
+1
View File
@@ -0,0 +1 @@
17958
api/crontab_manager.py
+20 -11
View File
@@ -1,6 +1,7 @@
import uuid
from crontab import CronTab
class CrontabManager:
COMMENT_PREFIX = "wecker-alarm:"
@@ -21,29 +22,37 @@ class CrontabManager:
if job.comment.startswith(self.COMMENT_PREFIX):
alarm_id = job.comment.split(self.COMMENT_PREFIX)[1].strip()
# job.slices is a valid cron slice object, str(job.slices) gives the expression
alarms.append({
"id": alarm_id,
"cron_expression": str(job.slices),
"command": job.command,
"is_enabled": job.is_enabled()
})
alarms.append(
{
"id": alarm_id,
"cron_expression": str(job.slices),
"command": job.command,
"is_enabled": job.is_enabled(),
}
)
return alarms
def set_alarm(self, alarm_id: str | None, cron_expression: str, command: str, is_enabled: bool = True):
def set_alarm(
self,
alarm_id: str | None,
cron_expression: str,
command: str,
is_enabled: bool = True,
):
if not alarm_id:
alarm_id = str(uuid.uuid4())
cron = self._get_cron()
comment = f"{self.COMMENT_PREFIX}{alarm_id}"
# Remove existing if any
cron.remove_all(comment=comment)
# Create new
job = cron.new(command=command, comment=comment)
job.setall(cron_expression)
job.enable(is_enabled)
cron.write()
return alarm_id
api/main.py
+4 -1
View File
@@ -11,16 +11,18 @@ load_dotenv()
API_KEY_NAME = "X-API-Key"
api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=False)
def get_api_key(api_key_header: str = Security(api_key_header)):
expected_api_key = os.getenv("API_KEY")
if not expected_api_key:
# If no key is configured, deny all requests for safety
raise HTTPException(status_code=500, detail="API_KEY not configured on server")
if api_key_header == expected_api_key:
return api_key_header
raise HTTPException(status_code=401, detail="Invalid or missing API Key")
graphql_app = GraphQLRouter(schema)
app = FastAPI(title="Wecker API")
@@ -28,6 +30,7 @@ app = FastAPI(title="Wecker API")
# Add auth dependency to the graphql route
app.include_router(graphql_app, prefix="/graphql", dependencies=[Depends(get_api_key)])
@app.get("/health")
def health_check():
return {"status": "ok"}
api/schema.py
+16 -4
View File
@@ -1,12 +1,16 @@
import strawberry
from typing import List, Optional
import os
import sys
from pathlib import Path
from api.crontab_manager import CrontabManager
def get_manager():
tabfile = os.getenv("TABFILE")
return CrontabManager(tabfile=tabfile)
@strawberry.type
class Alarm:
id: str
@@ -14,6 +18,7 @@ class Alarm:
command: str
is_enabled: bool
@strawberry.type
class Query:
@strawberry.field
@@ -30,28 +35,34 @@ class Query:
return Alarm(**a)
return None
@strawberry.type
class Mutation:
@strawberry.field
def set_alarm(
self,
cron_expression: str,
command: str,
command: Optional[str] = None,
is_enabled: bool = True,
id: Optional[str] = None
id: Optional[str] = None,
) -> Alarm:
if command is None:
project_root = Path(__file__).parent.parent.absolute()
python_exec = sys.executable
command = f"cd {project_root} && {python_exec} wecker.py > wecker.log 2>&1"
manager = get_manager()
new_id = manager.set_alarm(
alarm_id=id,
cron_expression=cron_expression,
command=command,
is_enabled=is_enabled
is_enabled=is_enabled,
)
return Alarm(
id=new_id,
cron_expression=cron_expression,
command=command,
is_enabled=is_enabled
is_enabled=is_enabled,
)
@strawberry.field
@@ -65,4 +76,5 @@ class Mutation:
return True
return False
schema = strawberry.Schema(query=Query, mutation=Mutation)
tests/test_api.py
+13 -8
View File
@@ -13,6 +13,7 @@ from api.main import app # noqa: E402
client = TestClient(app)
def test_auth_missing():
query = """
query {
@@ -24,6 +25,7 @@ def test_auth_missing():
response = client.post("/graphql", json={"query": query})
assert response.status_code == 401
def test_auth_invalid():
query = """
query {
@@ -32,12 +34,15 @@ def test_auth_invalid():
}
}
"""
response = client.post("/graphql", json={"query": query}, headers={"X-API-Key": "wrong"})
response = client.post(
"/graphql", json={"query": query}, headers={"X-API-Key": "wrong"}
)
assert response.status_code == 401
def test_graphql_workflow():
headers = {"X-API-Key": "test-secret"}
# 1. Get empty alarms
query_get = """
query {
@@ -49,7 +54,7 @@ def test_graphql_workflow():
res = client.post("/graphql", json={"query": query_get}, headers=headers)
assert res.status_code == 200
assert res.json()["data"]["getAlarms"] == []
# 2. Set alarm
mutation_set = """
mutation {
@@ -68,12 +73,12 @@ def test_graphql_workflow():
assert alarm["command"] == "python wecker.py"
assert alarm["isEnabled"] is True
alarm_id = alarm["id"]
# 3. Get alarms lists it
res = client.post("/graphql", json={"query": query_get}, headers=headers)
assert len(res.json()["data"]["getAlarms"]) == 1
assert res.json()["data"]["getAlarms"][0]["id"] == alarm_id
# 4. Get specific alarm
query_one = f"""
query {{
@@ -86,7 +91,7 @@ def test_graphql_workflow():
res = client.post("/graphql", json={"query": query_one}, headers=headers)
assert res.json()["data"]["getAlarm"]["id"] == alarm_id
assert res.json()["data"]["getAlarm"]["cronExpression"] == "30 7 * * *"
# 5. Update alarm
mutation_update = f"""
mutation {{
@@ -102,7 +107,7 @@ def test_graphql_workflow():
assert alarm_updated["id"] == alarm_id
assert alarm_updated["cronExpression"] == "0 8 * * *"
assert alarm_updated["isEnabled"] is False
# 6. Delete alarm
mutation_delete = f"""
mutation {{
@@ -111,7 +116,7 @@ def test_graphql_workflow():
"""
res = client.post("/graphql", json={"query": mutation_delete}, headers=headers)
assert res.json()["data"]["deleteAlarm"] is True
# 7. List again is empty
res = client.post("/graphql", json={"query": query_get}, headers=headers)
assert res.json()["data"]["getAlarms"] == []
tests/test_crontab.py
+13 -8
View File
@@ -3,16 +3,19 @@ import tempfile
import uuid
from api.crontab_manager import CrontabManager
@pytest.fixture
def crontab_file():
with tempfile.NamedTemporaryFile(mode='w+', delete=False) as f:
with tempfile.NamedTemporaryFile(mode="w+", delete=False) as f:
pass
yield f.name
def test_crontab_manager_empty(crontab_file):
manager = CrontabManager(tabfile=crontab_file)
assert manager.get_alarms() == []
def test_add_and_list_alarm(crontab_file):
manager = CrontabManager(tabfile=crontab_file)
alarm_id = str(uuid.uuid4())
@@ -20,9 +23,9 @@ def test_add_and_list_alarm(crontab_file):
alarm_id=alarm_id,
cron_expression="30 7 * * *",
command="python wecker.py",
is_enabled=True
is_enabled=True,
)
alarms = manager.get_alarms()
assert len(alarms) == 1
assert alarms[0]["id"] == alarm_id
@@ -30,6 +33,7 @@ def test_add_and_list_alarm(crontab_file):
assert alarms[0]["command"] == "python wecker.py"
assert alarms[0]["is_enabled"] is True
def test_update_alarm(crontab_file):
manager = CrontabManager(tabfile=crontab_file)
alarm_id = "test-id"
@@ -37,27 +41,28 @@ def test_update_alarm(crontab_file):
alarm_id=alarm_id,
cron_expression="30 7 * * *",
command="python wecker.py",
is_enabled=True
is_enabled=True,
)
manager.set_alarm(
alarm_id=alarm_id,
cron_expression="0 8 * * *",
command="python wecker.py --loud",
is_enabled=False
is_enabled=False,
)
alarms = manager.get_alarms()
assert len(alarms) == 1
assert alarms[0]["cron_expression"] == "0 8 * * *"
assert alarms[0]["command"] == "python wecker.py --loud"
assert alarms[0]["is_enabled"] is False
def test_delete_alarm(crontab_file):
manager = CrontabManager(tabfile=crontab_file)
alarm_id = "test-id-2"
manager.set_alarm(alarm_id, "0 0 * * *", "cmd", True)
assert len(manager.get_alarms()) == 1
manager.delete_alarm(alarm_id)
assert len(manager.get_alarms()) == 0
tests/test_wecker.py
+40 -33
View File
@@ -3,32 +3,36 @@ import sys
from unittest.mock import MagicMock, patch
# Mock RPi.GPIO and pygame before importing wecker
sys.modules['RPi'] = MagicMock()
sys.modules['RPi.GPIO'] = MagicMock()
sys.modules['pygame'] = MagicMock()
sys.modules["RPi"] = MagicMock()
sys.modules["RPi.GPIO"] = MagicMock()
sys.modules["pygame"] = MagicMock()
import wecker # noqa: E402
@pytest.fixture
def mock_gpio():
with patch.object(wecker, 'GPIO') as mock:
with patch.object(wecker, "GPIO") as mock:
mock.LOW = 0
mock.HIGH = 1
yield mock
@pytest.fixture
def mock_pygame():
with patch.object(wecker, 'pygame') as mock:
with patch.object(wecker, "pygame") as mock:
yield mock
def test_set_led(mock_gpio):
wecker.set_led(True)
mock_gpio.output.assert_called_with(wecker.LED_PIN, mock_gpio.LOW)
wecker.set_led(False)
mock_gpio.output.assert_called_with(wecker.LED_PIN, mock_gpio.HIGH)
@patch('wecker.time.sleep')
@patch("wecker.time.sleep")
def test_blink_led(mock_sleep, mock_gpio):
wecker.blink_led(2)
# 2 blinks = 4 sleep calls, 2 set_led(True), 2 set_led(False)
@@ -36,35 +40,37 @@ def test_blink_led(mock_sleep, mock_gpio):
# GPIO output called 4 times total (on, off, on, off)
assert mock_gpio.output.call_count == 4
@patch('wecker.time.time')
@patch("wecker.time.time")
def test_run_alarm_start_to_wait(mock_time, mock_gpio, mock_pygame):
# Mock pygame.mixer.get_init() to return True so music plays
mock_pygame.mixer.get_init.return_value = True
# Just run it in test mode, it should execute the loop once and exit
wecker.run_alarm(test_mode=True)
assert mock_pygame.mixer.music.play.called
def test_state_machine_evaluation(mock_gpio, mock_pygame):
clock = wecker.AlarmClock()
# Transition to ringing -> wait before blink
clock.update(100.0, True) # button press
clock.update(100.0, True) # button press
assert clock.state == wecker.STATE_WAIT_BEFORE_BLINK
# Wait 3 seconds -> blinking
with patch('wecker.blink_led'):
with patch("wecker.blink_led"):
clock.update(103.1, False)
assert clock.state == wecker.STATE_BLINKING
with patch('wecker.blink_led'):
with patch("wecker.blink_led"):
clock.update(103.2, False)
assert clock.state == wecker.STATE_WAIT_FOR_INPUT
assert clock.target_blinks >= 1
# Set the user presses to be correct
clock.target_blinks = 3
# Press 1
clock.update(104.0, True)
clock.update(104.1, False)
@@ -74,46 +80,47 @@ def test_state_machine_evaluation(mock_gpio, mock_pygame):
# Press 3
clock.update(105.0, True)
clock.update(105.1, False)
assert clock.user_presses == 3
# Wait 3 seconds to evaluate
clock.update(108.2, False) # Triggers state change
keep_running = clock.update(108.3, False) # Triggers evaluation
clock.update(108.2, False) # Triggers state change
keep_running = clock.update(108.3, 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
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
@patch('wecker.time.time')
@patch("wecker.time.time")
def test_blinking_updates_time_correctly(mock_time, mock_gpio, mock_pygame):
clock = wecker.AlarmClock()
clock.state = wecker.STATE_BLINKING
clock.target_blinks = 4
mock_time.return_value = 200.0
with patch('wecker.blink_led'):
clock.update(100.0, False)
with patch("wecker.blink_led"):
clock.update(100.0, False)
assert clock.state == wecker.STATE_WAIT_FOR_INPUT
assert clock.last_interaction_time == 200.0
wecker.py
+43 -32
View File
@@ -8,11 +8,8 @@ import sys
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(message)s',
handlers=[
logging.FileHandler("wecker.log"),
logging.StreamHandler()
]
format="%(asctime)s - %(message)s",
handlers=[logging.FileHandler("wecker.log"), logging.StreamHandler()],
)
# GPIO Setup
@@ -33,7 +30,7 @@ try:
except Exception as e:
logging.error(f"Error loading audio file: {e}")
# Don't exit in test mode
if 'pytest' not in sys.modules:
if "pytest" not in sys.modules:
sys.exit(1)
# State Machine states for the flow
@@ -50,14 +47,16 @@ user_presses = 0
last_interaction_time = 0
button_was_pressed = False
def set_led(on):
"""Turns the LED on or off (LOW = ON, HIGH = OFF)"""
if 'GPIO' in globals() and hasattr(GPIO, 'output'):
if "GPIO" in globals() and hasattr(GPIO, "output"):
if on:
GPIO.output(LED_PIN, GPIO.LOW)
else:
GPIO.output(LED_PIN, GPIO.HIGH)
def blink_led(times):
"""Blinks the LED a specific number of times (blocking)"""
for _ in range(times):
@@ -66,6 +65,7 @@ def blink_led(times):
set_led(False)
time.sleep(0.3) # LED off for 300ms
class AlarmClock:
def __init__(self):
self.state = STATE_RINGING
@@ -84,66 +84,76 @@ class AlarmClock:
if self.state == STATE_RINGING:
if button_just_pressed:
logging.info("Alarm button pressed! Puzzle started. Waiting 3 seconds...")
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
elif self.state == STATE_BLINKING:
# Move the blink_led out of the update loop for testability,
# Move the blink_led out of the update loop for testability,
# or just call it directly. Here we call it.
blink_led(self.target_blinks)
logging.info("Blinking finished. Waiting for input...")
self.state = STATE_WAIT_FOR_INPUT
self.user_presses = 0
self.last_interaction_time = time.time() # Use time.time() to account for blocking blink_led
if 'GPIO' in globals() and hasattr(GPIO, 'input'):
self.button_was_pressed = (GPIO.input(BUTTON_PIN) == GPIO.LOW)
self.last_interaction_time = (
time.time()
) # Use time.time() to account for blocking blink_led
if "GPIO" in globals() and hasattr(GPIO, "input"):
self.button_was_pressed = GPIO.input(BUTTON_PIN) == GPIO.LOW
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}")
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():
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
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
return True # Keep running
def run_alarm(test_mode=False):
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():
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
@@ -153,22 +163,23 @@ def run_alarm(test_mode=False):
try:
while True:
now = time.time()
if 'GPIO' in globals() and hasattr(GPIO, 'input'):
is_pressed = (GPIO.input(BUTTON_PIN) == GPIO.LOW)
if "GPIO" in globals() and hasattr(GPIO, "input"):
is_pressed = GPIO.input(BUTTON_PIN) == GPIO.LOW
else:
is_pressed = False
keep_running = clock.update(now, is_pressed)
if not keep_running or test_mode:
break
time.sleep(0.02)
except KeyboardInterrupt:
logging.info("Manually aborted (CTRL+C).")
if __name__ == "__main__":
try:
run_alarm()