115 lines
3.9 KiB
Python
115 lines
3.9 KiB
Python
|
"""Alarm scheduler — load config and check firing schedule."""
|
||
|
|
|
||
|
|
import json
|
||
|
|
import logging
|
||
|
|
import re
|
||
|
|
from datetime import datetime
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
log = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
DEFAULT_CONFIG_PATH = Path(__file__).parent / "config" / "alarms.json"
|
||
|
|
|
||
|
|
VALID_DAYS = {"Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"}
|
||
|
|
TIME_RE = re.compile(r"^([01]\d|2[0-3])[0-5]\d$")
|
||
|
|
|
||
|
|
|
||
|
|
def load_config(path: Path) -> dict | None:
|
||
|
|
"""Read and validate alarm config JSON.
|
||
|
|
|
||
|
|
Returns the config dict, or None if the file is missing, empty, or invalid.
|
||
|
|
Never raises — logs warnings and returns None on any problem.
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
text = path.read_text(encoding="utf-8").strip()
|
||
|
|
except FileNotFoundError:
|
||
|
|
log.warning("Config file not found: %s", path)
|
||
|
|
return None
|
||
|
|
except OSError as e:
|
||
|
|
log.warning("Cannot read config %s: %s", path, e)
|
||
|
|
return None
|
||
|
|
|
||
|
|
if not text:
|
||
|
|
log.warning("Config file is empty: %s", path)
|
||
|
|
return None
|
||
|
|
|
||
|
|
try:
|
||
|
|
data = json.loads(text)
|
||
|
|
except json.JSONDecodeError as e:
|
||
|
|
log.warning("Invalid JSON in %s: %s", path, e)
|
||
|
|
return None
|
||
|
|
|
||
|
|
# Empty object or empty list means "no alarms"
|
||
|
|
if not data or (isinstance(data, list) and len(data) == 0):
|
||
|
|
log.info("Config is empty — no alarms configured")
|
||
|
|
return None
|
||
|
|
|
||
|
|
if not isinstance(data, dict):
|
||
|
|
log.warning("Config must be a JSON object, got %s", type(data).__name__)
|
||
|
|
return None
|
||
|
|
|
||
|
|
# Validate alarm_time (required)
|
||
|
|
alarm_time = data.get("alarm_time")
|
||
|
|
if alarm_time is None:
|
||
|
|
log.warning("Missing required field 'alarm_time'")
|
||
|
|
return None
|
||
|
|
if not isinstance(alarm_time, str) or not TIME_RE.match(alarm_time):
|
||
|
|
log.warning("Invalid alarm_time '%s' — must be 4-digit HHMM (e.g. '0730')", alarm_time)
|
||
|
|
return None
|
||
|
|
|
||
|
|
# Validate alarm_days (optional)
|
||
|
|
alarm_days = data.get("alarm_days")
|
||
|
|
if alarm_days is not None:
|
||
|
|
if not isinstance(alarm_days, list) or not all(isinstance(d, str) for d in alarm_days):
|
||
|
|
log.warning("alarm_days must be a list of strings")
|
||
|
|
return None
|
||
|
|
bad = [d for d in alarm_days if d not in VALID_DAYS]
|
||
|
|
if bad:
|
||
|
|
log.warning("Invalid day abbreviations in alarm_days: %s", bad)
|
||
|
|
return None
|
||
|
|
|
||
|
|
# Validate alarm_dates (optional)
|
||
|
|
alarm_dates = data.get("alarm_dates")
|
||
|
|
if alarm_dates is not None:
|
||
|
|
if not isinstance(alarm_dates, list) or not all(isinstance(d, str) for d in alarm_dates):
|
||
|
|
log.warning("alarm_dates must be a list of strings")
|
||
|
|
return None
|
||
|
|
date_re = re.compile(r"^(0[1-9]|1[0-2])/(0[1-9]|[12]\d|3[01])$")
|
||
|
|
bad = [d for d in alarm_dates if not date_re.match(d)]
|
||
|
|
if bad:
|
||
|
|
log.warning("Invalid date formats in alarm_dates (expected MM/DD): %s", bad)
|
||
|
|
return None
|
||
|
|
|
||
|
|
log.info("Alarm config loaded: time=%s days=%s", alarm_time, alarm_days or "(every day)")
|
||
|
|
return data
|
||
|
|
|
||
|
|
|
||
|
|
def should_fire(config: dict) -> bool:
|
||
|
|
"""Check if the alarm should fire right now based on config schedule.
|
||
|
|
|
||
|
|
Rules:
|
||
|
|
- alarm_time must match current HHMM
|
||
|
|
- If alarm_days is present, today's 3-letter abbreviation must be in the list
|
||
|
|
- If alarm_days is absent but alarm_dates is present, today's MM/DD must match
|
||
|
|
- If neither alarm_days nor alarm_dates is present, fires every day
|
||
|
|
- If both are present, alarm_days wins (alarm_dates ignored)
|
||
|
|
"""
|
||
|
|
now = datetime.now()
|
||
|
|
current_hhmm = now.strftime("%H%M")
|
||
|
|
|
||
|
|
if config["alarm_time"] != current_hhmm:
|
||
|
|
return False
|
||
|
|
|
||
|
|
alarm_days = config.get("alarm_days")
|
||
|
|
alarm_dates = config.get("alarm_dates")
|
||
|
|
|
||
|
|
# alarm_days takes priority over alarm_dates
|
||
|
|
if alarm_days is not None:
|
||
|
|
return now.strftime("%a") in alarm_days
|
||
|
|
|
||
|
|
if alarm_dates is not None:
|
||
|
|
return now.strftime("%m/%d") in alarm_dates
|
||
|
|
|
||
|
|
# Neither specified — fire every day
|
||
|
|
return True
|