Add PSK Reporter MQTT listener script

Co-Authored-By: Claude <noreply@anthropic.com>

scripts/pskr_mqtt_listen.py

@@ -0,0 +1,355 @@
+#!/usr/bin/env python3
+"""
+Connect to the PSK Reporter MQTT firehose over raw TCP and print each
+received spot as a one-line JSON summary.
+
+ZERO dependencies — uses only Python stdlib (socket + struct + json).
+Implements just enough MQTT 3.1.1 to CONNECT, SUBSCRIBE, and read
+PUBLISH packets at QoS 0.
+
+Matches the subscription filter used by the prop app:
+  pskr/filter/v2/<band>/+/+/+/+/+/291/291
+
+Usage:
+  python scripts/pskr_mqtt_listen.py               # all default bands
+  python scripts/pskr_mqtt_listen.py --bands 2m     # single band
+  python scripts/pskr_mqtt_listen.py --debug        # hex dump of every packet
+"""
+
+import argparse
+import json
+import os
+import select
+import socket
+import struct
+import sys
+import time
+
+BROKER_HOST = "mqtt.pskreporter.info"
+BROKER_PORT = 1883
+KEEPALIVE_SEC = 60
+US_DXCC = 291
+
+BANDS = ["2m", "70cm", "23cm", "13cm", "9cm", "6cm", "3cm", "1.25cm"]
+
+DEBUG = False
+
+# ── minimal MQTT 3.1.1 primitives ────────────────────────────────────
+
+def _vbint(n: int) -> bytes:
+    if n == 0:
+        return b"\x00"
+    parts = []
+    while n > 0:
+        parts.append(n & 0x7F)
+        n >>= 7
+    for i in range(len(parts) - 1):
+        parts[i] |= 0x80
+    return bytes(parts)
+
+
+def _fixed_header(byte1: int, body: bytes) -> bytes:
+    return bytes([byte1]) + _vbint(len(body)) + body
+
+
+def mqtt_connect(client_id: str) -> bytes:
+    """
+    Build a CONNECT packet matching the Elixir Pskr.Mqtt.connect/2:
+      <<4::16, "MQTT", 0x04, 0x02, keepalive::16,
+        byte_size(client_id)::16, client_id::binary>>
+    """
+    cid = client_id.encode()
+    body = (
+        b"\x00\x04MQTT"             # protocol name + 2-byte length prefix
+        b"\x04"                      # protocol level (3.1.1)
+        b"\x02"                      # flags: clean session, no will, no auth
+        + struct.pack(">H", KEEPALIVE_SEC)
+        + struct.pack(">H", len(cid))
+        + cid
+    )
+    return _fixed_header(0x10, body)
+
+
+def mqtt_subscribe(packet_id: int, topics: list[tuple[str, int]]) -> bytes:
+    body = bytearray()
+    body.extend(struct.pack(">H", packet_id))
+    for topic, qos in topics:
+        t = topic.encode()
+        body.extend(struct.pack(">H", len(t)))
+        body.extend(t)
+        body.append(qos)
+    return _fixed_header(0x82, bytes(body))
+
+
+def mqtt_pingreq() -> bytes:
+    return b"\xc0\x00"
+
+
+def mqtt_disconnect() -> bytes:
+    return b"\xe0\x00"
+
+
+# ── socket / packet helpers ──────────────────────────────────────────
+
+def _recv_exact(sock: socket.socket, n: int) -> bytes | None:
+    """Read exactly n bytes. Returns None on EOF (not on timeout — callers
+    must use select to ensure data is ready before calling)."""
+    buf = bytearray()
+    while len(buf) < n:
+        try:
+            chunk = sock.recv(n - len(buf))
+        except (ConnectionError, OSError):
+            return None
+        if not chunk:
+            return None
+        buf.extend(chunk)
+    return bytes(buf)
+
+
+class MqttProtocolError(Exception):
+    pass
+
+
+def read_packet(sock: socket.socket) -> tuple[int, bytes]:
+    """
+    Read one complete MQTT packet. Blocks until a full packet arrives.
+    Raises MqttProtocolError on parse failure. Returns None on clean EOF.
+    """
+    b0 = _recv_exact(sock, 1)
+    if b0 is None:
+        return None
+    type_flags = b0[0]
+
+    # remaining length (variable-byte integer)
+    remaining = 0
+    shift = 0
+    while True:
+        b = _recv_exact(sock, 1)
+        if b is None:
+            return None
+        byte = b[0]
+        remaining |= (byte & 0x7F) << shift
+        if not (byte & 0x80):
+            break
+        shift += 7
+        if shift > 21:
+            raise MqttProtocolError("varint too long")
+
+    body = _recv_exact(sock, remaining) if remaining > 0 else b""
+    if body is None:
+        return None
+    return type_flags, body
+
+
+# ── helpers ──────────────────────────────────────────────────────────
+
+def _s(v, default="?"):
+    if v:
+        return str(v)
+    return default
+
+
+def fmt_ts(unix_val):
+    if unix_val is None:
+        return "?"
+    try:
+        return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(int(unix_val)))
+    except (OverflowError, ValueError, OSError):
+        return str(unix_val)
+
+
+# ── main ─────────────────────────────────────────────────────────────
+
+def main():
+    global DEBUG
+
+    parser = argparse.ArgumentParser(description="Listen to PSK Reporter MQTT spots (no-deps raw-TCP)")
+    parser.add_argument("--bands", nargs="+", default=BANDS,
+                        help=f"Bands to subscribe (default: {' '.join(BANDS)})")
+    parser.add_argument("--debug", action="store_true", help="Hex dump every packet")
+    args = parser.parse_args()
+
+    DEBUG = args.debug
+
+    bands = args.bands
+    topics = [(f"pskr/filter/v2/{b}/+/+/+/+/+/{US_DXCC}/{US_DXCC}", 0) for b in bands]
+
+    suffix = os.urandom(4).hex()[:8]
+    hostname = socket.gethostname().split(".")[0].replace(".", "-")
+    client_id = f"microwaveprop-debug-{hostname}-{suffix}"
+
+    # ── resolve + connect ──
+    print(f"Resolving {BROKER_HOST} …", file=sys.stderr)
+    addrs = socket.getaddrinfo(BROKER_HOST, BROKER_PORT, socket.AF_INET, socket.SOCK_STREAM)
+    ip = addrs[0][4][0]
+    print(f"Connecting to {BROKER_HOST} ({ip}):{BROKER_PORT} …", file=sys.stderr)
+    sock = socket.create_connection((ip, BROKER_PORT), timeout=10)
+    sock.setblocking(True)
+
+    # ── CONNECT ──
+    connect_pkt = mqtt_connect(client_id)
+    if DEBUG:
+        print(f"\n>> CONNECT ({len(connect_pkt)} bytes): {connect_pkt.hex()}", file=sys.stderr)
+        print(f"   client_id={client_id}", file=sys.stderr)
+    sock.sendall(connect_pkt)
+
+    result = read_packet(sock)
+    if result is None:
+        print("ERROR: no CONNACK — broker closed connection immediately", file=sys.stderr)
+        sys.exit(1)
+    tf, body = result
+    if DEBUG:
+        print(f"<< type=0x{tf:02X} body={body.hex()}", file=sys.stderr)
+
+    if tf >> 4 != 2:
+        print(f"ERROR: expected CONNACK (0x20), got type 0x{tf:02X}", file=sys.stderr)
+        sys.exit(1)
+
+    return_code = body[1]
+    session_present = bool(body[0] & 1)
+    if return_code != 0:
+        codes = {1: "unacceptable protocol version", 2: "identifier rejected",
+                 3: "server unavailable", 4: "bad user/password", 5: "not authorized"}
+        reason = codes.get(return_code, f"unknown ({return_code})")
+        print(f"ERROR: CONNACK refused: {reason}", file=sys.stderr)
+        sys.exit(1)
+    print(f"Connected! session_present={session_present}", file=sys.stderr)
+
+    # ── SUBSCRIBE ──
+    sub_pkt = mqtt_subscribe(1, topics)
+    if DEBUG:
+        print(f"\n>> SUBSCRIBE ({len(sub_pkt)} bytes): {sub_pkt[:80].hex()}...", file=sys.stderr)
+    sock.sendall(sub_pkt)
+
+    result = read_packet(sock)
+    if result is None:
+        print("ERROR: no SUBACK — broker closed connection", file=sys.stderr)
+        sys.exit(1)
+    tf, body = result
+    if DEBUG:
+        print(f"<< type=0x{tf:02X} body={body.hex()}", file=sys.stderr)
+
+    if tf != 0x90:
+        print(f"ERROR: expected SUBACK (0x90), got type 0x{tf:02X}", file=sys.stderr)
+        sys.exit(1)
+
+    granted = body[2:]
+    failures = sum(1 for b in granted if b == 0x80)
+    if failures == len(granted):
+        print(f"ERROR: all {len(granted)} subscriptions rejected (0x80)", file=sys.stderr)
+        sys.exit(1)
+    if failures:
+        print(f"WARNING: {failures}/{len(granted)} subscriptions rejected", file=sys.stderr)
+
+    print(f"Subscribed to {len(bands)} band(s): {', '.join(bands)}", file=sys.stderr)
+    print("Listening for spots…\n", file=sys.stderr)
+
+    # ── read loop ──
+    last_ping = time.monotonic()
+    count = 0
+
+    while True:
+        # Wait for data or keepalive deadline, whichever comes first.
+        # select timeout = time until next PINGREQ, but never more than
+        # KEEPALIVE_SEC seconds so we don't hang forever if the broker
+        # goes silent.
+        now = time.monotonic()
+        ping_deadline = last_ping + KEEPALIVE_SEC - 5
+        wait = max(0.1, min(ping_deadline - now, KEEPALIVE_SEC))
+
+        readable, _, errored = select.select([sock], [], [sock], wait)
+
+        if errored:
+            print("Socket error (select).", file=sys.stderr)
+            break
+
+        now = time.monotonic()
+
+        # Time to send a PINGREQ?
+        if now - last_ping >= KEEPALIVE_SEC - 5:
+            try:
+                sock.sendall(mqtt_pingreq())
+            except (ConnectionError, OSError):
+                print("Connection lost (ping send failed).", file=sys.stderr)
+                break
+            last_ping = now
+            if DEBUG:
+                print(">> PINGREQ", file=sys.stderr)
+
+        if not readable:
+            continue
+
+        # Data available — read one full packet
+        try:
+            result = read_packet(sock)
+        except MqttProtocolError as e:
+            print(f"Protocol error: {e} — disconnecting", file=sys.stderr)
+            break
+
+        if result is None:
+            print("Connection closed by broker.", file=sys.stderr)
+            break
+
+        tf_pkt, body_pkt = result
+
+        if DEBUG:
+            ptype = tf_pkt >> 4
+            labels = {1: "CONNECT", 2: "CONNACK", 3: "PUBLISH", 4: "PUBACK",
+                      8: "SUBSCRIBE", 9: "SUBACK", 12: "PINGREQ", 13: "PINGRESP", 14: "DISCONNECT"}
+            print(f"<< {labels.get(ptype, '?')} (0x{tf_pkt:02X}) len={len(body_pkt)}", file=sys.stderr)
+
+        # PUBLISH (QoS 0)
+        if (tf_pkt & 0xF0) == 0x30:
+            qos = (tf_pkt >> 1) & 0x03
+            if qos != 0:
+                if DEBUG:
+                    print(f"  skipping QoS {qos}", file=sys.stderr)
+                continue
+
+            pos = 0
+            topic_len = struct.unpack(">H", body_pkt[pos:pos+2])[0]
+            pos += 2
+            topic = body_pkt[pos:pos+topic_len].decode(errors="replace")
+            pos += topic_len
+            payload_bytes = body_pkt[pos:]
+
+            try:
+                spot = json.loads(payload_bytes)
+            except json.JSONDecodeError:
+                print(json.dumps({"error": "bad_json", "raw": payload_bytes.decode(errors="replace")[:200]}))
+                continue
+
+            ts = spot.get("t_tx") or spot.get("t")
+            out = {
+                "time":    fmt_ts(ts),
+                "band":    _s(spot.get("b")),
+                "freq_hz": spot.get("f"),
+                "mode":    _s(spot.get("md")),
+                "snr_db":  spot.get("rp"),
+                "tx_call": _s(spot.get("sc")),
+                "tx_grid": _s(spot.get("sl")),
+                "rx_call": _s(spot.get("rc")),
+                "rx_grid": _s(spot.get("rl")),
+                "topic":   topic,
+            }
+            print(json.dumps(out), flush=True)
+            count += 1
+
+        # PINGRESP
+        elif tf_pkt == 0xD0:
+            pass
+
+        # SUBACK (can arrive outside handshake)
+        elif tf_pkt == 0x90:
+            pass
+
+        else:
+            if DEBUG:
+                print(f"  unhandled packet", file=sys.stderr)
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except KeyboardInterrupt:
+        print("\nDisconnecting…", file=sys.stderr)