diff --git a/Laid Back - Sunshine Reggae.mp3 b/Laid Back - Sunshine Reggae.mp3 new file mode 100644 index 0000000..6fb5b18 Binary files /dev/null and b/Laid Back - Sunshine Reggae.mp3 differ diff --git a/README.md b/README.md index f9050db..d54c46e 100644 --- a/README.md +++ b/README.md @@ -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 } } ``` diff --git a/api.pid b/api.pid new file mode 100644 index 0000000..dd41798 --- /dev/null +++ b/api.pid @@ -0,0 +1 @@ +17958 diff --git a/api/crontab_manager.py b/api/crontab_manager.py index f907636..18f7602 100644 --- a/api/crontab_manager.py +++ b/api/crontab_manager.py @@ -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 diff --git a/api/main.py b/api/main.py index 33c2a1e..9ac29d9 100644 --- a/api/main.py +++ b/api/main.py @@ -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"} diff --git a/api/schema.py b/api/schema.py index 16eba66..c05d5f9 100644 --- a/api/schema.py +++ b/api/schema.py @@ -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) diff --git a/tests/test_api.py b/tests/test_api.py index 1d7db80..90a0351 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -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"] == [] diff --git a/tests/test_crontab.py b/tests/test_crontab.py index 3b49336..0bbb89e 100644 --- a/tests/test_crontab.py +++ b/tests/test_crontab.py @@ -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 diff --git a/tests/test_wecker.py b/tests/test_wecker.py index 4b73d63..1c80612 100644 --- a/tests/test_wecker.py +++ b/tests/test_wecker.py @@ -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 diff --git a/wecker.py b/wecker.py index 9ff750b..5acdedc 100644 --- a/wecker.py +++ b/wecker.py @@ -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()