audio_rust_bridge.py
Rama main · réplica de lectura del repositorio original.
"""
audio_rust_bridge.py — Python ↔ Rust ham_audio bridge
"""
import asyncio, json, subprocess, pathlib, struct, os, time
from config import VERBOSE
CTRL_PORT = int(os.environ.get("HAM_CTRL_PORT", 9400))
WS_PORT = int(os.environ.get("HAM_WS_PORT", 9401))
DEEPCW_PORT = int(os.environ.get("HAM_DEEPCW_PORT", 9402))
EXE_PATHS = [
pathlib.Path(__file__).parent / "ham_audio.exe",
pathlib.Path(__file__).parent / "ham_audio" / "target" / "release" / "ham_audio.exe",
]
# ── Windows Job Object: guarantee ham_audio.exe dies with us ────────────────
# LIVE BUG (2026-08-21): RustAudioBridge.stop() sends {"cmd":"Shutdown"} and
# terminate()s the child - but nothing in the whole codebase ever calls
# stop() on app exit (no atexit/signal handler anywhere). Closing HAMCTRL
# via the console window's X button, Ctrl+C, or a crash all leave
# ham_audio.exe running as an orphan. After enough restart cycles, orphaned
# instances pile up - each new launch panics trying to re-bind the ctrl
# port ("address already in use") because an old orphan still holds it,
# and the app ends up talking to a STALE ham_audio.exe from a previous
# session/build instead of the current one - confirmed live: RX audio from
# a fresh install/rebuild was actually being served (or not served) by an
# old orphaned process, invisible to the Python-side log entirely (it just
# connects to whatever's listening on the port, doesn't know it's stale).
#
# Fix: a Windows Job Object with JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE - the
# OS itself kills every process assigned to the job the moment the job
# handle closes, which Windows does automatically when THIS process exits
# for ANY reason (clean shutdown, Ctrl+C, console X button, or a hard
# crash) - no explicit cleanup code path to forget to call. This is the
# standard pattern browsers/IDEs use for exactly this problem; a plain
# atexit handler would only cover the clean-exit case, which isn't the one
# that actually bit us here.
_job_handle = None
def _get_or_create_job_object():
"""Returns a Job Object handle with kill-on-close set, creating it on
first use. Windows-only; returns None on any failure (best-effort -
if this doesn't work, behavior just falls back to the pre-existing
"may orphan on abrupt exit" state, not a functional regression)."""
global _job_handle
if os.name != 'nt':
return None
if _job_handle is not None:
return _job_handle
try:
import ctypes
from ctypes import wintypes
class _JOBOBJECT_BASIC_LIMIT_INFORMATION(ctypes.Structure):
_fields_ = [
("PerProcessUserTimeLimit", ctypes.c_int64),
("PerJobUserTimeLimit", ctypes.c_int64),
("LimitFlags", wintypes.DWORD),
("MinimumWorkingSetSize", ctypes.c_size_t),
("MaximumWorkingSetSize", ctypes.c_size_t),
("ActiveProcessLimit", wintypes.DWORD),
("Affinity", ctypes.c_size_t),
("PriorityClass", wintypes.DWORD),
("SchedulingClass", wintypes.DWORD),
]
class _IO_COUNTERS(ctypes.Structure):
_fields_ = [
("ReadOperationCount", ctypes.c_uint64),
("WriteOperationCount", ctypes.c_uint64),
("OtherOperationCount", ctypes.c_uint64),
("ReadTransferCount", ctypes.c_uint64),
("WriteTransferCount", ctypes.c_uint64),
("OtherTransferCount", ctypes.c_uint64),
]
class _JOBOBJECT_EXTENDED_LIMIT_INFORMATION(ctypes.Structure):
_fields_ = [
("BasicLimitInformation", _JOBOBJECT_BASIC_LIMIT_INFORMATION),
("IoInfo", _IO_COUNTERS),
("ProcessMemoryLimit", ctypes.c_size_t),
("JobMemoryLimit", ctypes.c_size_t),
("PeakProcessMemoryUsed", ctypes.c_size_t),
("PeakJobMemoryUsed", ctypes.c_size_t),
]
JobObjectExtendedLimitInformation = 9
JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE = 0x00002000
kernel32 = ctypes.windll.kernel32
h = kernel32.CreateJobObjectW(None, None)
if not h:
return None
info = _JOBOBJECT_EXTENDED_LIMIT_INFORMATION()
info.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE
ok = kernel32.SetInformationJobObject(
h, JobObjectExtendedLimitInformation,
ctypes.byref(info), ctypes.sizeof(info)
)
if not ok:
kernel32.CloseHandle(h)
return None
_job_handle = h
return h
except Exception as e:
print(f"[audio_bridge] Job Object setup failed (non-fatal, "
f"ham_audio.exe may orphan on abrupt exit): {e}", flush=True)
return None
def _assign_to_kill_on_close_job(proc: subprocess.Popen):
"""Assigns `proc` to the shared kill-on-close Job Object, so Windows
kills it automatically when THIS process exits. Best-effort - failure
just means the pre-existing orphan risk remains for that process."""
job = _get_or_create_job_object()
if not job:
return
try:
import ctypes
kernel32 = ctypes.windll.kernel32
# subprocess.Popen on Windows exposes the raw process HANDLE via
# ._handle (an int) - documented CPython implementation detail on
# this platform, stable since it backs Popen.pid/wait() themselves.
h_process = int(proc._handle)
if not kernel32.AssignProcessToJobObject(job, h_process):
print("[audio_bridge] AssignProcessToJobObject failed "
"(non-fatal)", flush=True)
except Exception as e:
print(f"[audio_bridge] AssignProcessToJobObject error (non-fatal): {e}", flush=True)
class RustAudioBridge:
def __init__(self):
self._proc = None
self._ctrl_r = None
self._ctrl_w = None
self._connected = False
self._hub = None
self._cfg = {}
self._ft8_receiver = None # Ft8RustReceiver instance
self._stopping = False # set by stop() so the watchdog doesn't "helpfully" relaunch a deliberate shutdown
self._watchdog_task = None
self._restart_count = 0 # consecutive QUICK (<5s) unexpected exits — crash-loop guard
self._last_start_at = 0.0
self._slow_retry = False # True once the fast guard gave up and we fell back to a patient retry
async def start(self, hub=None, cfg: dict = None):
self._hub = hub
if cfg is not None:
self._cfg = cfg
self._connected = False
self._ctrl_r = None
self._ctrl_w = None
exe = next((p for p in EXE_PATHS if p.exists()), None)
if not exe:
print("[audio_bridge] ham_audio.exe not found", flush=True)
return False
# Stop the old process
if self._proc:
try: self._proc.terminate()
except Exception: pass
await asyncio.sleep(0.5)
env = os.environ.copy()
# Set the audio device via env vars
rx_dev = self._cfg.get("rxDevice", "")
tx_dev = self._cfg.get("txDevice", "")
if rx_dev: env["HAM_RX_DEVICE"] = rx_dev
if tx_dev: env["HAM_TX_DEVICE"] = tx_dev
env["HAM_BITRATE"] = str(self._cfg.get("bitrate", 24000))
# ── SSL CERTIFICATE PATH FOR RUST (WSS 9443) ──────────────────────
# PROBLEM: Rust couldn't find the cert -> "SSL cert not found" -> WSS
# off -> no audio. Python KNOWS the cert path -> we pass it to Rust
# via HAM_SSL_CERT/HAM_SSL_KEY (Rust reads these env vars — confirmed
# in config.rs). KEY POINT: we look for the cert in the DATA
# DIRECTORY (APPDATA), NOT a relative path — because in the EXE the
# working directory is _MEIxxxx and a relative tunnel_config.json
# can't be found there. We use DATA from config.py (the same source
# of truth as the rest of the app).
try:
import json as _json
import pathlib as _pl
_cp = _kp = ""
# 1. Data directory from config.py (APPDATA\HAMCTRL) — deterministic.
try:
from config import DATA as _DATA
_data_dir = _pl.Path(_DATA)
except Exception:
_data_dir = _pl.Path(__file__).parent
# 2. Try tunnel_config.json IN THE DATA DIRECTORY (not relatively).
for _cand in (_data_dir / "tunnel_config.json",
_pl.Path("tunnel_config.json")):
if _cand.exists():
try:
_t = _json.loads(_cand.read_text())
_cp = _t.get("certPath", "") or _cp
_kp = _t.get("keyPath", "") or _kp
if _cp and _kp:
break
except Exception:
pass
# 3. Fallback: the standard Let's Encrypt location in the data directory.
# letsencrypt\config\live\<domain>\{fullchain,privkey}.pem
if not (_cp and _kp and _pl.Path(_cp).exists() and _pl.Path(_kp).exists()):
_le = _data_dir / "letsencrypt" / "config" / "live"
if _le.exists():
for _dom in _le.iterdir():
_fc = _dom / "fullchain.pem"
_pk = _dom / "privkey.pem"
if _fc.exists() and _pk.exists():
_cp, _kp = str(_fc), str(_pk)
break
# Pass it to Rust if real files were found.
if _cp and _kp and _pl.Path(_cp).exists() and _pl.Path(_kp).exists():
env["HAM_SSL_CERT"] = str(_cp)
env["HAM_SSL_KEY"] = str(_kp)
print(f"[audio_bridge] SSL cert for Rust: {_cp}", flush=True)
else:
print(f"[audio_bridge] WARNING: could not find a cert for Rust "
f"(looked in {_data_dir}) — WSS will be OFF, no audio. "
f"Check whether the cert is in letsencrypt\\config\\live\\", flush=True)
except Exception as _e:
print(f"[audio_bridge] WARNING: could not read the cert for Rust: {_e}", flush=True)
# FIX: ham_audio.exe used to ALWAYS get its own visible console
# (CREATE_NEW_CONSOLE), added at the time to see its own "[build]
# ..." version marker separately from the Python log while
# debugging the Rust decoder. For a normal release this means TWO
# windows pop up (HAM RADIO CTRL's own console + this one) for
# every user, most of whom never need Rust's own console at all.
# Tied to the same VERBOSE flag as the rest of the reduced-noise
# logging: hidden by default (CREATE_NO_WINDOW - the process still
# runs exactly the same, just without a visible window), shown
# again with HAM_VERBOSE=1 / VERBOSE=1 in .env for anyone who
# actually wants to watch ham_audio's own console output live.
if os.name == 'nt':
_creationflags = subprocess.CREATE_NEW_CONSOLE if VERBOSE else subprocess.CREATE_NO_WINDOW
else:
_creationflags = 0
# CREATE_NO_WINDOW gives the child no console at all, so its
# stdio handles are invalid unless explicitly redirected here -
# send them to DEVNULL rather than risk ham_audio.exe erroring out
# on a print() with nowhere to go. Not needed (and not applied)
# in verbose mode, where it gets a real console of its own.
_stdio = {} if VERBOSE else {"stdout": subprocess.DEVNULL, "stderr": subprocess.DEVNULL}
self._proc = subprocess.Popen(
[str(exe)], env=env,
cwd=str(exe.parent),
creationflags=_creationflags,
**_stdio,
)
self._stopping = False
self._last_start_at = time.time()
# Kill-on-close Job Object - see the big comment above EXE_PATHS.
# Guarantees this ham_audio.exe dies with us even if we exit
# abruptly (console X button, crash) and never reach stop().
_assign_to_kill_on_close_job(self._proc)
# Log the EXE's modification timestamp on the Python side -
# ham_audio.exe runs as a separate process in its OWN console
# (CREATE_NEW_CONSOLE), so its own "[build] ..." print (main.rs)
# does NOT end up in the same log the operator actually pastes/
# checks. Without this there's no way to tell from this log whether
# a `cargo build` actually produced the binary under test, or
# whether it's stale from before the rebuild.
import datetime as _dt
_mtime = _dt.datetime.fromtimestamp(exe.stat().st_mtime).strftime("%Y-%m-%d %H:%M:%S")
print(f"[audio_bridge] Started ham_audio.exe PID={self._proc.pid} "
f"(cwd={exe.parent}, file modified={_mtime})", flush=True)
for _ in range(20):
await asyncio.sleep(0.3)
try:
reader, writer = await asyncio.open_connection("127.0.0.1", CTRL_PORT)
self._ctrl_r = reader
self._ctrl_w = writer
self._connected = True
break
except Exception:
pass
if not self._connected:
print("[audio_bridge] Could not connect to ham_audio", flush=True)
return False
print("[audio_bridge] Connected to ham_audio", flush=True)
# Auto-start RX
rx_dev = self._cfg.get("rxDevice", "")
bitrate = int(self._cfg.get("bitrate", 24000))
vol = float(self._cfg.get("volume", 1.0))
if rx_dev:
await self._send_ctrl({"cmd": "SetRxDevice", "name": rx_dev})
await self._send_ctrl({"cmd": "SetBitrate", "bps": bitrate})
await self._send_ctrl({"cmd": "SetVolume", "vol": vol})
print(f"[audio_bridge] RX auto-start: dev='{rx_dev}' bitrate={bitrate}", flush=True)
# Start the FT8 receiver (Python listens, Rust connects)
from ft8_rust_receiver import Ft8RustReceiver
self._ft8_receiver = Ft8RustReceiver(port=9444)
await self._ft8_receiver.start()
# Watchdog: restarts ham_audio.exe if it exits on its own (crash, or
# a deliberate self-exit — see the Rust-side comment in audio.rs
# about giving up on in-process WASAPI recovery after repeated
# failed reopen attempts and exiting instead, so a full fresh
# process/COM-enumerator state can pick the device up cleanly).
# FIX: reported live — a full HAMCTRL restart always restored RX
# audio after the radio's USB Audio CODEC disappeared/reappeared
# (power loss), but the in-process WASAPI reopen path alone did
# not, even though it kept sending frames (silent ones). Nothing
# anywhere previously restarted ham_audio.exe if IT exited - one
# was needed to make a Rust-side "just exit and let Python redo
# it fresh" strategy safe to use at all.
if self._watchdog_task is None or self._watchdog_task.done():
self._watchdog_task = asyncio.create_task(self._watchdog())
return True
async def _watchdog(self):
# NOTE: this loop deliberately never returns after restarting (only
# on stop()/crash-loop-giveup) and instead keeps monitoring
# self._proc as start() replaces it - see the guard in start()
# ("if self._watchdog_task is None or ...done()"): while THIS
# coroutine is still suspended inside its own `await self.start(...)`
# call below, that guard correctly sees the running task as "not
# done" and skips spawning a duplicate - which only works because
# this loop keeps going afterward instead of returning. Returning
# here would silently leave the freshly-restarted process
# unmonitored until some unrelated future start() call happened to
# run again.
while True:
await asyncio.sleep(30.0 if self._slow_retry else 2.0)
if self._stopping:
return
proc = self._proc
if proc is None or proc.poll() is None:
continue # still running (or not started yet)
ran_for = time.time() - self._last_start_at
if ran_for < 5.0:
self._restart_count += 1
else:
self._restart_count = 0
self._slow_retry = False # a run that actually stuck - back to normal fast monitoring
if self._restart_count >= 5 and not self._slow_retry:
# FIX (2026-09-21, live report: "po zaniku pradu audio nie
# wstaje, radio sie budzi ale nie ma audio na www, konsola
# mowi ze jest rust zaladowany" - a 24/7 unattended station).
# This guard used to `return` here, PERMANENTLY stopping the
# watchdog until a human manually restarted the whole
# server. It exists to stop hammering a genuinely-broken
# device, but 5 quick strikes (<5s each, ~25s total) is
# nowhere near enough headroom for a real power outage: the
# radio's own USB Audio CODEC (audio.rs run_rx_thread) can
# legitimately take longer than that to re-enumerate after
# power returns, and a full PC reboot means Windows itself
# may still be finishing its own USB stack init. Giving up
# forever here turned a recoverable, temporary hardware
# delay into a permanent outage - exactly backwards for a
# station meant to run 24/7 unattended. Now it falls back to
# a much slower cadence (every 30s, see the sleep above)
# instead of stopping - patient enough not to hammer truly
# dead hardware, but it always recovers on its own once the
# device is actually ready, with no manual restart required.
print(f"[audio_bridge] ham_audio.exe exited {self._restart_count} times in a row "
f"within 5s of starting - switching to a slow retry (every 30s) instead of "
f"giving up; will recover automatically once the audio device is ready",
flush=True)
self._slow_retry = True
print(f"[audio_bridge] ham_audio.exe exited unexpectedly (ran {ran_for:.1f}s) "
f"— restarting automatically"
+ (" (slow retry mode)" if self._slow_retry else ""), flush=True)
await self.start(self._hub, self._cfg)
async def stop(self):
self._stopping = True
try: await self._send_ctrl({"cmd": "Shutdown"})
except Exception: pass
if self._proc: self._proc.terminate()
self._connected = False
# ── Control commands ─────────────────────────────────────────────────────
async def _send_ctrl(self, cmd: dict) -> dict:
if not self._ctrl_w: return {"error": "not connected"}
self._ctrl_w.write((json.dumps(cmd) + "\n").encode())
await self._ctrl_w.drain()
try:
resp = await asyncio.wait_for(self._ctrl_r.readline(), timeout=2.0)
return json.loads(resp.decode())
except Exception as e:
return {"error": str(e)}
async def get_status(self) -> dict:
return await self._send_ctrl({"cmd": "GetStatus"})
async def list_devices(self) -> list:
# OBSERVED IN PRODUCTION: ListDevices can be unstable — it sometimes
# returns an empty list (WASAPI/cpal momentarily busy, e.g. during an
# RX device hot-swap) even though the cards genuinely exist.
# Confirmed via frontend logs: the same call moments later returns
# the correct 3+ cards. Instead of returning an empty list (which
# the frontend interprets as "no cards" and silently loses the
# saved card selection in the UI even though config.json has it
# correctly), we retry 2 more times with a short delay before
# concluding the cards genuinely aren't there.
for attempt in range(3):
r = await self._send_ctrl({"cmd": "ListDevices"})
if isinstance(r, list) and r:
return r
if attempt < 2:
await asyncio.sleep(0.3)
return r if isinstance(r, list) else []
async def set_rx_device(self, name: str):
return await self._send_ctrl({"cmd": "SetRxDevice", "name": name})
async def set_tx_device(self, name: str):
return await self._send_ctrl({"cmd": "SetTxDevice", "name": name})
async def set_volume(self, vol: float):
return await self._send_ctrl({"cmd": "SetVolume", "vol": vol})
# ── FT8 decode (from Rust) ───────────────────────────────────────────────
async def ft8_enable_rx(self, enabled: bool, mode: str = "FT8"):
"""Enable/disable FT8 RX decode in Rust ham_audio.exe."""
await self._send_ctrl({"cmd": "SetFt8Mode", "mode": mode})
await self._send_ctrl({"cmd": "SetFt8Rx", "enabled": enabled})
if self._ft8_receiver:
self._ft8_receiver.enable(enabled)
self._ft8_receiver.set_mode(mode)
print(f"[audio_bridge] FT8 RX enabled={enabled} mode={mode}", flush=True)
async def set_ap_hints(self, own_call: str, partner_call: str | None, queue: list[str],
rx_freq: float | None = None):
"""Push AP (a priori) decode hints to Rust ham_audio.exe - operator's
own callsign, current QSO partner (if any), and the Call-1st queue.
Rust biases the LDPC channel LLR toward these when a normal (blind)
decode fails, to recover weak replies addressed to us. Mirrors
ft8_enable_rx's fire-and-forget SetXxx pattern - safe to call often
(e.g. every time the QSO engine's state changes), Rust just keeps
whatever was pushed most recently."""
await self._send_ctrl({
"cmd": "SetApHints",
"own_call": own_call or "",
"partner_call": partner_call,
"queue": list(queue or []),
# Czestotliwosc audio RX (marker podaza za partnerem) - AP z
# partnerem (pelne komunikaty RR73/73/raporty) dziala tylko w
# jej poblizu, patrz decode::ap / AP_RX_WIN_HZ w ham_audio.
"rx_freq": float(rx_freq) if rx_freq is not None else None,
})
async def ft8_get_decode(self) -> dict | None:
"""Fetch one FT8 decode result from the Rust queue (timeout=0.1s)."""
if not self._ft8_receiver:
return None
return await self._ft8_receiver.get_decode()
@property
def ft8_rx_enabled(self) -> bool:
return self._ft8_receiver.enabled if self._ft8_receiver else False
rust_audio = RustAudioBridge()