home_assistant/sunrise_alarm.py

#!/usr/bin/env python3
"""
Poll local time and trigger a WLED sunrise once per day.

Example:
    python sunrise_alarm.py
    python sunrise_alarm.py --alarm 08:00 --duration 30 --poll-seconds 20
"""
import argparse
import json
import sys
import time
from copy import deepcopy
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, Dict

from wled_push import DEFAULT_HOST, push


ROOT = Path(__file__).resolve().parent
DEFAULT_CONFIG = ROOT / "wled_config" / "sunrise_8am.json"
DEFAULT_STATE = ROOT / ".sunrise_alarm_state.json"


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Trigger a WLED sunrise once per day")
    parser.add_argument("--host", "-H", default=DEFAULT_HOST, help=f"WLED host (default: {DEFAULT_HOST})")
    parser.add_argument("--config", "-c", default=str(DEFAULT_CONFIG), help="Sunrise payload JSON")
    parser.add_argument("--alarm", "-a", default="08:00", help="Alarm time in local HH:MM (default: 08:00)")
    parser.add_argument(
        "--duration",
        "-d",
        type=int,
        default=30,
        help="Sunrise duration in minutes; script starts this many minutes before the alarm",
    )
    parser.add_argument(
        "--poll-seconds",
        "-p",
        type=int,
        default=20,
        help="How often to check the clock (default: 20)",
    )
    parser.add_argument(
        "--window-seconds",
        "-w",
        type=int,
        default=60,
        help="Trigger window size in seconds to tolerate clock drift (default: 60)",
    )
    parser.add_argument(
        "--state-file",
        default=str(DEFAULT_STATE),
        help="File used to remember the last day that was triggered",
    )
    parser.add_argument(
        "--run-now",
        action="store_true",
        help="Trigger immediately once and exit",
    )
    return parser.parse_args()


def parse_alarm_time(value: str) -> tuple[int, int]:
    try:
        hour_str, minute_str = value.split(":", 1)
        hour = int(hour_str)
        minute = int(minute_str)
    except ValueError as exc:
        raise argparse.ArgumentTypeError(f"Invalid alarm time '{value}', expected HH:MM") from exc

    if not (0 <= hour <= 23 and 0 <= minute <= 59):
        raise argparse.ArgumentTypeError(f"Invalid alarm time '{value}', expected HH:MM")
    return hour, minute


def load_payload(path: Path, duration_minutes: int) -> Dict[str, Any]:
    with path.open() as handle:
        payload = json.load(handle)

    payload = deepcopy(payload)
    payload.setdefault("nl", {})
    payload["nl"]["on"] = True
    payload["nl"]["dur"] = duration_minutes
    payload["nl"].setdefault("mode", 3)
    payload["nl"].setdefault("tbri", 255)
    payload["on"] = True
    payload.setdefault("bri", 1)
    return payload


def load_last_triggered(path: Path) -> str:
    if not path.exists():
        return ""

    try:
        with path.open() as handle:
            data = json.load(handle)
    except (json.JSONDecodeError, OSError):
        return ""

    return str(data.get("last_triggered_date", ""))


def save_last_triggered(path: Path, day: str) -> None:
    path.write_text(json.dumps({"last_triggered_date": day}, indent=2) + "\n")


def compute_trigger_time(now: datetime, alarm_hour: int, alarm_minute: int, duration_minutes: int) -> datetime:
    alarm_at = now.replace(hour=alarm_hour, minute=alarm_minute, second=0, microsecond=0)
    return alarm_at - timedelta(minutes=duration_minutes)


def should_trigger(now: datetime, trigger_at: datetime, window_seconds: int, last_triggered_date: str) -> bool:
    if last_triggered_date == now.date().isoformat():
        return False
    return trigger_at <= now < trigger_at + timedelta(seconds=window_seconds)


def trigger(host: str, payload: Dict[str, Any], state_file: Path) -> None:
    print(f"[{datetime.now().astimezone().isoformat(timespec='seconds')}] triggering sunrise on {host}")
    push(host, payload, return_state=False)
    save_last_triggered(state_file, datetime.now().date().isoformat())


def main() -> int:
    args = parse_args()
    alarm_hour, alarm_minute = parse_alarm_time(args.alarm)
    config_path = Path(args.config).expanduser().resolve()
    state_file = Path(args.state_file).expanduser().resolve()

    if not config_path.exists():
        print(f"Config file not found: {config_path}", file=sys.stderr)
        return 1
    if args.duration <= 0:
        print("--duration must be greater than 0", file=sys.stderr)
        return 1
    if args.poll_seconds <= 0:
        print("--poll-seconds must be greater than 0", file=sys.stderr)
        return 1
    if args.window_seconds <= 0:
        print("--window-seconds must be greater than 0", file=sys.stderr)
        return 1

    payload = load_payload(config_path, args.duration)

    if args.run_now:
        trigger(args.host, payload, state_file)
        return 0

    print(
        f"Polling for sunrise: alarm={args.alarm}, duration={args.duration}m, "
        f"start={args.alarm} minus {args.duration}m, host={args.host}"
    )

    while True:
        now = datetime.now().astimezone()
        trigger_at = compute_trigger_time(now, alarm_hour, alarm_minute, args.duration)
        last_triggered_date = load_last_triggered(state_file)

        if should_trigger(now, trigger_at, args.window_seconds, last_triggered_date):
            try:
                payload = load_payload(config_path, args.duration)
                trigger(args.host, payload, state_file)
            except Exception as exc:  # noqa: BLE001
                print(f"[{now.isoformat(timespec='seconds')}] trigger failed: {exc}", file=sys.stderr)

        time.sleep(args.poll_seconds)


if __name__ == "__main__":
    raise SystemExit(main())