#!/usr/bin/env python3 """ WZDX Field Device Monitor ========================= Polls a WZDX field-device feed on a fixed interval, detects field devices that are no longer communicating, and emails an alert to one or more recipients. A device is considered NOT COMMUNICATING when EITHER of these is true: * its `device_status` is one of DOWN_STATUSES (e.g. "error" / "unknown"), OR * its `update_date` (last report time) is older than STALE_THRESHOLD_MINUTES. Behavior: * One alert email per poll cycle listing every device that has *newly* gone down since the previous cycle (no repeated spam while a device stays down). * A recovery email when a previously-down device starts reporting normally again. Only dependency beyond the standard library is `requests`: pip install requests """ import sys import time import smtplib import logging from datetime import datetime, timezone from email.message import EmailMessage import requests from requests.auth import HTTPBasicAuth # ============================ CONFIGURATION ============================ # --- WZDX feed --- WZDX_FEED_URL = "https://projects.slndrtech.com/Salander/api/wzdx/v42/deviceFeed/5tjm0tva5is76sd33mp85gpj4s" FEED_USERNAME = "JP Story" FEED_PASSWORD = "l3tmeslndr" FEED_TIMEOUT_SECONDS = 30 # --- Alert recipients --- ALERT_RECIPIENTS = ["jp@slndrtech.com"] # --- Outbound email (SMTP) --- SMTP_HOST = "smtp-relay.brevo.com" SMTP_PORT = 587 SMTP_USERNAME = "a7bfc2001@smtp-brevo.com" SMTP_PASSWORD = "bskWFnIXIvjIr67" SMTP_USE_TLS = True # True for STARTTLS (port 587) EMAIL_FROM = "wzdx-monitor@example.com" # --- Detection settings --- POLL_INTERVAL_SECONDS = 60 # how often to poll the feed STALE_THRESHOLD_MINUTES = 5 # no update in this many minutes = down DOWN_STATUSES = {"error", "unknown"} # device_status values treated as down SEND_RECOVERY_ALERTS = True # Alert if the feed itself becomes unreachable (won't repeat while it stays down) ALERT_ON_FEED_FAILURE = True # ====================================================================== logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)-7s %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) log = logging.getLogger("wzdx-monitor") # --------------------------------------------------------------------------- # Feed fetching # --------------------------------------------------------------------------- def fetch_feed(): """Fetch and JSON-decode the WZDX feed. Returns a dict, or raises on error.""" auth = None if FEED_USERNAME or FEED_PASSWORD: auth = HTTPBasicAuth(FEED_USERNAME, FEED_PASSWORD) resp = requests.get(WZDX_FEED_URL, auth=auth, timeout=FEED_TIMEOUT_SECONDS) resp.raise_for_status() return resp.json() # --------------------------------------------------------------------------- # Feed parsing # --------------------------------------------------------------------------- def _first(props, *keys): """Return the first present, non-None value among the given keys.""" for k in keys: if isinstance(props, dict) and props.get(k) is not None: return props[k] return None def extract_devices(feed): """ Normalize the WZDX FeatureCollection into a list of simple device dicts: {id, name, type, status, update_date (datetime|None)} Field-device details usually live under properties.core_details, but this reads flexibly so it also works if the producer flattens them. """ devices = [] for feature in feed.get("features", []) or []: props = feature.get("properties", {}) or {} core = props.get("core_details", props) # fall back to flat properties device_id = ( feature.get("id") or _first(core, "id", "name") or _first(props, "id", "name") ) name = _first(core, "name", "description") or device_id or "(unnamed device)" dev_type = _first(core, "device_type") or _first(props, "device_type") status = _first(core, "device_status", "status") or _first(props, "device_status", "status") raw_update = _first(core, "update_date", "updated_timestamp") or _first( props, "update_date", "updated_timestamp" ) devices.append( { "id": str(device_id) if device_id is not None else name, "name": name, "type": dev_type, "status": (status or "").lower() if isinstance(status, str) else status, "update_date": _parse_timestamp(raw_update), "raw_update": raw_update, } ) return devices def _parse_timestamp(value): """Parse an ISO-8601 timestamp (WZDX style) into an aware UTC datetime.""" if not value or not isinstance(value, str): return None text = value.strip().replace("Z", "+00:00") try: dt = datetime.fromisoformat(text) except ValueError: return None if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt.astimezone(timezone.utc) # --------------------------------------------------------------------------- # Detection # --------------------------------------------------------------------------- def evaluate_devices(devices, now=None): """ Return {device_id: {"device": , "reason": }} for every device currently considered NOT communicating. """ now = now or datetime.now(timezone.utc) down = {} for d in devices: reason = None if d["status"] in DOWN_STATUSES: reason = f"device_status = '{d['status']}'" elif d["update_date"] is not None: age_min = (now - d["update_date"]).total_seconds() / 60.0 if age_min > STALE_THRESHOLD_MINUTES: reason = ( f"no update in {age_min:.1f} min " f"(last update {d['raw_update']}, threshold " f"{STALE_THRESHOLD_MINUTES} min)" ) if reason: down[d["id"]] = {"device": d, "reason": reason} return down # --------------------------------------------------------------------------- # Email # --------------------------------------------------------------------------- def send_email(subject, body): msg = EmailMessage() msg["Subject"] = subject msg["From"] = EMAIL_FROM msg["To"] = ", ".join(ALERT_RECIPIENTS) msg.set_content(body) with smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=30) as server: if SMTP_USE_TLS: server.starttls() if SMTP_USERNAME or SMTP_PASSWORD: server.login(SMTP_USERNAME, SMTP_PASSWORD) server.send_message(msg) log.info("Alert email sent to %s | %s", ", ".join(ALERT_RECIPIENTS), subject) def _device_line(entry): d = entry["device"] dtype = f" [{d['type']}]" if d.get("type") else "" return f" - {d['name']}{dtype} (id: {d['id']})\n reason: {entry['reason']}" def send_down_alert(newly_down): ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") count = len(newly_down) subject = f"[WZDX ALERT] {count} device(s) not communicating" lines = [ f"{count} WZDX field device(s) stopped communicating as of {ts}.", f"Feed: {WZDX_FEED_URL}", "", "Affected devices:", ] lines += [_device_line(e) for e in newly_down.values()] send_email(subject, "\n".join(lines)) def send_recovery_alert(recovered): ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") count = len(recovered) subject = f"[WZDX RECOVERED] {count} device(s) communicating again" lines = [ f"{count} WZDX field device(s) resumed communicating as of {ts}.", f"Feed: {WZDX_FEED_URL}", "", "Recovered devices:", ] for d in recovered.values(): dtype = f" [{d['type']}]" if d.get("type") else "" lines.append(f" - {d['name']}{dtype} (id: {d['id']})") send_email(subject, "\n".join(lines)) def send_feed_failure_alert(error): subject = "[WZDX ALERT] Feed unreachable" body = ( f"The WZDX monitor could not retrieve the feed as of " f"{datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC')}.\n\n" f"Feed: {WZDX_FEED_URL}\nError: {error}" ) send_email(subject, body) # --------------------------------------------------------------------------- # Main loop # --------------------------------------------------------------------------- def run_once(state): """ Execute a single poll cycle, mutating `state` in place. state = { "down": {id: {"device", "reason"}}, # devices currently considered down "feed_failed": bool, # whether last cycle failed to fetch } """ try: feed = fetch_feed() except Exception as exc: # network / auth / decode errors log.error("Failed to fetch feed: %s", exc) if ALERT_ON_FEED_FAILURE and not state["feed_failed"]: try: send_feed_failure_alert(exc) except Exception as mail_exc: log.error("Could not send feed-failure email: %s", mail_exc) state["feed_failed"] = True return if state["feed_failed"]: log.info("Feed reachable again.") state["feed_failed"] = False devices = extract_devices(feed) present_ids = {d["id"] for d in devices} down_now = evaluate_devices(devices) down_before = state["down"] # Newly down = down now but not previously flagged. newly_down = {k: v for k, v in down_now.items() if k not in down_before} # Recovered = was down, is still present in the feed, and is now healthy. recovered = { k: v["device"] for k, v in down_before.items() if k in present_ids and k not in down_now } # Devices that were down and have vanished from the feed are dropped quietly # (a work zone ending legitimately removes its devices). vanished = [k for k in down_before if k not in present_ids and k not in down_now] log.info( "Polled %d device(s): %d down (%d new), %d recovered, %d vanished.", len(devices), len(down_now), len(newly_down), len(recovered), len(vanished), ) if newly_down: try: send_down_alert(newly_down) except Exception as exc: log.error("Could not send down alert: %s", exc) if recovered and SEND_RECOVERY_ALERTS: try: send_recovery_alert(recovered) except Exception as exc: log.error("Could not send recovery alert: %s", exc) state["down"] = down_now def main(): log.info("Starting WZDX device monitor (poll every %ss).", POLL_INTERVAL_SECONDS) log.info("Feed: %s", WZDX_FEED_URL) state = {"down": {}, "feed_failed": False} while True: start = time.monotonic() try: run_once(state) except Exception as exc: # never let the loop die log.exception("Unexpected error in poll cycle: %s", exc) elapsed = time.monotonic() - start time.sleep(max(0, POLL_INTERVAL_SECONDS - elapsed)) if __name__ == "__main__": try: main() except KeyboardInterrupt: log.info("Stopped by user.") sys.exit(0)