""" 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\\{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()