diff --git a/scripts/pskr_mqtt_listen.py b/scripts/pskr_mqtt_listen.py new file mode 100755 index 00000000..751c118d --- /dev/null +++ b/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//+/+/+/+/+/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)