136 lines
4.5 KiB
Python
136 lines
4.5 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$")
|
|
DATE_RE = re.compile(r"^(0[1-9]|1[0-2])/(0[1-9]|[12]\d|3[01])$")
|
|
|
|
|
|
def _validate_entry(entry: dict, index: int) -> dict | None:
|
|
"""Validate a single alarm entry. Returns it if valid, None otherwise."""
|
|
if not isinstance(entry, dict):
|
|
log.warning("Alarm #%d: expected object, got %s", index, type(entry).__name__)
|
|
return None
|
|
|
|
alarm_time = entry.get("alarm_time")
|
|
if alarm_time is None:
|
|
log.warning("Alarm #%d: missing required field 'alarm_time'", index)
|
|
return None
|
|
if not isinstance(alarm_time, str) or not TIME_RE.match(alarm_time):
|
|
log.warning("Alarm #%d: invalid alarm_time '%s' — must be 4-digit HHMM", index, alarm_time)
|
|
return None
|
|
|
|
alarm_days = entry.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 #%d: alarm_days must be a list of strings", index)
|
|
return None
|
|
bad = [d for d in alarm_days if d not in VALID_DAYS]
|
|
if bad:
|
|
log.warning("Alarm #%d: invalid day abbreviations: %s", index, bad)
|
|
return None
|
|
|
|
alarm_dates = entry.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 #%d: alarm_dates must be a list of strings", index)
|
|
return None
|
|
bad = [d for d in alarm_dates if not DATE_RE.match(d)]
|
|
if bad:
|
|
log.warning("Alarm #%d: invalid date formats (expected MM/DD): %s", index, bad)
|
|
return None
|
|
|
|
log.info("Alarm #%d: time=%s days=%s", index, alarm_time, alarm_days or "(every day)")
|
|
return entry
|
|
|
|
|
|
def load_config(path: Path) -> list[dict] | None:
|
|
"""Read and validate alarm config JSON.
|
|
|
|
Accepts either a single alarm object or an array of alarm objects.
|
|
Returns a list of valid alarm dicts, or None if the file is missing,
|
|
empty, or contains no valid entries.
|
|
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
|
|
|
|
if not data:
|
|
log.info("Config is empty — no alarms configured")
|
|
return None
|
|
|
|
# Normalize to list
|
|
if isinstance(data, dict):
|
|
entries = [data]
|
|
elif isinstance(data, list):
|
|
entries = data
|
|
else:
|
|
log.warning("Config must be a JSON object or array, got %s", type(data).__name__)
|
|
return None
|
|
|
|
valid = []
|
|
for i, entry in enumerate(entries):
|
|
result = _validate_entry(entry, i)
|
|
if result is not None:
|
|
valid.append(result)
|
|
|
|
if not valid:
|
|
log.warning("No valid alarm entries in %s", path)
|
|
return None
|
|
|
|
log.info("Loaded %d alarm(s) from %s", len(valid), path)
|
|
return valid
|
|
|
|
|
|
def should_fire(config: dict) -> bool:
|
|
"""Check if a single alarm entry should fire right now.
|
|
|
|
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")
|
|
|
|
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
|
|
|
|
return True
|