wsjtx_udp.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""
wsjtx_udp.py — WSJT-X UDP receiver
=====================================
WSJT-X sends UDP packets (port 2237) with FT8/FT4/JT65 decodes.
This module listens on UDP, parses the packets, and broadcasts them over
WebSocket to all connected browser clients.
WSJT-X configuration (on the computer next to the radio):
Settings → Reporting → UDP Server: localhost (or 127.0.0.1)
Settings → Reporting → UDP port: 2237
Settings → Reporting → ✓ Accept UDP requests
Protocol: https://sourceforge.net/p/wsjt/wsjtx/ci/master/tree/NetworkMessage.hpp
"""
import asyncio
import struct
import socket
from typing import Callable, Awaitable
# ── WSJT-X packet types ──────────────────────────────────────────────────────
MSG_HEARTBEAT = 0
MSG_STATUS = 1
MSG_DECODE = 2
MSG_CLEAR = 3
MSG_QSO_LOGGED = 5
MSG_CLOSE = 6
MAGIC = 0xADBCCBDA
SCHEMA = 2
# ── Parser ─────────────────────────────────────────────────────────────────────
def _read_u8(buf, pos):
return struct.unpack_from('>B', buf, pos)[0], pos + 1
def _read_u32(buf, pos):
return struct.unpack_from('>I', buf, pos)[0], pos + 4
def _read_i32(buf, pos):
return struct.unpack_from('>i', buf, pos)[0], pos + 4
def _read_bool(buf, pos):
v, pos = _read_u8(buf, pos)
return bool(v), pos
def _read_f64(buf, pos):
return struct.unpack_from('>d', buf, pos)[0], pos + 8
def _read_str(buf, pos):
"""Qt string: 4-byte length (0xFFFFFFFF = null), then UTF-8."""
length, pos = _read_u32(buf, pos)
if length == 0xFFFFFFFF:
return '', pos
s = buf[pos:pos + length].decode('utf-8', errors='replace')
return s, pos + length
def _read_qtime(buf, pos):
"""QTime: ms since midnight."""
ms, pos = _read_u32(buf, pos)
h = ms // 3600000
m = (ms % 3600000) // 60000
s = (ms % 60000) // 1000
return f"{h:02d}{m:02d}{s:02d}", pos
def parse_packet(data: bytes) -> dict | None:
"""Parse a WSJT-X UDP packet. Returns a dict, or None on error."""
try:
if len(data) < 8:
return None
magic, pos = _read_u32(data, 0)
schema, pos = _read_u32(data, pos)
if magic != MAGIC:
return None
msg_type, pos = _read_u32(data, pos)
_id, pos = _read_str(data, pos) # WSJT-X station ID
# ── Heartbeat ─────────────────────────────────────────────────────────
if msg_type == MSG_HEARTBEAT:
max_schema, pos = _read_u32(data, pos)
version, pos = _read_str(data, pos)
revision, pos = _read_str(data, pos)
return {
"type": "wsjtx_status",
"running": True,
"id": _id,
"version": version,
"text": f"WSJT-X {version} connected",
}
# ── Status ────────────────────────────────────────────────────────────
if msg_type == MSG_STATUS:
freq, pos = _read_u32(data, pos) # Hz (int only, not f64!)
mode, pos = _read_str(data, pos)
dx_call, pos = _read_str(data, pos)
report, pos = _read_str(data, pos)
tx_mode, pos = _read_str(data, pos)
tx_enabled,pos = _read_bool(data, pos)
transmit, pos = _read_bool(data, pos)
decoding, pos = _read_bool(data, pos)
rx_df, pos = _read_u32(data, pos)
tx_df, pos = _read_u32(data, pos)
de_call, pos = _read_str(data, pos)
de_grid, pos = _read_str(data, pos)
dx_grid, pos = _read_str(data, pos)
return {
"type": "wsjtx_status",
"running": True,
"freq": freq,
"mode": mode,
"txMode": tx_mode,
"dxCall": dx_call,
"deCall": de_call,
"deGrid": de_grid,
"txEnabled": tx_enabled,
"transmit": transmit,
"decoding": decoding,
"rxDF": rx_df,
"txDF": tx_df,
"text": f"{'📡 TX' if transmit else '📻 RX'} {mode} {freq/1e6:.3f}MHz",
}
# ── Decode ────────────────────────────────────────────────────────────
if msg_type == MSG_DECODE:
is_new, pos = _read_bool(data, pos)
time_ms, pos = _read_u32(data, pos)
snr, pos = _read_i32(data, pos)
delta_t, pos = _read_f64(data, pos)
delta_f, pos = _read_u32(data, pos)
mode_str, pos = _read_str(data, pos)
message, pos = _read_str(data, pos)
low_conf, pos = _read_bool(data, pos)
h = time_ms // 3600000
m = (time_ms % 3600000) // 60000
s = (time_ms % 60000) // 1000
time_str = f"{h:02d}{m:02d}{s:02d}"
return {
"type": "wsjtx_decode",
"isNew": is_new,
"timeStr": time_str,
"snr": snr,
"deltaTime": round(delta_t, 1),
"deltaFreq": delta_f,
"mode": mode_str,
"message": message,
"lowConf": low_conf,
}
# ── Clear ─────────────────────────────────────────────────────────────
if msg_type == MSG_CLEAR:
return {"type": "wsjtx_clear"}
# ── QSO logged ────────────────────────────────────────────────────────
if msg_type == MSG_QSO_LOGGED:
# Per WSJT-X's NetworkMessage.hpp QSOLogged:
# DateTimeOff(QDateTime), Frequency(u64/u32), DXCall, DXGrid,
# TxPower, Comments, Name, DateTimeOn, OperatorCall,
# MyCall, MyGrid, ExchangeSent, ExchangeRcvd [, PropMode, Adif]
_dt_off, pos = _read_u32(data, pos) # DateTimeOff (msec) — skipped
_dt_off2, pos = _read_u32(data, pos) # DateTimeOff (utcOffset) — skipped
freq, pos = _read_u32(data, pos) # Frequency Hz
mode, pos = _read_str(data, pos) # Mode (FT8/FT4/JT65...)
dx_call, pos = _read_str(data, pos) # DXCall
dx_grid, pos = _read_str(data, pos) # DXGrid
tx_power, pos = _read_str(data, pos) # TxPower
comments, pos = _read_str(data, pos) # Comments
name, pos = _read_str(data, pos) # Name
# DateTimeOn
_dt_on, pos = _read_u32(data, pos)
_dt_on2, pos = _read_u32(data, pos)
op_call, pos = _read_str(data, pos) # OperatorCall
my_call, pos = _read_str(data, pos) # MyCall
my_grid, pos = _read_str(data, pos) # MyGrid
# ExchangeSent and ExchangeRcvd = signal reports (e.g. "-12", "+05")
rst_sent, pos = _read_str(data, pos) # ExchangeSent
rst_rcvd, pos = _read_str(data, pos) # ExchangeRcvd
return {
"type": "wsjtx_qso_logged",
"dxCall": dx_call,
"dxGrid": dx_grid,
"freq": freq,
"mode": mode,
"rstSent": rst_sent, # e.g. "-12 dB"
"rstRcvd": rst_rcvd, # e.g. "+05 dB"
"myCall": my_call,
"myGrid": my_grid,
"txPower": tx_power,
"comments": comments,
}
# ── Close ─────────────────────────────────────────────────────────────
if msg_type == MSG_CLOSE:
return {"type": "wsjtx_status", "running": False, "text": "WSJT-X disconnected"}
except Exception as e:
pass # corrupted packet — ignore
return None
# ── UDP server ────────────────────────────────────────────────────────────────
class WsjtxUdpServer:
"""Listens for UDP from WSJT-X and broadcasts over WebSocket."""
def __init__(self, broadcast_fn: Callable):
"""
broadcast_fn: async function(dict) → sends to all WS clients
"""
self.broadcast = broadcast_fn
self._transport = None
self._running = False
self._port = 2237
self.packets_rx = 0
self.decodes_rx = 0
def is_running(self) -> bool:
return self._running
async def start(self, port: int = 2237, host: str = "0.0.0.0") -> bool:
"""Start listening on UDP."""
if self._running:
await self.stop()
self._port = port
loop = asyncio.get_running_loop()
try:
transport, protocol = await loop.create_datagram_endpoint(
lambda: _UdpProtocol(self._on_packet),
local_addr=(host, port),
)
self._transport = transport
self._running = True
print(f"[wsjtx] UDP listening on {host}:{port}")
print(f"[wsjtx] WSJT-X: Settings -> Reporting -> UDP Server: localhost, Port: {port}")
return True
except OSError as e:
print(f"[wsjtx] UDP error: {e}")
if e.errno == 10048 or e.errno == 98: # port in use
print(f"[wsjtx] Port {port} in use — check whether another program is using UDP {port}")
return False
async def stop(self):
"""Stop listening."""
if self._transport:
self._transport.close()
self._transport = None
self._running = False
print("[wsjtx] UDP stopped")
def _on_packet(self, data: bytes, addr):
"""Callback from the UDP protocol — parse and broadcast."""
self.packets_rx += 1
msg = parse_packet(data)
if msg:
if msg["type"] == "wsjtx_decode":
self.decodes_rx += 1
asyncio.create_task(self.broadcast(msg))
def get_status(self) -> dict:
return {
"running": self._running,
"port": self._port,
"packets_rx": self.packets_rx,
"decodes_rx": self.decodes_rx,
}
class _UdpProtocol(asyncio.DatagramProtocol):
def __init__(self, callback):
self._cb = callback
def datagram_received(self, data, addr):
self._cb(data, addr)
def error_received(self, exc):
print(f"[wsjtx] UDP protocol error: {exc}")