nr-flow-validator/scripts/diag.py

#!/usr/bin/env python3
"""Live diagnostic helpers — invoked by justfile diagnostics recipes."""
import argparse
import json
import re
import sys
import time
import urllib.request

try:
    import paho.mqtt.client as mqtt
except ImportError:
    mqtt = None
import yaml


def show():
    url = "http://192.168.81.147:1880/flows"
    with urllib.request.urlopen(url, timeout=5) as r:
        d = json.load(r)
    print(f"nodes: {len(d)}")
    funcs = [n.get("name", "?") for n in d if n.get("type") == "function"]
    print(f"functions: {funcs}")
    btn = [n for n in d if n.get("type") == "mqtt in" and n.get("topic") == "zigbee2mqtt/Door Button"]
    if btn:
        print(f"button-in wires: {btn[0].get('wires')}")
    out_targets = set()
    for n in d:
        for wire in n.get("wires", []):
            for tid in wire:
                tgt = next((x for x in d if x["id"] == tid), None)
                if tgt and tgt.get("type") == "mqtt out":
                    out_targets.add(tgt.get("topic", "?"))
    print(f"mqtt-out topics: {sorted(out_targets)}")


def diff(flow_path):
    with urllib.request.urlopen("http://192.168.81.147:1880/flows", timeout=5) as r:
        deployed = json.load(r)
    with open(flow_path) as f:
        generated = json.load(f)

    def norm(n):
        out = dict(n)
        for k in ("id", "broker", "group", "page", "theme", "ui", "z"):
            if isinstance(out.get(k), str):
                out[k] = re.sub(r"[a-f0-9]+", "X", out[k])
        if "wires" in out and isinstance(out["wires"], list):
            out["wires"] = [
                [re.sub(r"[a-f0-9]+", "X", x) for x in w] for w in out["wires"]
            ]
        return json.dumps(out, sort_keys=True)

    a = sorted([norm(n) for n in generated])
    b = sorted([norm(n) for n in deployed])
    if a == b:
        print(f"MATCH: deployed flow structurally identical to {flow_path}")
        return 0
    import difflib
    for line in difflib.unified_diff(a, b, lineterm=""):
        print(line)
    return 1


def z2m_devices(z2m_yaml_path):
    with open(z2m_yaml_path) as f:
        d = yaml.safe_load(f)
    print("Devices:")
    for ieee, info in d.get("devices", {}).items():
        print(f"  {info.get('friendly_name'):20s} {ieee}")
    print("Groups:")
    for gid, info in d.get("groups", {}).items():
        print(f"  #{gid} {info.get('friendly_name')}: {info.get('devices')}")


def mqtt_pub(host, port, topic, payload):
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
    client.connect(host, port, 60)
    client.loop_start()
    info = client.publish(topic, payload)
    info.wait_for_publish(timeout=5)
    client.loop_stop()
    client.disconnect()


def mqtt_button(host, port, action, watch_topic, watch_secs):
    received = []

    def on_message(c, u, msg):
        received.append((msg.topic, msg.payload.decode()))

    sub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
    sub.on_message = on_message
    sub.connect(host, port, 60)
    sub.loop_start()
    sub.subscribe(watch_topic)

    time.sleep(1)

    pub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
    pub.connect(host, port, 60)
    pub.loop_start()
    info = pub.publish("zigbee2mqtt/Door Button", json.dumps({"action": action}))
    info.wait_for_publish(timeout=5)

    deadline = time.monotonic() + watch_secs
    while time.monotonic() < deadline:
        time.sleep(0.1)

    sub.loop_stop()
    sub.disconnect()
    pub.loop_stop()
    pub.disconnect()

    print(f"=== Published action={action!r} → {watch_topic} (watched {watch_secs}s) ===")
    if not received:
        print("(no messages received)")
        return
    for topic, payload in received:
        print(f"  {topic}  {payload}")


def main():
    ap = argparse.ArgumentParser()
    sub = ap.add_subparsers(dest="cmd", required=True)
    sub.add_parser("show", help="Show live Node-RED flow summary")
    pdiff = sub.add_parser("diff", help="Diff deployed against repo JSON")
    pdiff.add_argument("flow_path")
    pz2m = sub.add_parser("z2m-devices", help="Show paired zigbee devices and groups")
    pz2m.add_argument("z2m_yaml_path")
    pbutton = sub.add_parser("mqtt-button", help="Publish a button action and watch /set topics")
    pbutton.add_argument("--host", default="192.168.81.147")
    pbutton.add_argument("--port", type=int, default=1883)
    pbutton.add_argument("--action", default="single")
    pbutton.add_argument("--watch-topic", default="zigbee2mqtt/+/set")
    pbutton.add_argument("--watch-secs", type=float, default=4.0)
    args = ap.parse_args()
    if args.cmd == "show":
        show()
    elif args.cmd == "diff":
        sys.exit(diff(args.flow_path))
    elif args.cmd == "z2m-devices":
        z2m_devices(args.z2m_yaml_path)
    elif args.cmd == "mqtt-button":
        mqtt_button(args.host, args.port, args.action, args.watch_topic, args.watch_secs)


if __name__ == "__main__":
    main()