rigcat.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""
rigcat.py — radio control via Hamlib (rigctld) over TCP.
On Replit (no rigctld) -> simulation.
NOTE: Hamlib is NOT a Python library. It's a separate program,
`rigctld(.exe)`. The server launches it itself and talks to it over TCP
(default 127.0.0.1:4532). Windows install: download Hamlib
(hamlib-w64-*.zip) from hamlib.github.io, extract it e.g. to C:\\hamlib.
The server looks for rigctld.exe in:
- HAMLIB_PATH from .env
- C:\\Program Files\\hamlib-w64-4.7.1\\bin\\rigctld.exe
- C:\\hamlib\\bin\\rigctld.exe
- PATH (where rigctld)
"""
import asyncio, re, sys, socket, subprocess, threading
from pathlib import Path
from config import HAMLIB, RIGCTLD_PORT, RIGCTLD_LOG, ENV
class RigCAT:
def __init__(self):
self.freq = 14074000
self.mode = "USB"
self.bw = 2400
self.ptt = False
self.split = False
self.freq_b = 14074000
self.s_meter = 0.0
# Icom-CI-V-only concepts with no Hamlib-generic equivalent here -
# kept as inert defaults purely so the shared status-broadcast dict
# in webapp.py (which reads these unconditionally for both driver
# types) doesn't crash with AttributeError on a non-CI-V rig.
self.filter_num = 1
self.preamp = 0
self.attenuator = False
self.tuner = False
# KEYSPD (CW keyer speed) etc. - CI-V-populated cache read by the
# CW-send code path regardless of driver (self.rig.level_values.get(...)).
self.level_values = {}
# DTR/RTS CW keyer state (see configure_keyer/send_cw_dtr_rts below)
self._keyer_port = ''
self._keyer_line = 'DTR'
self._keyer_ser = None
self._keyer_stop = threading.Event()
self._keyer_th = None
self.connected = False
self.sim = True
self._proc = None
self._logf = None
self._last_cmd = None # argv of the last successful rigctld spawn - see _respawn_rigctld
self._lock = asyncio.Lock()
self.last_err = ""
self.last_msg = ""
self.rigctld_path = ""
self.rigctld_found = False
async def _cmd_once(self, cmd: str, timeout: float) -> str:
r, w = await asyncio.wait_for(
asyncio.open_connection("127.0.0.1", RIGCTLD_PORT), timeout=3.0)
try:
w.write((cmd + "\n").encode())
await w.drain()
buf = ""
while "RPRT" not in buf:
# 1st read: full timeout; subsequent ones (once we have
# data): short, so we don't wait full seconds when
# rigctld doesn't send RPRT.
t = timeout if not buf else 0.2
try:
chunk = await asyncio.wait_for(r.read(512), timeout=t)
except asyncio.TimeoutError:
if buf.strip():
break # we have data without RPRT — good enough
raise
if not chunk: break
buf += chunk.decode(errors="replace")
if "RPRT" in buf:
data = buf[:buf.index("RPRT")].strip()
m = re.search(r"RPRT\s+(-?\d+)", buf)
code = int(m.group(1)) if m else 0
if code != 0:
raise IOError(f"RPRT {code}")
else:
# only accept a complete response (with a line ending)
if "\n" not in buf:
# NOTE: kept in Polish - this message can end up in
# self.last_err, which is returned to the frontend UI
# (see connect()'s except block), not just logged.
raise asyncio.TimeoutError("niekompletna odpowiedz bez RPRT")
data = buf.strip()
return data
finally:
try: w.close()
except: pass
async def _cmd(self, cmd: str, timeout=4.0) -> str:
async with self._lock:
try:
return await self._cmd_once(cmd, timeout)
except (ConnectionRefusedError, ConnectionResetError, BrokenPipeError, OSError) as e:
# rigctld died mid-session (crashed, or was killed
# externally) - live-seen 2026-09-02: after ONE such
# failure, every subsequent slider command failed the same
# way (WinError 1225 "connection refused") until the
# operator manually clicked Connect again, even though
# freq/mode kept working right up to that point (they go
# through this same _cmd(), so rigctld really was gone, not
# just this one command). Since _lock already serializes
# every _cmd() call, only the FIRST caller to hit a dead
# port pays for the respawn - everyone queued behind it on
# _lock simply gets a live rigctld by the time their turn
# comes.
if not await self._respawn_rigctld():
raise
return await self._cmd_once(cmd, timeout)
async def _respawn_rigctld(self) -> bool:
"""Re-launch rigctld.exe with the exact argv from the last
successful connect() (self._last_cmd), so a died/crashed rigctld
recovers on its own instead of leaving the radio uncontrollable
until a manual reconnect. Best-effort: returns False (caller then
raises the original error) if there's no known command to relaunch
or the relaunch itself fails."""
if not getattr(self, "_last_cmd", None):
return False
print("[rig] rigctld nie odpowiada (padl?) - probuje ponownego startu "
"z ostatnia znana konfiguracja")
if self._proc:
try: self._proc.terminate()
except: pass
self._proc = None
try:
self._logf = open(RIGCTLD_LOG, "wb")
except Exception:
self._logf = None
try:
self._proc = subprocess.Popen(
self._last_cmd,
stdout=(self._logf or subprocess.DEVNULL),
stderr=(self._logf or subprocess.PIPE))
await asyncio.sleep(3.5)
if self._proc.poll() is not None:
print(f"[rig] restart rigctld nie powiodl sie: {self._rigctld_log_tail()}")
return False
print(f"[rig] rigctld wystartowal ponownie na porcie {RIGCTLD_PORT}")
return True
except Exception as e:
print(f"[rig] restart rigctld error: {e}")
return False
async def get_freq(self) -> int:
if self.sim: return self.freq
try:
r = await self._cmd("f")
except Exception as e:
print(f"[rig] get_freq ERROR: {type(e).__name__}: {e}")
raise
# rigctld 'f' returns just the Hz number on its own line — first
# look for a full number-only line, then fall back to any large
# token (resilience).
for l in (ln.strip() for ln in r.splitlines()):
if re.fullmatch(r"\d{6,}", l):
return int(l)
return await self._parse_freq_response(r)
async def get_freq_live(self, timeout: float = 0.3):
"""get_freq() here already queries rigctld live every call (no
optimistic local cache like civ.py's self.freq) - alias for
call-signature compatibility with civ.py's verify-after-retune
path (webapp.py's ft8_qsy handler)."""
return await self.get_freq()
async def _parse_freq_response(self, r: str):
for tok in re.findall(r"\d+", r):
if int(tok) > 100000:
return int(tok)
print(f"[rig] get_freq: could not recognize a frequency in the response: {r!r}")
return self.freq
async def get_mode(self) -> tuple[str, int]:
if self.sim: return self.mode, self.bw
r = await self._cmd("m")
lines = [l.strip() for l in r.strip().splitlines() if l.strip()]
mode = lines[0] if lines else "USB"
bw = 2400
for l in lines[1:]:
if l.isdigit():
bw = int(l); break
return mode, bw
async def get_smeter(self) -> float:
if self.sim: return self.s_meter
try:
r = await self._cmd("l STRENGTH")
except Exception as e:
print(f"[rig] get_smeter ERROR: {type(e).__name__}: {e}")
return self.s_meter
try:
db = float(re.findall(r"-?\d+\.?\d*", r)[0])
except Exception:
return self.s_meter
# Hamlib STRENGTH = dB relative to S9 (0 dB = S9). The panel expects
# S-units: above S9 ~6 dB/S-unit, beyond that the "S9+xx dB" scale
# (val>9 -> (val-9)*10 dB).
s = 9 + db / 6.0 if db <= 0 else 9 + db / 10.0
return max(0.0, s)
async def set_freq(self, hz: int):
if self.sim: self.freq = hz; return
await self._cmd(f"F {hz}"); self.freq = hz
async def set_mode(self, m, bw=0, fil=0):
# fil: Icom CI-V filter-slot selector (FIL1/2/3), no Hamlib-generic
# equivalent - accepted for call-signature compatibility with
# civ.py's set_mode() (both drivers are called through the same
# code path in webapp.py) and otherwise ignored.
if self.sim:
self.mode = m; self.bw = bw if bw else self.bw; return
await self._cmd(f"M {m} {bw}"); self.mode = m
if bw: self.bw = bw
async def set_ptt(self, on: bool):
if self.sim: self.ptt = on; return
await self._cmd(f"T {1 if on else 0}"); self.ptt = on
# ── CW keying via DTR/RTS ────────────────────────────────────────────
# Mirrors civ.py's DTR/RTS keyer (same Morse table/timing, same
# key-down/key-up thread), with two differences forced by how Hamlib
# works here: PTT goes through set_ptt() (Hamlib 'T' command over TCP)
# instead of a raw CI-V PTT byte, and there is NO "same port as CAT"
# fallback — RigCAT talks to rigctld over TCP and never holds a raw
# serial handle to the radio, so a separate DTR/RTS port is mandatory
# (unlike civ.py, where an empty port means "reuse the CI-V serial
# connection").
_MORSE_TABLE = {
'A':'.-', 'B':'-...', 'C':'-.-.', 'D':'-..',
'E':'.', 'F':'..-.', 'G':'--.', 'H':'....',
'I':'..', 'J':'.---', 'K':'-.-', 'L':'.-..',
'M':'--', 'N':'-.', 'O':'---', 'P':'.--.',
'Q':'--.-', 'R':'.-.', 'S':'...', 'T':'-',
'U':'..-', 'V':'...-', 'W':'.--', 'X':'-..-',
'Y':'-.--', 'Z':'--..',
'0':'-----','1':'.----','2':'..---','3':'...--',
'4':'....-','5':'.....','6':'-....','7':'--...',
'8':'---..','9':'----.',
'.':'.-.-.-', ',':'--..--', '?':'..--..',
'/':'-..-.', '+':'.-.-.', '-':'-....-',
'=':'-...-', ':':'---...', '\'':'.----.',
'@':'.--.-.', '(':'-.--.', ')':'-.--.-',
' ': None, # word gap
}
def _cw_text_duration_s(self, text: str, wpm: int) -> float:
"""Exact time to send `text` in CW at `wpm`, in seconds - same
PARIS timing as civ.py's version, used by webapp.py's CW safety
timer to know when a background keying thread will finish."""
wpm = max(5, min(60, wpm))
dit = 1.200 / wpm
units = 0.0
first_char = True
for ch in text.upper():
if ch == ' ':
units += 7.0
first_char = True
continue
code = self._MORSE_TABLE.get(ch)
if not code:
continue
if not first_char:
units += 3.0
first_char = False
for i, sym in enumerate(code):
units += 3.0 if sym == '-' else 1.0
if i < len(code) - 1:
units += 1.0
return units * dit
def configure_keyer(self, port: str, line: str):
"""
Configure the DTR/RTS keyer on a dedicated serial port. Unlike
civ.py, an empty port has nothing to fall back to here — RigCAT has
no serial handle of its own — so it just leaves the keyer
unconfigured (send_cw_dtr_rts then raises a clear error instead of
silently doing nothing).
"""
import serial as _serial
self._keyer_line = (line or 'DTR').upper()
if self._keyer_ser:
try: self._keyer_ser.close()
except Exception: pass
self._keyer_ser = None
self._keyer_port = (port or '').strip()
if not self._keyer_port:
print("[rigcat] configure_keyer: no CW DTR/RTS port set - "
"CW keying will fail until one is configured (Hamlib has "
"no CI-V port to fall back to)", flush=True)
return
try:
self._keyer_ser = _serial.Serial(self._keyer_port, baudrate=9600, timeout=0.1)
# Zero DTR/RTS right away - otherwise opening the port throws a KEY DOWN.
try:
self._keyer_ser.dtr = False
self._keyer_ser.rts = False
except Exception:
pass
print(f"[rigcat] configure_keyer: opened {self._keyer_port} for {self._keyer_line} keying", flush=True)
except Exception as e:
print(f"[rigcat] configure_keyer: error opening {self._keyer_port}: {e}", flush=True)
self._keyer_ser = None
def _set_key(self, state: bool):
"""Set the DTR or RTS line (True=KEY DOWN, False=KEY UP)."""
ser = self._keyer_ser
if not ser:
return
try:
if self._keyer_line == 'RTS':
ser.rts = state
else:
ser.dtr = state
except Exception as e:
print(f"[rigcat] _set_key error: {e}", flush=True)
def _send_morse_blocking(self, text: str, wpm: int, stop_event):
"""Synchronous CW sending via DTR/RTS (called in a separate thread).
dit_ms = 1200 / wpm (standard Morse timing, PARIS word = 50 units).
PTT is handled by the async caller BEFORE this thread starts /
AFTER it ends - this function only toggles the key line.
"""
import time as _time
wpm = max(5, min(60, wpm))
dit = 1.200 / wpm
def key_down(t):
self._set_key(True)
_time.sleep(t)
self._set_key(False)
for char in text.upper():
if stop_event.is_set():
break
if char == ' ':
_time.sleep(dit * 7) # word gap
continue
code = self._MORSE_TABLE.get(char)
if not code:
continue
for i, sym in enumerate(code):
if stop_event.is_set():
break
key_down(dit * 3 if sym == '-' else dit)
if i < len(code) - 1:
_time.sleep(dit) # gap between elements
_time.sleep(dit * 3) # gap between letters
self._set_key(False) # make sure the key is released
async def send_cw_dtr_rts(self, text: str, wpm: int):
"""Send CW text via DTR/RTS keying. Returns as soon as PTT/keying is
SCHEDULED (matches civ.py's contract - webapp.py broadcasts the PTT
toast right after calling this and expects it back near-instantly,
not after the whole message finishes sending). The actual PTT-on ->
keying -> PTT-off sequence runs in a background task."""
self._keyer_stop.set()
if self._keyer_th and self._keyer_th.is_alive():
self._keyer_th.join(timeout=0.5)
self._keyer_stop.clear()
if self.sim:
print(f"[rigcat] SIM DTR/RTS: {text!r} @ {wpm} WPM", flush=True)
return
if not self._keyer_ser:
raise RuntimeError(
"Brak skonfigurowanego portu DTR/RTS do CW (ustaw osobny port "
"w ustawieniach CW - Hamlib nie ma wspolnego portu z CAT).")
asyncio.create_task(self._send_cw_dtr_rts_task(text, wpm))
async def _send_cw_dtr_rts_task(self, text: str, wpm: int):
await self.set_ptt(True)
try:
# Give the radio a moment to switch to transmit before the first dit.
await asyncio.sleep(0.25)
stop_ev = self._keyer_stop
th = threading.Thread(
target=self._send_morse_blocking,
args=(text, wpm, stop_ev),
daemon=True, name="rigcat-morse-keyer")
self._keyer_th = th
th.start()
await asyncio.to_thread(th.join)
finally:
await self.set_ptt(False)
async def stop_cw_dtr_rts(self):
"""Abort DTR/RTS sending immediately."""
self._keyer_stop.set()
self._set_key(False)
await self.set_ptt(False)
async def set_level(self, l, v):
if self.sim: return
await self._cmd(f"L {l} {v}")
async def set_split(self, on: bool):
if self.sim: self.split = on; return
await self._cmd(f"S {1 if on else 0} VFOB"); self.split = on
async def set_vfo(self, vfo: str):
"""Switch the active VFO (VFOA/VFOB) — Hamlib command 'V'."""
if self.sim: return
await self._cmd(f"V {vfo}")
async def set_func(self, func: str, on: bool):
"""Enable/disable a radio function (NB/VOX/COMP/...) — Hamlib command 'U'."""
if self.sim: return
await self._cmd(f"U {func} {1 if on else 0}")
# ── Generic-radio mappings for features webapp.py otherwise only
# calls on civ.py (CI-V) ─────────────────────────────────────────────
# Universality: a rig added later via "any Hamlib-supported model" must
# not crash just because it isn't an Icom with civ.py's exact CI-V
# command set. Where Hamlib has a real generic equivalent, use it here
# (rigctld itself returns RPRT<0 for anything a specific rig backend
# doesn't actually support - safe, non-destructive, no guessing about
# rig-specific byte sequences the way a raw CI-V command would need).
async def set_tuner(self, on: bool):
"""Antenna tuner on/off - Hamlib func TUNER, a real 1:1 generic
equivalent of civ.py's CI-V 1C 01 <00|01>."""
await self.set_func("TUNER", on)
async def set_preamp(self, level: int):
"""Preamp on/off/stage - Hamlib level PREAMP. NOTE: unlike civ.py's
Icom-specific 0/1/2 (OFF/P.AMP1/P.AMP2) index, Hamlib's PREAMP level
is a raw dB gain value whose valid steps are rig-specific (queried
via 'l PREAMP' range, which we don't do here) - passing the index
through as-is is a best-effort default, not a verified mapping for
every rig. rigctld rejects an out-of-range value harmlessly."""
if self.sim: self.preamp = level; return
await self.set_level("PREAMP", level)
self.preamp = level
async def set_attenuator(self, on: bool):
"""Attenuator on/off - Hamlib level ATT. Same caveat as set_preamp:
Hamlib expects a dB value from the rig's own supported steps: 20dB
matches civ.py's fixed Icom attenuator step (CI-V 11), used here as
a reasonable default, not a verified value for every rig."""
if self.sim: self.attenuator = on; return
await self.set_level("ATT", 20 if on else 0)
self.attenuator = on
async def vfo_swap(self):
"""Exchange VFO A/B - Hamlib VFO_OP 'G XCHG' (real generic
equivalent of civ.py's CI-V 07 B0)."""
if self.sim:
self.freq, self.freq_b = self.freq_b, self.freq
return
await self._cmd("G XCHG")
try:
self.freq = await self.get_freq()
except Exception:
pass
async def vfo_equalize(self):
"""Copy VFO A -> B - Hamlib VFO_OP 'G CPY' (real generic
equivalent of civ.py's CI-V 07 A0)."""
if self.sim:
self.freq_b = self.freq
return
await self._cmd("G CPY")
async def set_freq_b(self, hz: int):
"""Set VFO B's frequency without disturbing the active VFO.
Hamlib's basic rigctld protocol has no direct 'set this Hz on VFOB
specifically' command - switch to B, set freq, switch back to A.
Not atomic (a callsign-selecting client could see a brief VFOB->A
flicker), but civ.py's own version isn't atomic either (multiple
CI-V transactions) and this path is only used for occasional split
setup, not hot-path tuning."""
if self.sim: self.freq_b = hz; return
await self._cmd("V VFOB")
await self._cmd(f"F {hz}")
await self._cmd("V VFOA")
self.freq_b = hz
async def start_tuner_autotune(self):
"""Hamlib has no generic 'trigger an autotune cycle now' action
(TUNER is a plain on/off func, unlike civ.py's CI-V 1C 01 01+02
two-step start sequence) - raise a clear, caught-by-the-caller
error instead of silently doing nothing or guessing at a
rig-specific command that would generate unintended TX."""
raise RuntimeError(
"Autotune tunera nie jest dostepny dla tego sterownika (Hamlib) - "
"wlacz/wylacz tuner recznie (funkcja TUNER dziala), a strojenie "
"wykonaj na samym radiu.")
async def get_capabilities(self) -> dict:
"""
Returns the full structure of discovered radio capabilities:
{"actions": [...], "sliders": [...], "raw_caps": {feature_id: bool}}
- raw_caps: simple bools for rigs/features.py (freq_set, ptt, split, ...)
- actions: buttons (VFO A/B, NB/VOX/COMP/... toggles)
- sliders: adjustments with a range (RFPOWER, AF, MICGAIN, ...)
In simulation mode (self.sim) returns empty lists/dict — the admin
can configure manually when testing without hardware.
"""
if self.sim:
return {"actions": [], "sliders": [], "raw_caps": {}}
from hamlib_caps import discover_capabilities
return await discover_capabilities(RIGCTLD_PORT)
async def connect(self, cfg: dict, override: dict | None = None) -> bool:
# Pick the radio: by rigId from override, otherwise the first one.
# Merge in the form values (override), so CIV/port/model from the UI
# take effect immediately.
rigs = cfg.get("rigs") or [{}]
rig = rigs[0]
if override:
rid = override.get("rigId") or override.get("id")
if rid is not None:
rig = next((r for r in rigs if str(r.get("id")) == str(rid)), rig)
rig = {**rig, **{k: v for k, v in override.items() if v}}
model = str(rig.get("model", ENV.get("RIG1_MODEL", "3073")))
port = rig.get("port", ENV.get("RIG1_PORT", "COM3"))
speed = str(rig.get("speed", ENV.get("RIG1_SPEED", "19200")))
# Find rigctld
ham = HAMLIB
self.rigctld_found = False
if Path(ham).exists():
self.rigctld_found = True
else:
import glob as _glob
_cands = [
r"C:\Program Files\hamlib-w64-4.7.1\bin\rigctld.exe",
r"C:\hamlib\bin\rigctld.exe",
]
# Any hamlib-w64-* version in Program Files
_cands += sorted(_glob.glob(r"C:\Program Files\hamlib-*\bin\rigctld.exe"), reverse=True)
_cands += sorted(_glob.glob(r"C:\hamlib*\bin\rigctld.exe"), reverse=True)
for _c in _cands:
if Path(_c).exists(): ham = _c; self.rigctld_found = True; break
if not self.rigctld_found:
try:
_which = "where" if sys.platform == "win32" else "which"
_r = subprocess.run([_which, "rigctld"], capture_output=True, text=True, timeout=2)
if _r.returncode == 0 and _r.stdout.strip():
ham = _r.stdout.strip().splitlines()[0]
self.rigctld_found = True
except: pass
self.rigctld_path = ham
print(f"[rig] rigctld: {ham} ({'found' if self.rigctld_found else 'NOT found'})")
if not self.rigctld_found:
self.last_err = ("Hamlib (rigctld.exe) nie znaleziony. Zainstaluj Hamlib dla Windows "
"i rozpakuj do C:\\hamlib (lub Program Files), albo dodaj rigctld do PATH.")
self.last_msg = ""
self.sim = True
self.connected = False
print(f"[rig] {self.last_err}")
return False
try:
civ = rig.get("civ") or rig.get("civAddr") or ENV.get("RIG1_CIV", "")
cmd = [ham, "-m", model, "-r", port, "-s", speed,
"-t", str(RIGCTLD_PORT), "--set-conf=serial_handshake=None"]
if civ:
civ_clean = (str(civ).strip().replace("0x", "").replace("0X", "")
.replace("h", "").replace("H", ""))
# Hamlib's real config token is 'civaddr' (no underscore) -
# 'civ_addr' is silently REJECTED ("no such token"), live-
# confirmed via rigctld -vvvvv log 2026-09-01: rigctld then
# falls back to the Icom backend's own hardcoded default CI-V
# address for that model instead of the one actually
# configured on the radio. Harmless when they happen to
# match (as for this IC-746, default=0x56); silently wrong
# for any radio whose CI-V address was changed from factory
# default, or when addressing one of several rigs on a
# shared CI-V bus.
cmd += [f"--set-conf=civaddr=0x{civ_clean}"]
print(f"[rig] CIV addr: 0x{civ_clean}")
# Verbose rigctld -> rigctld.log file (diagnostics, for when the radio doesn't respond)
cmd += ["-vvvvv"]
print(f"[rig] CMD: {' '.join(cmd)}")
print(f"[rig] rigctld log: {RIGCTLD_LOG}")
# Remembered for _respawn_rigctld() - if rigctld dies mid-session
# (crash, or killed externally), _cmd() relaunches it with this
# exact argv instead of leaving every command broken until the
# operator manually reconnects from the UI.
self._last_cmd = cmd
# Clean up any of OUR OWN orphaned rigctld.exe processes first -
# e.g. surviving an earlier abrupt close/EXE update within the
# same dev/test session. An orphan holds the COM port
# exclusively, so a freshly spawned rigctld here can open the
# TCP port fine but then fails to open the (already-locked)
# serial port and exits moments later - leaving nothing on
# RIGCTLD_PORT (ConnectionRefusedError on the next _cmd() call,
# live-seen 2026-09-01 mid-CW-test). Filtered by OUR specific
# "-t {RIGCTLD_PORT}" command-line marker so an independently
# run rigctld (e.g. started by N1MM for its own use, on a
# different port) is never touched.
if sys.platform == "win32":
try:
_ps = (f"Get-CimInstance Win32_Process -Filter \"Name='rigctld.exe'\" | "
f"Where-Object {{ $_.CommandLine -like '*-t {RIGCTLD_PORT}*' }} | "
f"ForEach-Object {{ Stop-Process -Id $_.ProcessId -Force }}")
subprocess.run(["powershell", "-NoProfile", "-Command", _ps],
capture_output=True, timeout=5)
except Exception:
pass
# Is rigctld already running on this port (e.g. started by N1MM)?
_port_busy = False
try:
_t = socket.socket(); _t.settimeout(0.5)
_port_busy = _t.connect_ex(("127.0.0.1", RIGCTLD_PORT)) == 0
_t.close()
except: pass
if _port_busy:
print(f"[rig] Port {RIGCTLD_PORT} in use — connecting to the existing rigctld")
else:
try:
self._logf = open(RIGCTLD_LOG, "wb")
except Exception:
self._logf = None
self._proc = subprocess.Popen(
cmd,
stdout=(self._logf or subprocess.DEVNULL),
stderr=(self._logf or subprocess.PIPE))
await asyncio.sleep(3.5)
if self._proc.poll() is not None:
# NOTE: kept in Polish - this text can end up in
# self.last_err via the except block below, which is
# returned to the frontend UI, not just logged.
se = self._rigctld_log_tail() or "rigctld zamknal sie bez komunikatu"
raise IOError(f"rigctld zakonczyl sie: {se}")
print(f"[rig] rigctld started on port {RIGCTLD_PORT}")
self.sim = False
# The radio may need a moment — a few attempts at reading the frequency
last_exc = None
for _try in range(3):
try:
self.freq = await self.get_freq()
last_exc = None
break
except Exception as ex:
last_exc = ex
await asyncio.sleep(1.0)
if last_exc:
raise last_exc
self.mode, self.bw = await self.get_mode()
self.connected = True
self.last_err = ""
self.last_msg = (f"{rig.get('name','Radio')} model={model} {port} "
f"{speed}bd — {self.freq}Hz {self.mode}")
print(f"[rig] {self.last_msg}")
return True
except Exception as e:
tail = self._rigctld_log_tail()
base = f"{type(e).__name__}: {e}".strip()
if isinstance(e, (asyncio.TimeoutError, TimeoutError)) and not str(e):
# NOTE: kept in Polish - assigned to self.last_err below,
# which is returned to the frontend UI, not just logged.
base = ("Radio nie odpowiada (timeout). rigctld dziala, ale nie dostaje "
"odpowiedzi z radia — sprawdz: zwolnij COM (zamknij RCForb/inny program), "
"CI-V baud w radiu = 19200, model, CI-V address.")
self.last_err = base + (f"\n--- rigctld.log ---\n{tail}" if tail else "")
self.last_msg = ""
print(f"[rig] connection ERROR ({type(e).__name__}): {e}")
if tail:
print(f"[rig] rigctld.log:\n{tail}")
self.sim = True
self.connected = False
if self._proc:
try: self._proc.terminate()
except: pass
self._proc = None
if getattr(self, "_logf", None):
try: self._logf.close()
except: pass
self._logf = None
return False
def _rigctld_log_tail(self, n: int = 1800) -> str:
try:
with open(RIGCTLD_LOG, "rb") as f:
data = f.read()
return data[-n:].decode("utf-8", errors="replace").strip()
except Exception:
return ""
def close(self):
if self._proc:
try:
self._proc.terminate()
self._proc.wait(timeout=2.0)
except Exception:
try: self._proc.kill()
except: pass
self._proc = None
if getattr(self, "_logf", None):
try: self._logf.close()
except: pass
self._logf = None