homelab/ansible/roles/jmri/templates/jmri-monitor.py.j2

#!/opt/jmri-monitor/bin/python3
"""
jmri-monitor — Leviton Decora Smart Wi-Fi layout power monitor
Managed by Ansible — do not edit manually.

Polls the Leviton cloud API for the "{{ jmri_leviton_switch_name }}" switch state.
  - Switch ON  after being OFF → systemctl start jmri.service
  - Switch OFF for {{ jmri_monitor_stop_delay }}s  → systemctl stop jmri.service

Quiet hours {{ jmri_monitor_quiet_start }}:00–{{ jmri_monitor_quiet_end }}:00: no polling (layout assumed off).
"""

import subprocess
import time
import logging
import sys
from datetime import datetime

LEVITON_EMAIL    = "{{ jmri_leviton_email }}"
LEVITON_PASSWORD = "{{ jmri_leviton_password }}"
SWITCH_NAME      = "{{ jmri_leviton_switch_name }}"
POLL_INTERVAL    = {{ jmri_monitor_poll_interval }}
QUIET_START      = {{ jmri_monitor_quiet_start }}
QUIET_END        = {{ jmri_monitor_quiet_end }}
STOP_DELAY       = {{ jmri_monitor_stop_delay }}
JMRI_SERVICE     = "jmri.service"

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [jmri-monitor] %(levelname)s: %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
    stream=sys.stdout,
)
log = logging.getLogger(__name__)


def systemctl(action):
    try:
        result = subprocess.run(
            ["systemctl", action, JMRI_SERVICE],
            capture_output=True, text=True, timeout=30,
        )
        if result.returncode == 0:
            log.info("systemctl %s %s: OK", action, JMRI_SERVICE)
        else:
            log.warning("systemctl %s %s: %s", action, JMRI_SERVICE, result.stderr.strip())
    except Exception as e:
        log.error("systemctl %s failed: %s", action, e)


def is_quiet_hours():
    hour = datetime.now().hour
    if QUIET_START < QUIET_END:
        return QUIET_START <= hour < QUIET_END
    else:
        # wraps midnight e.g. 23–6
        return hour >= QUIET_START or hour < QUIET_END


def get_switch_state():
    """Returns True if switch is ON, False if OFF, None on error."""
    try:
        from decora_wifi import DecoraWiFiSession
        from decora_wifi.models.residential_account import ResidentialAccount

        session = DecoraWiFiSession()
        person = session.login(LEVITON_EMAIL, LEVITON_PASSWORD)
        if not person:
            log.error("Leviton login failed")
            return None

        perms = person.get_residential_permissions()
        for perm in perms:
            acct_id = perm.data.get('residentialAccountId')
            if not acct_id:
                continue
            acct = ResidentialAccount(session, acct_id)
            acct.refresh()
            for residence in acct.get_residences():
                for switch in residence.get_iot_switches():
                    if switch.data.get('name') == SWITCH_NAME:
                        state = switch.data.get('power', 'OFF')
                        session.call_api('/Person/logout', {}, 'post')
                        return state == 'ON'

        all_names = []
        for perm in perms:
            acct_id = perm.data.get('residentialAccountId')
            if acct_id:
                acct = ResidentialAccount(session, acct_id)
                acct.refresh()
                for r in acct.get_residences():
                    all_names += [s.data.get('name') for s in r.get_iot_switches()]
        log.warning("Switch '%s' not found — available: %s", SWITCH_NAME, all_names)
        session.call_api('/Person/logout', {}, 'post')
        return None

    except ImportError:
        log.error("decora_wifi not installed")
        return None
    except Exception as e:
        log.error("Error querying Leviton API: %s", e)
        return None


def main():
    log.info("Layout power monitor starting")
    log.info("Switch: '%s'  Poll: %ds  Quiet: %02d:00–%02d:00  Stop delay: %ds",
             SWITCH_NAME, POLL_INTERVAL, QUIET_START, QUIET_END, STOP_DELAY)

    layout_on = False
    off_since = None

    while True:
        if is_quiet_hours():
            log.debug("Quiet hours — sleeping 60s")
            # If layout was on when quiet hours started, stop JMRI
            if layout_on:
                log.info("Quiet hours began — stopping JMRI")
                layout_on = False
                off_since = None
                systemctl("stop")
            time.sleep(60)
            continue

        state = get_switch_state()

        if state is True:
            off_since = None
            if not layout_on:
                log.info("Layout switch ON — starting JMRI")
                layout_on = True
                systemctl("start")

        elif state is False:
            if layout_on:
                if off_since is None:
                    off_since = time.monotonic()
                    log.info("Layout switch OFF — waiting %ds before stopping JMRI", STOP_DELAY)
                elif time.monotonic() - off_since >= STOP_DELAY:
                    log.info("Layout switch OFF for %ds — stopping JMRI", STOP_DELAY)
                    layout_on = False
                    off_since = None
                    systemctl("stop")
            else:
                off_since = None

        else:
            # API error — don't change state, try again next poll
            log.warning("Could not determine switch state — retrying in %ds", POLL_INTERVAL)

        time.sleep(POLL_INTERVAL)


if __name__ == "__main__":
    main()