diff --git a/scripts/csi-udp-relay.py b/scripts/csi-udp-relay.py new file mode 100644 index 00000000..0c481749 --- /dev/null +++ b/scripts/csi-udp-relay.py @@ -0,0 +1,66 @@ +#!/usr/bin/env python3 +"""Firewall-free CSI UDP relay for local Windows ESP32 testing. + +On Windows, a freshly-built binary (e.g. `wifi-densepose calibrate-serve`) is +blocked from receiving inbound LAN UDP by Windows Defender Firewall unless an +admin adds an allow rule. `python.exe` is typically already allowed. This relay +binds the public CSI port, receives the ESP32's frames, and forwards each +datagram verbatim to a loopback port where the calibration server listens +(loopback is exempt from the inbound firewall). No admin required. + +Usage: + python scripts/csi-udp-relay.py --listen 5005 --forward 5006 + +Then run the calibration server on the loopback port: + wifi-densepose calibrate-serve --udp-bind 127.0.0.1 --udp-port 5006 + +Frames are passed through byte-for-byte; the relay never parses or mutates them. +""" +import argparse +import socket +import time + + +def main() -> None: + ap = argparse.ArgumentParser(description="Forward ESP32 CSI UDP to a loopback port (no admin).") + ap.add_argument("--listen", type=int, default=5005, help="public UDP port the ESP32 streams to") + ap.add_argument("--listen-host", default="0.0.0.0", help="bind address for the public port") + ap.add_argument("--forward", type=int, default=5006, help="loopback port the calibration server listens on") + ap.add_argument("--forward-host", default="127.0.0.1", help="loopback host to forward to") + ap.add_argument("--quiet", action="store_true", help="suppress the periodic stats line") + args = ap.parse_args() + + rx = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + rx.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + rx.bind((args.listen_host, args.listen)) + rx.settimeout(1.0) + tx = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + dst = (args.forward_host, args.forward) + + print(f"[relay] {args.listen_host}:{args.listen} -> {dst[0]}:{dst[1]} (Ctrl-C to stop)") + count = 0 + last_report = time.time() + last_src = None + try: + while True: + try: + data, src = rx.recvfrom(2048) + except socket.timeout: + data = None + if data: + tx.sendto(data, dst) + count += 1 + last_src = src + now = time.time() + if not args.quiet and now - last_report >= 5.0: + print(f"[relay] forwarded {count} frames (last src={last_src})") + last_report = now + except KeyboardInterrupt: + print(f"\n[relay] stopped after {count} frames") + finally: + rx.close() + tx.close() + + +if __name__ == "__main__": + main()