#!/usr/bin/env python3 """ civ.py — direct CI-V (Icom) driver for a radio with a scope. DIRECT mode: the server opens the COM port itself and handles SIMULTANEOUSLY: - control (frequency 0x03/0x05, mode 0x04/0x06, PTT 0x1C00, S-meter 0x15 02) - wideband SCOPE (waterfall) — CI-V 0x27 0x00 stream Used ONLY for radios with a built-in spectrum scope (IC-7300/7610/705/9100/7100). Other radios (IC-746 etc.) still go through rigctld (RigCAT) — unchanged. The (async) interface is IDENTICAL to RigCAT, so App.rig can be swapped on the fly. Model-specific parameters (CI-V address, baud rate, mode mapping, scope parameters) are loaded from rigs/civ_profiles.py based on model_id — so adding/fixing a model needs no changes to this file. No pyserial / no COM port (e.g. on Replit) -> SIMULATION mode: generates a moving spectrum + telemetry, so the panel works even without hardware. NOTE ON TRANSLATION SCOPE: self.last_err/self.last_msg reach the browser verbatim — webapp.py returns them directly in an API response body (same pattern as rigcat.py's last_err/last_msg, see webapp.py:2783-2784). Every literal string assigned to them in this file is deliberately left in Polish; only comments, docstrings, and print()/self.log() console text are translated. """ import asyncio, threading, time, math, random import numpy as np try: import serial # pyserial HAS_SERIAL = True except Exception: HAS_SERIAL = False from rigs import get_civ_profile from config import VERBOSE CTRL_ADDR = 0xE0 # adres kontrolera (PC) # ── BCD <-> numbers ──────────────────────────────────────────────────────────── def bcd_to_freq(b: bytes) -> int: """Icom freq: 5 bytes BCD little-endian (least significant digits first).""" hz = 0 mult = 1 for byte in b: hz += (byte & 0x0F) * mult; mult *= 10 hz += (byte >> 4) * mult; mult *= 10 return hz def freq_to_bcd(hz: int, n: int = 5) -> bytes: s = f"{int(hz):0{n*2}d}"[-n * 2:] # last n*2 digits rev = s[::-1] # rev[0] = units out = bytearray() for i in range(0, n * 2, 2): lo = int(rev[i]); hi = int(rev[i + 1]) out.append((hi << 4) | lo) return bytes(out) def bcd2(b: bytes) -> int: """2-byte BCD (e.g. S-meter 0000..0255) -> int.""" if len(b) < 2: return 0 return (b[0] >> 4) * 1000 + (b[0] & 0x0F) * 100 + (b[1] >> 4) * 10 + (b[1] & 0x0F) # Mapping for subcommand 27 15 (Scope span, Center/SCROLL-C mode) -> kHz. # FIX: this used to be {2500, 5000, 10000, 25000, 50000, 100000, 250000, # 500000} - a guessed "1-2.5-5" decade sequence. The IC-7300 actually # uses a "1-2-5" sequence instead (confirmed against Hamlib's ic7300.c # .spectrum_spans table, reverse-engineered from the real radio) - so # 2500/25000/250000 (2.5/25/250 kHz) don't exist on this radio at all # and got NGed on every attempt, while 5000/10000/50000/100000/500000 # happened to coincide with real values and worked. Also adds 20000 # (20kHz) and 1000000 (1MHz, the widest span) which weren't offered # before. SCOPE_SPAN_KHZ = { 5000: 5, 10000: 10, 20000: 20, 50000: 50, 100000: 100, 200000: 200, 500000: 500, 1000000: 1000, } # Exact IC-7300 filter width table for SSB/CW/RTTY (CI-V 1A 03, # idx 00-31 = 50Hz/100Hz steps, per the Instruction Manual): # idx 0-9: 50Hz steps -> 50, 100, 150, ..., 500 # idx 10-31: 100Hz steps -> 600, 700, ..., 2700 _SSB_FILTER_TABLE = {} for _i in range(0, 32): _SSB_FILTER_TABLE[_i] = (50 + _i*50) if _i <= 9 else (600 + (_i-10)*100) # idx 32-40: SSB-D range (DATA mode, USB-D/LSB-D) — wider filter up to # 3600Hz, available ONLY when the radio is in DATA mode (1A 06). civ.py # doesn't currently track the data-mode flag here, so when we encounter # idx>31 (which shouldn't happen in voice SSB mode), we linearly # interpolate 2700->3600Hz as an approximation instead of returning an # incorrectly clamped 2700Hz. for _i in range(32, 41): _SSB_FILTER_TABLE[_i] = round(2700 + (_i-31) * (3600-2700) / 9) # AM/FM (CI-V 1A 03, idx 00=200Hz .. 49=10000Hz per the docs p.19-6). # The step isn't uniform in Icom's documentation — we use the known AM # filter table for the IC-7300 (typical values from 200Hz to 10kHz in # uneven steps, roughly matching the AM receiver's bandwidth). For idx>0 we # apply linear interpolation of 200..10000Hz over 50 points (0-49) as the # best available approximation without access to the full firmware table. _AMFM_FILTER_TABLE = {i: round(200 + i * (10000-200) / 49) for i in range(0, 50)} def filter_width_hz(mode: str, idx: int, data_mode: bool = False) -> int: """Exact (SSB/CW/RTTY) or approximate (AM/FM) filter width in Hz for a given CI-V 1A 03 index, depending on the operating mode. idx 32-40 (SSB-D, up to 3600Hz) is only available when the radio is in DATA mode (data_mode=True, polled via 1A 06). In voice SSB mode (data_mode=False) the radio shouldn't return idx>31 — if it did anyway, clamp to 2700Hz (idx 31) instead of extrapolating.""" if mode in ("AM", "FM"): return _AMFM_FILTER_TABLE.get(idx, _AMFM_FILTER_TABLE[49]) if idx > 31 and not data_mode: idx = 31 return _SSB_FILTER_TABLE.get(idx, _SSB_FILTER_TABLE[40]) def smeter_to_sunit(raw: int) -> float: """Raw Icom S-meter 0..255 -> S-units (S9 = 9, S9+60dB = 15).""" if raw <= 120: return raw / 120.0 * 9.0 return 9.0 + (raw - 120) / (241 - 120) * 60.0 / 10.0 class CivRig: def __init__(self, cfg: dict, broadcast_sync=None, log=print): self.cfg = cfg or {} self.bcast = broadcast_sync or (lambda m: None) self.log = log # ── attributes compatible with RigCAT ── self.freq = 14074000 self.freq_b = 14074000 # Time and value of the last LOCAL frequency set (set_freq, e.g. # from click-to-tune). Used in _handle_frame to ignore a "stale" # 0x03 reading from the radio that may arrive right after sending # 0x05 (a delayed CI-V echo/response), so it doesn't overwrite the # newly set frequency in the UI before the radio actually updates. self._local_freq_set_at = 0.0 self._local_freq_set_val = None self.mode = "USB" # The radio's active DSP filter (FIL1/2/3, CI-V 06 ) — # selects WHICH mechanically/programmatically configured filter to # use, not a change to its width (that's set in the radio's menu). self.filter_num = 1 # Preamp (16 02): 0=OFF, 1=P.AMP1, 2=P.AMP2 self.preamp = 0 # Attenuator (11): False=OFF, True=ATT 20dB ON self.attenuator = False # Antenna Tuner (1C 01): False=OFF (bypass), True=ON (in the chain) self.tuner = False # OVF (15 07): ADC/front-end overflow indicator (read-only) self.ovf = False self.bw = 2400 self.data_mode = False # DATA mode (USB-D/LSB-D), from 1A 06 — affects the filter table self.ptt = False # ALC/PWR peak-hold for the CURRENT transmission (reset in # set_ptt() on PTT-on). Live SSB voice makes single instantaneous # readings look "jumpy"/uncorrelated (huge dynamic range between a # loud syllable and a gap between words, sampled a few tens of ms # apart) — the peak seen across the whole TX, like a real ALC # meter's needle memory, is what actually tells you whether you're # driving the radio properly. self._alc_peak = 0.0 self._alc_peak_raw = 0 self._pwr_peak = 0.0 self._pwr_peak_raw = 0 # Telemetry coalescing (2026-08-25): freq/mode/freqB/split/smeter/ # txmeter used to call self.bcast() -> broadcast_sync -> # asyncio.run_coroutine_threadsafe() DIRECTLY from _reader_loop/ # _poller_loop (real OS threads) on every single change - up to a # few dozen cross-thread loop wakeups/sec during PTT (4 TX meters # every poll cycle) or during fast VFO knob spin on the radio's own # front panel (every transceive frame in _handle_frame). This is # the SAME pattern already found and fixed for the scope stream # (see _latest_scope / webapp.py::_scope_pump) - confirmed there # that frequent run_coroutine_threadsafe calls from a non-loop # thread measurably stall the asyncio loop (lag grows linearly # with call count), which now also delays WebRTC's own RTP send # timing since it lives on the same loop. These specific message # types are coalesced into this dict (last-value-wins per key) and # drained by webapp.py::_telemetry_pump at a steady rate instead of # synchronizing with the loop on every change. Both writer threads # (_reader_loop/_poller_loop) and the pump's atomic swap-and-drain # go through _telemetry_lock - a plain dict write is GIL-atomic on # its own, but the pump's "grab old dict, iterate its .values()" # step is NOT atomic against a writer that fetched the dict # reference just before the swap and inserts a brand-new key just # after - a lock closes that window instead of leaving it to luck. self._telemetry_pending = {} self._telemetry_lock = threading.Lock() self.split = False self.powered = True # radio's power state (CI-V 0x18). After # power OFF the radio doesn't respond — polling is skipped self.vfo = "VFOA" # active VFO — tracked locally (the IC-7300 # has no CI-V query for the active VFO). Default A # = the radio is always on VFO A after power-on. self.s_meter = 0.0 # Current Set Level slider values (CI-V 14 ), in UI units # (lvl['min']..lvl['max']) — read from the radio on connect and # periodically in the poller, so the UI sliders start from the # radio's actual settings, not from 0. self.level_values = {} self.connected = False self.sim = True self.last_err = "" self.last_msg = "" # Epoch seconds since we last fell into sim/disconnected, or None # while actually connected. The only previous signal of a CI-V # drop was a server-console log line the remote operator never # sees - this feeds a small "disconnected for Xs" badge in the UI # (see _start_sim()/_open() where it's set/cleared+broadcast, and # ws.js/ui.js's rig_disconnected handling). self._disconnected_since = None self.rigctld_path = "(tryb bezposredni CI-V)" self.rigctld_found = True # rigctld isn't used, but the UI expects this # ── connection configuration ── self.addr = 0x94 self.port = "COM3" self.speed = 115200 self.model = "3073" # ── model profile (CI-V) — set in connect() ── self.profile = get_civ_profile(self.model) self.mode_map = self.profile["mode_map"] self.mode_rev = {v: k for k, v in self.mode_map.items()} self.scope_max = self.profile["scope_max"] self.scope_header_len = self.profile["scope_header_len"] # ── internal state ── self._ser = None self._running = False self._reader_th = None self._poller_th = None self._sim_th = None self._reconnect_th = None # ── DTR/RTS keyer ───────────────────────────────────────────────────── # Serial port for DTR/RTS keying (may be the same one as CI-V or a # separate COM port from a USB adapter). Keys the DTR or RTS line # to generate a Morse signal from text (instead of CI-V cmd 17). self._keyer_port = None # separate port for DTR/RTS, or None (=use _ser) self._keyer_line = None # 'DTR' | 'RTS' | None self._keyer_ser = None # serial.Serial for a separate port, or a ref to _ser self._keyer_stop = threading.Event() self._keyer_th = None # transactions (one at a time) self._txlock = threading.Lock() self._wlock = threading.Lock() # CW/SSB PTT priority (2026-08-25): set True for the brief window # around a PTT ON/OFF command (or the whole CW send sequence) so # _poller_loop below skips its ENTIRE cycle instead of racing PTT # for the bus - the background poller (freq/mode/S-meter/TX # meters, several transacts back-to-back) was live-suspected of # crowding out PTT ON's ACK reply on a busy CI-V bus ("puste # strzały" - PTT logged as sent but never actually keyed). This is # a stronger guarantee than _txlock alone: _txlock only prevents # two transacts' RESPONSES from getting scrambled together, it # doesn't stop the poller from simply being first in line and # making PTT wait its turn. self._tx_priority_active = False self._resp_ev = threading.Event() self._resp_cmd = None self._resp_sub = None self._resp_payload = None self._resp_matched_cmd = None # which of the 'expect' codes actually matched (e.g. 0xFB vs 0xFA) — see _transact # CI-V TCP Bridge — subscribers to raw bytes from the radio. # Callback called for every chunk of bytes received from the port, # BEFORE it's parsed in _reader_loop. Used by civ_bridge.py to # broadcast raw CI-V to TCP clients (Logger32, HRD, etc.). # The callback must be thread-safe (called from the _reader_loop thread). self._bridge_listeners: list = [] self._bridge_lock = threading.Lock() # scope self.scope_running = False self._scope_acc = bytearray() self._latest_scope = None # freshest complete frame (read by the scope pump) self._scope_total = 0 self._scope_logged = 0 self._tx_logged = 0 self._rx_logged = 0 self._scope_last = 0.0 # scope header (parsed from 0x27 0x00, see _handle_scope) self._scope_mode_code = 0 # 00=Center,01=Fixed,02=SCROLL-C,03=SCROLL-F self._scope_center = self.freq self._scope_span_hz = 20000 # default 20kHz span (real IC-7300 value, see SCOPE_SPAN_KHZ) self._scope_lo = self.freq - self._scope_span_hz // 2 self._scope_hi = self.freq + self._scope_span_hz // 2 self._scope_oor = 0 # filter width (CI-V 1A 03), polled in the poller self._filter_idx = 0 self._filter_width_hz = 2400 # default SSB width # ════════════════════════════════════════════════════════════════════════ # Async interface (called by endpoints / rig_poll) # ════════════════════════════════════════════════════════════════════════ async def connect(self, cfg: dict, override: dict | None = None) -> bool: cfg = cfg or self.cfg or {} 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}} self.model = str(rig.get("model", "3073")) # Load the profile for this model self.profile = get_civ_profile(self.model) self.mode_map = self.profile["mode_map"] self.mode_rev = {v: k for k, v in self.mode_map.items()} self.scope_max = self.profile["scope_max"] self.scope_header_len = self.profile["scope_header_len"] self.port = rig.get("port", "COM3") try: self.speed = int(rig.get("speed", self.profile["default_baud"])) except Exception: self.speed = self.profile["default_baud"] civ = rig.get("civ") or rig.get("civAddr") if civ: self.addr = self._parse_civ(civ) else: self.addr = self.profile["default_addr"] self._rig_name = rig.get("name", self.profile.get("name", "Radio")) self.log(f"[civ] DIRECT CI-V mode — model={self.model} " f"({self.profile.get('name','?')}) port={self.port} " f"{self.speed}bd CI-V=0x{self.addr:02X}") if self.profile is not None and "NIEZWERYFIKOWANE" in self.profile.get("notes", ""): self.log(f"[civ] WARNING for {self.profile.get('name')}: {self.profile['notes']}") # If this object already had an open port (e.g. clicking "Connect" # again without changing the model — webapp.py doesn't create a new # CivRig in that case) — close the old connection before opening a # new one, otherwise serial.Serial() gets a PermissionError (port # already open by this same process). if self._ser is not None: self.log("[civ] Reconnecting — closing existing port") self.close() await asyncio.sleep(0.3) return await asyncio.to_thread(self._open) async def get_freq(self) -> int: return self.freq async def get_freq_live(self, timeout: float = 0.3): """Fresh CI-V frequency read straight from the radio (cmd 03), bypassing self.freq — set_freq() sets self.freq OPTIMISTICALLY, before the write is even attempted, so self.freq/get_freq() being "correct" proves nothing about whether the write actually reached the radio. Used by webapp.py's ft8_qsy handler to verify+retry a band-select retune instead of firing the set_freq() write blind. Returns None on timeout/no response (sim mode returns self.freq).""" if self.sim or not self._ser: return self.freq bp = await asyncio.to_thread(self._transact, bytes([0x03]), {0x03, 0x00}, timeout) if not bp: return None f = bcd_to_freq(bp[:5]) return f if f and f > 0 else None async def get_mode(self): return self.mode, self.bw async def get_smeter(self) -> float: return self.s_meter async def set_freq(self, hz: int): self.freq = int(hz) self._local_freq_set_at = time.time() self._local_freq_set_val = int(hz) if self.sim or not self._ser: return # Fire-and-forget: we don't wait for ACK (0xFB/0xFA) — the radio # rarely NAKs a frequency set, and waiting up to 0.4s for the ACK # caused a noticeable delay on click-to-tune / fast retuning. try: await asyncio.to_thread(self._write, bytes([0x05]) + freq_to_bcd(hz)) except Exception as e: self.log(f"[civ] set_freq write error: {e}") async def set_freq_b(self, hz: int): """Set the VFO-B (inactive VFO) frequency without switching to it. CI-V: 25 01 [5B freq BCD] (per doc p.19-13 "Selected/unselected VFO frequency settings", 01=unselected). Fire-and-forget like set_freq — the radio rarely NAKs a frequency set. """ self.freq_b = int(hz) if self.sim or not self._ser: return try: await asyncio.to_thread(self._write, bytes([0x25, 0x01]) + freq_to_bcd(hz)) except Exception as e: self.log(f"[civ] set_freq_b write error: {e}") def set_scope_span(self, span_hz: int) -> bool: """Set the waterfall span (Center mode). CI-V: 27 15 [VFO-select][5B BCD LE]. Returns True only if the radio actually ACKed (0xFB) the command, False on NG (0xFA), timeout, or an unsupported span_hz. Supported values (Hz): 5000, 10000, 20000, 50000, 100000, 200000, 500000, 1000000 — see SCOPE_SPAN_KHZ; anything else is rejected. FIX: the radio NGed every span change with the old encoding. Verified against Hamlib's icom.c (rig_set_level, RIG_LEVEL_SPECTRUM_SPAN case) - the real CI-V frame for 27 15 is 6 DATA bytes, not 5: byte 0 = spectrum VFO select (icom_get_spectrum_vfo(): 0x00 for a radio without a Sub receiver, which is the IC-7300 - our old code sent straight into the BCD span without this byte at all) byte 1-5 = 5-byte BCD-LE of span_hz/2, NOT span_hz - Icom represents scope span on the wire as a +/- (radius) value (confirmed symmetric in Hamlib: the SET path does val/2 before to_bcd, the GET/read-back path does *2 after from_bcd). Our old code encoded the full span_hz, which for every value except by pure coincidence produces a completely different (and invalid) number on the wire - explains the NG on every span tested live. Also see _transact's docstring — this was the only CI-V "set" command in this file that used to be fire-and-forget with zero response check, which is why the NG went unnoticed until logged. """ if span_hz not in SCOPE_SPAN_KHZ: return False if self.sim or not self._ser: self._scope_span_hz = span_hz return True half = span_hz // 2 # Break half into 5 BCD little-endian bytes (byte 0 = lowest 2 digits) digits = f"{half:010d}" # e.g. 12500 -> "0000012500" b = [] for i in range(5): lo = int(digits[9-i*2]) hi = int(digits[8-i*2]) b.append((hi << 4) | lo) hexstr = ' '.join(f'{x:02X}' for x in b) resp = self._transact(bytes([0x27, 0x15, 0x00]) + bytes(b), {0xFB, 0xFA}, 0.4) matched = self._resp_matched_cmd if resp is None: self.log(f"[civ] scope span -> {span_hz} Hz (VFO=00, half={half}, bytes: {hexstr}) — NO RESPONSE (timeout)") return False if matched == 0xFA: self.log(f"[civ] scope span -> {span_hz} Hz (VFO=00, half={half}, bytes: {hexstr}) — radio REJECTED (NG)") return False self._scope_span_hz = span_hz self.log(f"[civ] scope span -> {span_hz} Hz (VFO=00, half={half}, bytes: {hexstr}) — OK") return True def set_scope_mode(self, mode_code: int) -> bool: """Set the waterfall scope mode (Center vs Fixed). CI-V: 27 14 [VFO-select][mode byte] - mode_code 0x00=Center (okno podaza za VFO), 0x01=Fixed (stale okno brzegow pasma, nie podaza za VFO). 02/03 (SCROLL-C/SCROLL-F) istnieja w protokole ale nie sa tu wspierane - interfejs ma tylko dwustanowy przelacznik CENT/FIX. Returns True only if the radio ACKed (0xFB), False on NG/timeout/ bad value. Sam byte-layout (27 14 00 xx) jest juz uzywany fire-and-forget w _enable_scope() przy wlaczaniu scope'u (zawsze wymusza Center) - to tu jest pierwsza wersja z realnym ACK/NG-check, do recznego przelaczania przez usera. """ if mode_code not in (0x00, 0x01): return False if self.sim or not self._ser: self._scope_mode_code = mode_code return True resp = self._transact(bytes([0x27, 0x14, 0x00, mode_code]), {0xFB, 0xFA}, 0.4) matched = self._resp_matched_cmd if resp is None: self.log(f"[civ] scope mode -> {mode_code} (VFO=00) — NO RESPONSE (timeout)") return False if matched == 0xFA: self.log(f"[civ] scope mode -> {mode_code} (VFO=00) — radio REJECTED (NG)") return False self._scope_mode_code = mode_code self.log(f"[civ] scope mode -> {mode_code} (VFO=00) — OK") return True async def set_mode(self, mode: str, bw: int = 0, fil: int = 0): """ Set the mode (CI-V 06 ). fil: 1=FIL1, 2=FIL2, 3=FIL3 (mechanical/DSP filters configured directly in the radio — we only select WHICH one is active, we don't change its width). 0 = don't change (keep self.filter_num, or default to FIL1 if unknown). NOTE on USB-D/LSB-D: these are NOT separate CI-V mode bytes on the IC-7300 — the radio only understands the base mode (USB=1, LSB=0) + a separate "DATA mode" command (1A 06 01=ON/00=OFF). An earlier version of this function tried to find "USB-D" in mode_rev (which only contains the base modes from _BASE_MODE_MAP), didn't find it, and SILENTLY sent the default byte (1=USB) WITHOUT enabling DATA mode — the effect was the radio staying in plain USB, even though the UI showed "USB-D". Now: we recognize the "-D" suffix, send the base mode + 1A 06 01, and default to FIL1 (the widest/most commonly used DATA filter, matching the configuration already set in the radio — we don't change the filter's WIDTH, only which slot is active). """ is_data = mode.endswith("-D") base_mode = mode[:-2] if is_data else mode # "USB-D" -> "USB" self.mode = mode # keep the FULL name (with "-D") for the UI/higher-level logic if bw: self.bw = bw if fil in (1, 2, 3): self.filter_num = fil elif is_data and not getattr(self, "_data_filter_initialized", False): # First entry into DATA mode with no explicit filter given — # default to FIL1 (FIL1 already has a width of 3000Hz # configured directly in the radio). self.filter_num = 1 fb = getattr(self, "filter_num", 1) if self.sim or not self._ser: self.data_mode = is_data return # Older Icoms (e.g. IC-746) predate the DATA-mode/filter-byte # extension to the 06 command entirely - Hamlib's own icom.c sends # them a bare "06 " (icom_set_mode_without_data(), icmode_ext # forced to -1 for RIG_IS_IC746 among others). Sending our 3-byte # "06 " to one of these would be a malformed frame the # radio doesn't recognize. Flag lives in the model's civ_profiles.py # entry; defaults to True so every already-verified model (IC-7300 # family) is unaffected. _has_fil_byte = self.profile.get("mode_has_filter_byte", True) mb = self.mode_rev.get(base_mode, 1) _mode_cmd = bytes([0x06, mb, fb]) if _has_fil_byte else bytes([0x06, mb]) _mode_resp = await asyncio.to_thread(self._transact, _mode_cmd, {0xFB, 0xFA}, 0.4) if _mode_resp is None or self._resp_matched_cmd == 0xFA: _why = "no ACK (timeout)" if _mode_resp is None else "radio REJECTED (NG)" self.log(f"[civ] set_mode: base mode-set ({_mode_cmd.hex(' ')}) - {_why}") # Turn DATA mode on/off with a separate command (1A 06), independent # of the base mode — this is the actual USB<->USB-D switch on the # IC-7300. Radios whose profile says they don't have this command # at all (see _has_fil_byte comment above - same generation split) # just stay in the base mode; webapp.py already downgrades any # "USB-D" request to plain "USB" before it gets here for those # models, so is_data should never be True in practice, but the # profile check below is the real guard, not that upstream downgrade. # # FIX (audit 2026-08-26, cross-checked against the official CI-V # doc): this command takes TWO data bytes - # [1]=DATA mode(00/01), [2]=filter select (01/02/03=FIL1/2/3 when # ON, must be 00 when OFF) - a single-byte payload (just the on/off # flag) was sent here before, silently malformed on the wire. The # `expect` set was also wrong: {0x1A} with sub=0x06 waits for an # ECHO of the same command family (the read-QUERY response # format), not the ACK/NG (0xFB/0xFA) a SET command (one that # carries a data byte) actually gets - same pattern already fixed # for set_scope_span/set_usb_keying/set_bk_in. In practice this # `_transact` almost certainly always timed out silently (0.4s), # every single USB<->USB-D switch, with the result discarded # either way - self.data_mode was set unconditionally regardless # of whether the radio actually switched. if self.profile.get("data_mode_supported", True) and is_data != self.data_mode: _data_fil_byte = fb if is_data else 0x00 _data_resp = await asyncio.to_thread(self._transact, bytes([0x1A, 0x06, 0x01 if is_data else 0x00, _data_fil_byte]), {0xFB, 0xFA}, 0.4) if _data_resp is None or self._resp_matched_cmd == 0xFA: _why = "no ACK (timeout)" if _data_resp is None else "radio REJECTED (NG)" self.log(f"[civ] set_mode: DATA mode -> {is_data} FIL{_data_fil_byte} - {_why} (radio may not have switched)") self.data_mode = is_data self._data_filter_initialized = True self._filter_width_hz = filter_width_hz(self.mode, getattr(self, "_filter_idx", 0), self.data_mode) self.bcast({"type": "filter_width", "hz": self._filter_width_hz, "idx": getattr(self, "_filter_idx", 0), "mode": self.mode, "dataMode": self.data_mode}) async def set_ptt(self, on: bool): # FIX (reported live 2026-08-25): "CW i SSB maja piorytet w # nadawaniu, one nie moga nie docierac do radia" — this had the # exact same unchecked-ACK bug as send_cw_message's PTT ON did # before its fix: self.ptt was set to the REQUESTED state # unconditionally, before the CI-V command was even attempted, and # the transact's ACK/timeout result was discarded entirely. A # dropped/unanswered PTT command (busy CI-V bus - the same # background poller contention suspected in the CW bug) looked # identical to a real success: every caller (SSB mic, FT8 TX, # manual PTT button) believed transmit had started/stopped when # the radio might never have actually done so. # # Now: retries once on a missing ACK, and for PTT ON specifically # (SSB/FT8 - like the CW fix, no point pretending we're # transmitting if the radio never confirmed it) refuses to claim # self.ptt=True if still unconfirmed after the retry. PTT OFF # always ends up reflecting "off" regardless (same reasoning as # CW's PTT OFF - we want the intent to stop transmitting recorded # either way, a truly stuck radio needs to be visible via the log, # not via a state that pretends nothing is wrong). Wrapped in the # same self._tx_priority_active window as CW (see _poller_loop) # so the routine telemetry poller doesn't compete with PTT itself # for the bus - this is not raised as an exception (many call # sites across webapp.py, some in hot paths like FT8 TX) - a # failure is logged loudly and reflected in self.ptt instead. on = bool(on) if on and not self.ptt: # New transmission starting — reset ALC/PWR peak-hold so it # reflects THIS transmission, not a leftover from the last one. self._alc_peak = 0.0 self._alc_peak_raw = 0 self._pwr_peak = 0.0 self._pwr_peak_raw = 0 if self.sim or not self._ser: self.ptt = on return self._tx_priority_active = True try: resp = await asyncio.to_thread(self._transact, bytes([0x1C, 0x00, 0x01 if on else 0x00]), {0xFB, 0xFA}, 0.4) if resp is None: self.log(f"[civ] set_ptt({on}): no ACK from radio - retrying once") resp = await asyncio.to_thread(self._transact, bytes([0x1C, 0x00, 0x01 if on else 0x00]), {0xFB, 0xFA}, 0.4) if resp is None: self.log(f"[civ] set_ptt({on}): STILL no ACK after retry - radio may NOT have " f"{'started' if on else 'stopped'} transmitting!") if on: self.ptt = False return finally: self._tx_priority_active = False self.ptt = on # ── DTR/RTS Keyer ───────────────────────────────────────────────────────── # Morse table — character -> string of '.' and '-' _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. Uses standard Morse timing (PARIS): dit = 1 unit, dah = 3, gap between elements = 1, gap between letters = 3, gap between words = 7. Summing the real element counts (instead of assuming 10 units/char) makes the PTT hold match the keyer, so the carrier drops right after the last element and long macros are never cut short. """ 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 # word gap first_char = True continue code = self._MORSE_TABLE.get(ch) if not code: continue if not first_char: units += 3.0 # gap before this letter first_char = False for i, sym in enumerate(code): units += 3.0 if sym == '-' else 1.0 # dah / dit if i < len(code) - 1: units += 1.0 # intra-character gap return units * dit def set_usb_keying(self, value: int) -> bool: """Set the IC-7300's internal "USB Keying (CW)" setting (CI-V 1A 05 00 97 XX). value: 0=OFF, 1=USB(A) DTR, 2=USB(A) RTS, 3=USB(B) DTR, 4=USB(B) RTS. The IC-7300 exposes TWO virtual COM ports over its single rear [USB] connector; this setting tells the radio's own firmware to treat the DTR or RTS line of one of them as a direct hardware key input - exactly like an external transistor keyer wired into the KEY jack, but needs no extra cable. This is what lets configure_keyer() below key the radio for real over a second COM port from the SAME USB cable already used for CI-V, instead of requiring separate hardware. Returns True only on an actual radio ACK (0xFB), False on NG/timeout. """ if value not in (0, 1, 2, 3, 4): return False if self.sim or not self._ser: return True resp = self._transact(bytes([0x1A, 0x05, 0x00, 0x97, value]), {0xFB, 0xFA}, 0.4) matched = self._resp_matched_cmd if resp is None: self.log(f"[civ] USB Keying (CW) -> {value} — NO RESPONSE (timeout)") return False if matched == 0xFA: self.log(f"[civ] USB Keying (CW) -> {value} — radio REJECTED (NG)") return False self.log(f"[civ] USB Keying (CW) -> {value} — OK") return True def set_bk_in(self, on: bool) -> bool: """Set BK-IN (CI-V 16 47 00/01). Off by default at startup (safety - see the comment in connect(): BK-IN + remote cmd-17 CW caused a continuous carrier). LIVE-CONFIRMED 2026-08-26: BK-IN must be ON for the radio to react to the DTR/RTS "USB Keying" key line at all - with it off the radio simply ignores the key line outside an active PTT. Callers should turn this ON only while cwMethod is dtr/rts, and back OFF for cat/auto, to keep the original cmd-17 safety behavior intact. """ if self.sim or not self._ser: return True resp = self._transact(bytes([0x16, 0x47, 0x01 if on else 0x00]), {0xFB, 0xFA}, 0.4) matched = self._resp_matched_cmd if resp is None: self.log(f"[civ] BK-IN -> {on} — NO RESPONSE (timeout)") return False if matched == 0xFA: self.log(f"[civ] BK-IN -> {on} — radio REJECTED (NG)") return False self.log(f"[civ] BK-IN -> {on} — OK") return True def configure_keyer(self, port: str, line: str): """ Configure the DTR/RTS keyer. port: 'COM3', 'COM5' etc. (may be the same as CI-V or a separate one) line: 'DTR' or 'RTS' If port == '' (empty) — use the same port as CI-V (_ser). """ import serial as _serial self._keyer_line = line.upper() if line else 'DTR' # FIX (audit 2026-08-26): this compared against self._port, an # attribute that never existed on this class (the real one is # self.port, no underscore) - getattr's default ('') made the # comparison ALWAYS true for any non-empty port string, so typing # in the SAME port CI-V already uses (not leaving it blank) always # took the "separate port" branch below and tried to open a SECOND # serial.Serial() on a port Windows already has open for CI-V - # that open reliably fails (port busy), caught by the except # below, which sets _keyer_ser=None and silently falls back to # sharing self._ser via _set_key()'s own fallback anyway. Keying # still worked by accident through that fallback, but a real # second-port radio (IC-7610/9700) would have hit this same typo # in the OTHER direction with no safety net. Also compare # case-insensitively - Windows COM port names ("COM4" vs "com4") # aren't reliably normalized before reaching here. same_as_civ = (not port) or (port.strip().upper() == getattr(self, 'port', '').strip().upper()) if port and not same_as_civ: # Separate port for the keyer try: if self._keyer_ser and self._keyer_ser is not self._ser: try: self._keyer_ser.close() except: pass self._keyer_ser = _serial.Serial(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 self._keyer_port = port self.log(f"[keyer] Opened port {port} for {line} keying") except Exception as e: self.log(f"[keyer] Error opening port {port}: {e}") self._keyer_ser = None else: # Use the same port as CI-V. Deliberately do NOT cache self._ser # here - if the radio reconnects later (fresh serial.Serial # object), a cached reference would go stale and _set_key would # keep toggling DTR/RTS on a closed port. Leave _keyer_ser unset # so _set_key always reads self._ser fresh, at the moment of # each key-down/key-up. self._keyer_ser = None self._keyer_port = '' def _set_key(self, state: bool): """Set the DTR or RTS line (True=KEY DOWN, False=KEY UP).""" ser = self._keyer_ser or self._ser if not ser: return try: if self._keyer_line == 'RTS': ser.rts = state else: ser.dtr = state except Exception as e: self.log(f"[keyer] _set_key error: {e}") 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). stop_event: threading.Event — aborts sending immediately. connect() force-disables BK-IN at startup (safety default); the webapp layer re-enables it via set_bk_in(True) whenever cwMethod is dtr/rts (see /api/cw/method and _on_rig_reconnected in webapp.py). LIVE-CONFIRMED 2026-08-26: BK-IN must be ON for the radio to react to the DTR/RTS key line at all - with it off the radio ignores the key line entirely outside an active PTT (USB Keying (00 97) ACKs fine but nothing keys). PTT is still explicitly held here (below) for the whole message regardless of BK-IN, so the T/R relay has a guaranteed settle window before the first dit - DTR/RTS only shapes the dits/dahs WITHIN that PTT window. """ import time as _time wpm = max(5, min(60, wpm)) dit = 1.200 / wpm # dit duration in seconds if self.sim or not self._ser: return _ptt_resp = self._transact(bytes([0x1C, 0x00, 0x01]), {0xFB, 0xFA}, 0.4) _ptt_rejected = self._resp_matched_cmd == 0xFA if _ptt_resp is None or _ptt_rejected: _why = "radio REJECTED PTT ON (NG)" if _ptt_rejected else "no ACK from radio (timeout)" self.log(f"[keyer] DTR/RTS send: PTT ON - {_why} - aborting, nothing keyed") return self.ptt = True try: # Delay at the start — the radio needs a moment to switch to # transmit (T/R relay, tuner). Without this, the first dit goes # out before the radio is transmitting, and the start of the # character gets clipped. Configurable via cwPttDelay (ms). _ptt_delay = float(self.profile.get("cwPttDelay", 250)) / 1000.0 _time.sleep(max(0.05, _ptt_delay)) def key_down(t): self._set_key(True) _time.sleep(t) self._set_key(False) def pause(t): _time.sleep(t) for char in text.upper(): if stop_event.is_set(): break if char == ' ': pause(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 if sym == '.': key_down(dit) else: key_down(dit * 3) if i < len(code) - 1: pause(dit) # gap between elements pause(dit * 3) # gap between letters self._set_key(False) # make sure the key is released finally: try: self._transact(bytes([0x1C, 0x00, 0x00]), {0xFB, 0xFA}, 0.4) except Exception: pass self.ptt = False async def send_cw_dtr_rts(self, text: str, wpm: int): """ Send CW text via DTR/RTS keying (asynchronously, in a separate thread). Aborts a previous send if one is active. """ # Stop a previous send 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: self.log(f"[keyer] SIM DTR/RTS: {text!r} @ {wpm} WPM") return stop_ev = self._keyer_stop th = threading.Thread( target=self._send_morse_blocking, args=(text, wpm, stop_ev), daemon=True, name="morse-keyer" ) self._keyer_th = th th.start() async def stop_cw_dtr_rts(self): """Abort DTR/RTS sending immediately.""" self._keyer_stop.set() self._set_key(False) if not self.sim and self._ser: try: await asyncio.to_thread(self._transact, bytes([0x1C, 0x00, 0x00]), {0xFB, 0xFA}, 0.4) except Exception: pass self.ptt = False async def set_split(self, on: bool): self.split = bool(on) if self.sim or not self._ser: return await asyncio.to_thread(self._transact, bytes([0x0F, 0x01 if on else 0x00]), {0xFB, 0xFA}, 0.4) async def set_power(self, on: bool): """ Turn the radio on/off (CI-V cmd 0x18 0x01/0x00). IMPORTANT — power ON requires a "wakeup": when the radio is fully off, the UART isn't listening at full speed. Standard Icom sequence: send a long preamble of 0xFE bytes (~150ms at the given baud rate) BEFORE the 0x18 0x01 frame — this "wakes up" the radio's UART receiver. Requires in the radio: MENU > SET > Connectors > CI-V > "CI-V Transceive" = ON, and 12V power applied to the radio (USB alone isn't enough when the radio is in full OFF mode on some models). After "power OFF" the CI-V link stops responding to normal transactions (the radio isn't listening) — this is expected until we send wakeup+power ON. """ if self.sim or not self._ser: return # Remember the power state — after power OFF the radio doesn't # respond to any CI-V transactions. Without this, the telemetry # loop keeps polling the S-meter and prints "S-metr transact fail" # every second, spamming the log. self.powered = bool(on) if on: await asyncio.to_thread(self._wakeup_and_power_on) else: await asyncio.to_thread(self._transact, bytes([0x18, 0x00]), {0xFB, 0xFA}, 0.6) def _wakeup_and_power_on(self): """ Sends the 0xFE preamble (wakeup) + the 0x18 0x01 frame (power ON). Preamble duration depends on baud rate: at 115200bd Icom recommends at least ~150ms of continuous 0xFE transmission (about 150 bytes). After power-on the radio needs a few seconds to boot — we don't wait for an ACK (the transaction usually times out, since the radio isn't ready yet). """ try: preamble_len = max(150, self.speed // 768) # ~150ms of data preamble = b"\xfe" * preamble_len frame = bytes([0xFE, 0xFE, self.addr, CTRL_ADDR, 0x18, 0x01, 0xFD]) with self._wlock: if self._ser: self._ser.write(preamble) time.sleep(0.05) self._ser.write(frame) self._write_fail_count = 0 self.log(f"[civ] wakeup+power ON sent (preamble {preamble_len}B)") except Exception as e: self.log(f"[civ] wakeup_and_power_on error: {e}") # Same dead-port counter as _write() - this is the exact path # that failed to recover live during a simulated power-loss # test (radio came back, but the COM handle didn't) - without # this, a stuck port here just logs forever and never triggers # the close+reconnect self-heal. self._write_fail_count = getattr(self, "_write_fail_count", 0) + 1 if self._write_fail_count >= 3: self._handle_dead_port() async def set_vfo(self, vfo: str): """ Switch the active VFO — CI-V cmd 0x07. vfo: 'VFOA' -> 0x00, 'VFOB' -> 0x01 """ _name = "VFOA" if vfo.upper() in ("VFOA", "A") else "VFOB" self.vfo = _name # track the state (also in sim — the UI needs to know the truth) if self.sim or not self._ser: return sub = 0x00 if _name == "VFOA" else 0x01 await asyncio.to_thread(self._transact, bytes([0x07, sub]), {0xFB, 0xFA}, 0.4) async def vfo_equalize(self): """A->B: copy the frequency/mode from VFO A to VFO B (CI-V 07 A0).""" if self.sim or not self._ser: self.freq_b = self.freq return # 07 A0 copies A's freq/mode to B on the radio itself - deterministic, # so update our own freq_b to match immediately rather than leaving # it stale until some later unrelated poll happens to refresh it # (previously never updated here at all: the WS handler broadcast # this same, now-stale, self.freq_b right back to every client, # which looked fine in the UI but the radio's real VFO B - revealed # by actually switching to it or engaging split - still had the OLD # frequency). await asyncio.to_thread(self._transact, bytes([0x07, 0xA0]), {0xFB, 0xFA}, 0.4) self.freq_b = self.freq async def vfo_swap(self): """A<->B: swap VFO A and B (CI-V 07 B0).""" if self.sim or not self._ser: self.freq, self.freq_b = self.freq_b, self.freq return await asyncio.to_thread(self._transact, bytes([0x07, 0xB0]), {0xFB, 0xFA}, 0.4) self.freq, self.freq_b = self.freq_b, self.freq async def set_func(self, func: str, on: bool): """ Turn a radio function on/off — CI-V cmd 0x16 <00|01>. The name->subcommand mapping is in the model profile (profile['funcs']). """ if self.sim or not self._ser: return funcs = self.profile.get("funcs", {}) sub = funcs.get(func.upper()) if sub is None: self.log(f"[civ] set_func: unknown function '{func}' for {self.profile.get('name')}") return await asyncio.to_thread(self._transact, bytes([0x16, sub, 0x01 if on else 0x00]), {0xFB, 0xFA}, 0.4) async def set_preamp(self, level: int): """ Preamp — CI-V 16 02 <00|01|02>. level: 0=OFF, 1=Preamp1 (P.AMP1), 2=Preamp2 (P.AMP2, IC-7300/7610 etc. only) """ level = max(0, min(2, int(level))) self.preamp = level if self.sim or not self._ser: return await asyncio.to_thread(self._transact, bytes([0x16, 0x02, level]), {0xFB, 0xFA}, 0.4) async def set_attenuator(self, on: bool): """ Attenuator — CI-V 11 <00|20>. The IC-7300 has one 20dB attenuator stage: 00=OFF, 20=ON (20dB). """ on = bool(on) self.attenuator = on if self.sim or not self._ser: return await asyncio.to_thread(self._transact, bytes([0x11, 0x20 if on else 0x00]), {0xFB, 0xFA}, 0.4) async def set_tuner(self, on: bool): """ Antenna Tuner ON/OFF — CI-V 1C 01 <00|01>. 00=OFF (tuner bypassed), 01=ON (tuner switched into the signal path — required before AUTOTUNE). """ on = bool(on) self.tuner = on if self.sim or not self._ser: return await asyncio.to_thread(self._transact, bytes([0x1C, 0x01, 0x01 if on else 0x00]), {0xFB, 0xFA}, 0.4) async def start_tuner_autotune(self): """ Antenna Tuner START (auto-tuning cycle) — CI-V 1C 01 02. The radio starts searching for a match (a few seconds, generates a low-power TX signal). The tuner must be ON (01) before calling this — in practice the IC-7300 turns the tuner on automatically together with START, but we set it ON right before, just to be safe. """ self.tuner = True if self.sim or not self._ser: return # Step 1: Make sure the tuner is ON await asyncio.to_thread(self._transact, bytes([0x1C, 0x01, 0x01]), {0xFB, 0xFA}, 0.5) # Step 2: Short delay — the radio needs a moment to switch over await asyncio.sleep(0.1) # Step 3: START autotune — the radio generates a TX signal and tunes the ATU # 2.0s timeout since the radio may not respond right away (busy tuning) await asyncio.to_thread(self._transact, bytes([0x1C, 0x01, 0x02]), {0xFB, 0xFA}, 2.0) async def set_level(self, level_name: str, value: float): """ Set a level (e.g. RFPOWER, AF, MICGAIN) — CI-V cmd 0x14 . 'value' is in UI units (the level['min']..level['max'] range from the profile) and is linearly scaled to 0-255 for CI-V. The name->subcommand+range mapping is in the model profile (profile['levels']). """ levels = self.profile.get("levels", {}) lvl = levels.get(level_name.upper()) if lvl is None: self.log(f"[civ] set_level: unknown level '{level_name}' for {self.profile.get('name')}") return lo, hi = lvl["min"], lvl["max"] try: value = float(value) except (TypeError, ValueError): return # Scale value (lo..hi) -> civ_val (0..civ_max) civ_max = lvl.get("civ_max", 255) if hi != lo: frac = (value - lo) / (hi - lo) else: frac = 0.0 frac = max(0.0, min(1.0, frac)) civ_val = int(round(frac * civ_max)) if self.sim or not self._ser: return # CI-V 0x14: subcommand + 2-byte BCD (0000-0255) bcd = self._int_to_bcd2(civ_val) # Update the local cache right away (optimistically) — a read from # the radio will happen on the next _read_all_levels()/poller cycle # and may correct it, but this lets the UI/other clients see the # new value immediately after the WS broadcast (if webapp.py does one). self.level_values[level_name.upper()] = value if self.sim or not self._ser: return await asyncio.to_thread(self._transact, bytes([0x14, lvl["sub"]]) + bcd, {0xFB, 0xFA}, 0.4) def _read_level(self, level_name: str, lvl: dict) -> float | None: """ Read the current Set Level slider value (CI-V 14 ) from the radio and rescale the BCD (0..civ_max) to UI units (lvl['min']..lvl['max']). Returns None if the radio didn't respond. Called synchronously (in the connection/poller thread, not in the asyncio loop). """ p = self._transact(bytes([0x14, lvl["sub"]]), {0x14}, 0.3, sub=lvl["sub"]) if not p or len(p) < 3 or p[0] != lvl["sub"]: return None civ_val = bcd2(p[1:3]) civ_max = lvl.get("civ_max", 255) lo, hi = lvl["min"], lvl["max"] frac = civ_val / civ_max if civ_max else 0.0 frac = max(0.0, min(1.0, frac)) return lo + frac * (hi - lo) def _read_all_levels(self): """ Read the current values of ALL Set Level sliders defined in the radio profile and store them in self.level_values. Called once after connecting (_open) — so the UI starts from the radio's actual settings, not from 0/min. NOTE: NB_LEVEL and BKINDL have different subcommands (0x12 vs 0x0F after the fix), so there's no read collision between them anymore. """ levels = self.profile.get("levels", {}) for name, lvl in levels.items(): try: val = self._read_level(name, lvl) if val is not None: self.level_values[name] = val except Exception as e: self.log(f"[civ] _read_all_levels: error reading {name}: {e}") @staticmethod def _int_to_bcd2(v: int) -> bytes: """ 0-255 (or up to 999) -> 2-byte BCD like Icom CI-V Set Level (e.g. 0255 -> bytes([0x02, 0x55]), per the documentation: "00 00=min. to 02 55=max."). Inverse of bcd2(). FIX: the previous version returned bytes([(d1<<4)|d0, d2]) which for 255 gave [0x55, 0x02] — bytes swapped relative to bcd2() and the CI-V spec. After this fix, bcd2(_int_to_bcd2(v)) == v for v in 0..255. """ v = max(0, min(255, int(v))) hi = v // 100 # 0-2 lo = v % 100 # 0-99 byte0 = hi # 0x00, 0x01, or 0x02 (BCD == binary for 0-2) byte1 = ((lo // 10) << 4) | (lo % 10) return bytes([byte0, byte1]) # ── Advanced ("admin-only") radio settings ────────────────────────────── # Generic engine for rigs/civ_profiles.py's per-model "adv" table (CI-V # commands beyond the Set Level/Set Func tables above - modulation # input, AGC mode, VOX fine-tuning, NB depth/width, TX delay). See # that table's own docstring for the exact format of each "kind". # webapp.py gates every caller of get_adv/set_adv to the admin role - # civ.py itself doesn't know about users/permissions, same separation # of concerns as every other control in this file. async def get_adv(self, key: str): spec = self.profile.get("adv", {}).get(key.upper()) if spec is None or self.sim or not self._ser: return None return await asyncio.to_thread(self._read_adv, spec) def _read_adv(self, spec: dict): cmd = spec["cmd"] sub = cmd[1] p = self._transact(bytes(cmd), {cmd[0]}, 0.3, sub=sub) if not p or p[0] != sub: return None # 1A 05 xx xx commands carry 2 extra address bytes beyond [top, sub] # that 14/16 xx commands don't - verify they're echoed back correctly # before trusting what follows as the actual data (same principle as # the existing 1A 03/1A 06 sub= checks elsewhere in this file, just # with the deeper 1A 05 menu-item address also checked, not just sub). extra = len(cmd) - 2 if extra: if len(p) < 1 + extra or list(p[1:1 + extra]) != cmd[2:]: return None data = p[1 + extra:] return self._decode_adv(spec, data) @staticmethod def _decode_adv(spec: dict, data: bytes): kind = spec["kind"] if kind == "enum": # wire_base (2026-10-18, live NG on AGC_MODE): most "enum" CI-V # data bytes are 0-indexed (00=first option), but some - e.g. # 16 12 AGC time constant, 01=FAST/02=MID/03=SLOW - start at 1 # on the wire while the UI dropdown still wants a plain 0-based # index for spec["labels"]. Defaults to 0 so every other enum # entry (unaffected) decodes exactly as before. if not data: return None return data[0] - spec.get("wire_base", 0) if kind == "range": if not data: return None v = data[0] + spec.get("offset", 0) return round(v * spec["scale"], 2) if "scale" in spec else v if kind in ("bcd2pct", "bcd2range"): if len(data) < 2: return None raw = bcd2(data[:2]) if kind == "bcd2pct": return round(raw / 255.0, 4) lo, hi = spec["min"], spec["max"] return round(lo + raw / 255.0 * (hi - lo), 1) return None @staticmethod def _encode_adv(spec: dict, value) -> bytes | None: kind = spec["kind"] try: if kind == "enum": v = int(value) if not (0 <= v < spec["n"]): return None return bytes([v + spec.get("wire_base", 0)]) if kind == "range": # Inverse of _decode_adv's v = (raw + offset) * scale. raw = float(value) / spec.get("scale", 1) - spec.get("offset", 0) rawmax = spec.get("rawmax", 255) return bytes([max(0, min(rawmax, int(round(raw))))]) if kind in ("bcd2pct", "bcd2range"): if kind == "bcd2pct": frac = float(value) else: lo, hi = spec["min"], spec["max"] frac = (float(value) - lo) / (hi - lo) if hi != lo else 0.0 frac = max(0.0, min(1.0, frac)) civ = CivRig._int_to_bcd2(int(round(frac * 255))) return civ except (TypeError, ValueError): return None return None async def set_adv(self, key: str, value) -> bool: spec = self.profile.get("adv", {}).get(key.upper()) if spec is None: return False enc = self._encode_adv(spec, value) if enc is None: return False if self.sim or not self._ser: return True cmd = spec["cmd"] # RETRY (2026-10-18, live report: AGC_MODE "od czasu do czasu rzuca # bledem NG" with nothing else happening on the radio at the time - # same bus-contention suspicion as set_ptt's own retry-once above): # only retry on a TRUE timeout (p is None, no response of ANY kind - # likely swallowed by the background poller or dropped on the wire), # never on an explicit 0xFA NG - a genuine rejection means the VALUE # itself was refused, and resending the identical command would # just NG again for the same reason. p = await asyncio.to_thread(self._transact, bytes(cmd) + enc, {0xFB, 0xFA}, 0.4) if p is None: self.log(f"[civ] set_adv({key}): no response - retrying once") p = await asyncio.to_thread(self._transact, bytes(cmd) + enc, {0xFB, 0xFA}, 0.4) ok = p is not None and self._resp_matched_cmd == 0xFB if not ok: reason = "NG (0xFA)" if self._resp_matched_cmd == 0xFA else "no response even after retry" # VERIFY-BY-READBACK (2026-10-18, live report: "dalej pokazuje # NG ale na pewno dziala" - AGC audibly switched even though the # ACK round-trip failed). Likely story: the FIRST attempt's own # ACK actually arrived, just too late (past our 0.4s timeout) - # the radio HAD already applied it - and the retry we then sent # landed while the radio was still finishing up from the first # one, earning a real NG for being redundant/mistimed, not # because the value itself was ever rejected. A flaky ACK # doesn't prove the command didn't take, so check reality # instead of trusting it: read the value back and believe THAT. await asyncio.sleep(0.15) actual = await asyncio.to_thread(self._read_adv, spec) matches = (abs(actual - value) < 0.01 if isinstance(value, (int, float)) and isinstance(actual, (int, float)) else actual == value) if matches: self.log(f"[civ] set_adv({key}): ACK said {reason}, but readback confirms " f"the value DID take ({actual!r}) - reporting success") return True self.log(f"[civ] set_adv({key}): failed - {reason} (readback: {actual!r})") return ok # REMOVED (2026-10-18): RIT/ΔTX (CI-V 0x21) - get_rit/_read_rit/ # set_rit_hz/set_rit_enabled/set_dtx_enabled. RIT ON/OFF and ΔTX ON/OFF # (plain 00/01 bytes) worked; setting the Hz value gave NG every time, # even after fixing the BCD byte order to match this file's proven # frequency convention (freq_to_bcd/bcd_to_freq instead of the # Set-Level-style bcd2). Root cause not found - user decision: drop it # rather than keep chasing it ("nie potrzebujemy tego"). See webapp.py's # removed /api/rig/rit comment for the full removal list. async def get_capabilities(self) -> dict: """ Return the full structure of discovered CI-V capabilities: {"actions": [VFO A/B, func toggles], "sliders": [Set Level], "raw_caps": {feature_id: bool}} Builds the list from the model profile (profile['levels']/['funcs']) — analogous to hamlib_caps.discover_capabilities, but the source is the static CI-V profile (the radio isn't queried for its capabilities, since CI-V has no "dump_caps" command). """ raw_caps = dict(self.profile.get("capabilities", {})) actions = [] if raw_caps.get("vfo_ab"): actions.append({"id": "vfo_a", "label": "VFO A", "group": "vfo", "kind": "vfo_select", "value": "VFOA"}) actions.append({"id": "vfo_b", "label": "VFO B", "group": "vfo", "kind": "vfo_select", "value": "VFOB"}) if raw_caps.get("power"): # Universal label (PL/EN) instead of a Polish description — this # button renders the same in both frontend languages, the color # (green/red) shows the actual ON/OFF state. actions.append({"id": "power_toggle", "label": "PWR ON/OFF", "group": "tx", "kind": "power_toggle", "value": "POWER"}) func_labels = self.profile.get("func_labels", {}) for func_name in self.profile.get("funcs", {}): label = func_labels.get(func_name, func_name) actions.append({ "id": f"func_{func_name.lower()}", "label": label, "group": "func", "kind": "func_toggle", "value": func_name, }) sliders = [] for level_name, lvl in self.profile.get("levels", {}).items(): # Default value = lvl['min'] if not read from the radio yet # (e.g. simulation mode) — otherwise the real setting from # self.level_values (filled in by _read_all_levels() after # connecting, see _open()). current = self.level_values.get(level_name, lvl["min"]) # Slider step: for 0.0-1.0 ranges (scaled to 0-100% in the UI) # we compute a step proportional to the 255 CI-V levels. For # integer ranges (e.g. KEYSPD 6-48 WPM, CWPITCH 300-900Hz) the # step is 1 UI unit — more sensible than 0.16 WPM. if lvl["max"] <= 1.0: step = (lvl["max"] - lvl["min"]) / 255.0 if lvl["max"] != lvl["min"] else 1 else: step = 1 sliders.append({ "id": f"level_{level_name.lower()}", "label": lvl.get("label", level_name), "group": "level", "kind": "set_level", "param": level_name, "min": lvl["min"], "max": lvl["max"], "value": current, "step": step, }) return {"actions": actions, "sliders": sliders, "raw_caps": raw_caps} def close(self): self._running = False self.scope_running = False try: if self._ser: # turn off the scope stream, so the radio doesn't flood the port after disconnecting try: self._write(bytes([0x27, 0x11, 0x00])) time.sleep(0.05) except Exception: pass self._ser.close() except Exception: pass self._ser = None # ── scope control (from the /api/scope/start|stop endpoint) ── def scope_start(self): if self.sim or not self._ser: self.scope_running = True return {"ok": True, "sim": True, "running": True} self._enable_scope(True) self.scope_running = True return {"ok": True, "sim": False, "running": True} def scope_stop(self): self.scope_running = False if self._ser: try: self._enable_scope(False) except Exception: pass return {"ok": True, "running": False} # ════════════════════════════════════════════════════════════════════════ # Serial layer (threads) # ════════════════════════════════════════════════════════════════════════ def _parse_civ(self, civ) -> int: try: s = str(civ).strip().lower().replace("0x", "").replace("h", "") return int(s, 16) except Exception: return self.profile.get("default_addr", 0x94) def _open(self) -> bool: if not HAS_SERIAL: self.last_err = ("Brak biblioteki pyserial — tryb bezposredni CI-V wymaga pyserial. " "Uruchamiam SYMULACJE scope.") self.log(f"[civ] {self.last_err}") return self._start_sim() try: self._ser = serial.Serial(self.port, self.speed, timeout=0.05, write_timeout=1.0) except Exception as e: self.last_err = (f"Nie moge otworzyc {self.port} @ {self.speed}bd: {type(e).__name__}: {e}. " f"Sprawdz: wlasciwy COM, baud CI-V w radiu " f"(domyslnie {self.profile['default_baud']} dla {self.profile.get('name','tego modelu')}), " f"port wolny (zamknij rigctld/N1MM/RCForb). Uruchamiam SYMULACJE scope.") self.log(f"[civ] {self.last_err}") return self._start_sim() # CRITICAL: pyserial sets DTR=True and RTS=True by default when # opening a port. If the radio has DTR/RTS configured as PTT/CW-key # (typical for the IC-7300), the radio IMMEDIATELY goes into TX. We # force a low state right after opening - BEFORE any communication. # The keyer will raise the line itself when it's actually keying # (KEY DOWN). try: self._ser.dtr = False self._ser.rts = False except Exception as _e: self.log(f"[civ] Warning: can't zero out DTR/RTS: {_e}") self.sim = False self._running = True self._reader_th = threading.Thread(target=self._reader_loop, daemon=True) self._reader_th.start() # initial frequency / mode read ok_freq = False for _ in range(3): p = self._transact(bytes([0x03]), {0x03, 0x00}, 0.6) if p: f = bcd_to_freq(p) if f: self.freq = f; ok_freq = True; break time.sleep(0.4) if not ok_freq: # The radio may be physically OFF — the USB-CI-V port exists # (it's powered from USB) but the radio's UART is "asleep" and # doesn't respond to normal queries. Send wakeup+power-on # (0xFE preamble + 0x18 0x01, see IC-7300 manual *3) and retry # after giving the radio time to boot (~2-3s). self.log("[civ] 0x03 no response — trying wakeup+power ON (radio may be OFF)") self._wakeup_and_power_on() time.sleep(2.5) for _ in range(4): p = self._transact(bytes([0x03]), {0x03, 0x00}, 0.6) if p: f = bcd_to_freq(p) if f: self.freq = f; ok_freq = True self.log("[civ] Radio woken up via wakeup+power ON") break time.sleep(0.5) pm = self._transact(bytes([0x04]), {0x04, 0x01}, 0.6) if pm and len(pm) >= 1: base_pm = self.mode_map.get(pm[0]) if base_pm: self.mode = f"{base_pm}-D" if self.data_mode else base_pm if not ok_freq: self.last_err = (f"Radio nie odpowiada na CI-V (0x03), nawet po wakeup+power ON. " f"Sprawdz: model, CI-V address (domyslnie 0x{self.profile['default_addr']:02X} " f"dla {self.profile.get('name','tego modelu')}), " f"baud, kabel/COM, zasilanie 12V radia. Uruchamiam SYMULACJE scope.") self.log(f"[civ] {self.last_err}") self._running = False try: self._ser.close() except Exception: pass self._ser = None return self._start_sim() self.connected = True if self._disconnected_since is not None: self._disconnected_since = None self.bcast({"type": "rig_disconnected", "since": None}) self.last_err = "" self.last_msg = (f"{getattr(self,'_rig_name','Radio')} (CI-V bezposredni) {self.port} " f"{self.speed}bd — {self.freq}Hz {self.mode}") self.log(f"[civ] CONNECTED: {self.last_msg}") # turn on the scope stream + start the telemetry poller self._enable_scope(True) self.scope_running = True # Read the current settings of all sliders (Set Level, CI-V 14 # ) from the radio BEFORE announcing the connection — this way # get_capabilities() (called by _on_rig_reconnected in webapp.py) # returns real values and the UI sliders start from the radio's # actual settings, not from 0. self._read_all_levels() # Read frequency and mode once before starting the poller — # without this, self.freq holds its hardcoded default (14074000) # until the first _poller_loop cycle (a few seconds). Clients # connecting in that window get the wrong frequency in the 'init' # message and the VFO display shows e.g. 14.074 instead of the # radio's actual frequency. Same pattern as _read_all_levels() above. try: # freq: command 03, response 03 or 00 bp = self._transact(bytes([0x03]), {0x03, 0x00}, 0.4) if bp: f = bcd_to_freq(bp[:5]) if f and f > 0: self.freq = f self.log(f"[civ] startup read freq={f}Hz") # mode: command 04, response 04 or 01 bm = self._transact(bytes([0x04]), {0x04, 0x01}, 0.3) if bm and len(bm) >= 1: mode_byte = bm[0] MODE_MAP = {0x00:'LSB',0x01:'USB',0x02:'AM',0x03:'CW', 0x04:'RTTY',0x05:'FM',0x06:'CW-R',0x07:'RTTY-R', 0x08:'LSB-D',0x09:'USB-D',0x11:'PKTUSB',0x12:'PKTLSB'} if mode_byte in MODE_MAP: self.mode = MODE_MAP[mode_byte] self.log(f"[civ] startup read mode={self.mode}") except Exception as e: self.log(f"[civ] startup freq/mode read error (non-blocking): {e}") # Safety: disable BK-IN at startup so the radio doesn't go into # continuous TX. BK-IN (CI-V 16 47 00) may be left on from a # previous session or by mistake. The user can enable it manually # in the panel. try: self._transact(bytes([0x16, 0x47, 0x00]), {0xFB, 0xFA}, 0.4) self.log("[civ] BK-IN disabled at startup (safety)") except Exception as e: self.log(f"[civ] BK-IN disable error (non-blocking): {e}") self._poller_th = threading.Thread(target=self._poller_loop, daemon=True) self._poller_th.start() # Notify webapp.py (via WS broadcast) that a connection has been # (re-)established — the receiver should refresh _caps_cache and # send a fresh /api/rig/features (rig_features) to clients, since # the VFO/PWR/sliders panel could be empty if the connection # happened in the background (e.g. via _reconnect_loop after the # radio was turned on after the server started). try: self.bcast({"type": "rig_reconnected"}) except Exception as e: self.log(f"[civ] bcast rig_reconnected error: {e}") return True def _enable_scope(self, on: bool): # DIAGNOSTIC TEST: HAM_NO_SCOPE=1 forces scope OFF, to check whether # it's the scope stream causing loop stalls / RTT spikes / audio # tearing. If audio is smooth and RTT stable with the scope off — # the scope is the culprit (on 2 cores: the reader thread competing # for CPU). Remove the variable to return to normal operation. import os as _os if _os.environ.get("HAM_NO_SCOPE") == "1": on = False """ Turn the CI-V scope waveform data stream on/off. When turning it on, we set Center mode (27 14 00 00) — in this mode the radio sends center freq + span in the header and the scope window "follows" the current VFO (unlike Fixed mode, where the window is a fixed band-edge range and does NOT track the VFO). This achieves the "like the original scope screen" behavior that's expected. We also set the span to 25kHz (via set_scope_span) — gives a good compromise between preview width and filter-tracking precision. """ try: # Baud 115200 is REQUIRED for the scope. At 19200 (CI-V USB = # "Link to REMOTE") the radio doesn't send waveform data - the # most common cause of "no waterfall". if on and self.speed < 115200: self.log(f"[civ] WARNING baud={self.speed} — scope requires 115200! " f"Set MENU>SET>Connectors>CI-V baud=115200 and " f"CI-V USB Port=Unlink from REMOTE") self._write(bytes([0x27, 0x10, 0x01 if on else 0x00])) # scope ON/OFF time.sleep(0.05) self._write(bytes([0x27, 0x11, 0x01 if on else 0x00])) # data output ON/OFF if on: time.sleep(0.05) # Center mode (00) — the scope tracks the VFO self._write(bytes([0x27, 0x14, 0x00, 0x00])) time.sleep(0.05) # 20kHz span (closest real value to the old, nonexistent # 25kHz default - see SCOPE_SPAN_KHZ) - via set_scope_span # (was a hand-rolled write here with the same wrong byte # layout set_scope_span used to have before it got fixed: # missing the VFO-select byte and encoding the full span # instead of span/2. Reusing the now-correct, ACK-checked # method instead of duplicating the fix in two places.) self.set_scope_span(20000) self.log(f"[civ] scope output {'ON' if on else 'OFF'} (baud={self.speed})" + (" + Center mode 20kHz" if on else "")) except Exception as e: self.log(f"[civ] _enable_scope error: {e}") def _write(self, payload: bytes): frame = bytes([0xFE, 0xFE, self.addr, CTRL_ADDR]) + payload + bytes([0xFD]) # NOTE (perf): removed the per-write log — polling calls write # every ~100ms (poll freq/smeter/scope), and every print is a # blocking syscall (~0.5-2ms). Under load the logs piled up and # blocked the reader/writer thread. To enable for debugging: # uncomment the line below. # self.log(f"[civ] TX: {frame.hex()}") with self._wlock: if self._ser: try: self._ser.write(frame) self._write_fail_count = 0 except Exception: # SELF-HEAL — live-tested simulated power loss: after the # radio's USB power drops and comes back, Windows leaves # the COM handle "stuck" (PermissionError/ERROR_BAD_COMMAND, # "Urzadzenie nie rozpoznaje polecenia" on every write, # while reads keep timing out silently instead of raising # — nothing ever noticed the port was dead). The only # existing recovery path was webapp.py's _device_watchdog, # which only checks once an HOUR — the radio stayed # unreachable that whole time. A few consecutive write # failures in a row means the port itself is dead (not # one bad command), so close it and hand off to # _start_sim()/_reconnect_loop (the SAME recovery path # the watchdog uses) immediately instead of waiting. self._write_fail_count = getattr(self, "_write_fail_count", 0) + 1 if self._write_fail_count >= 3: self._handle_dead_port() raise def _handle_dead_port(self): self._write_fail_count = 0 self.connected = False self.log("[civ] port unresponsive after repeated write errors — " "closing and reconnecting") try: if self._ser: self._ser.close() except Exception: pass self._ser = None self._start_sim() def write_bytes_raw(self, data: bytes): """Write raw bytes to the CI-V port without the FE FE addr ctrl ... FD wrapper. Used by civ_bridge for TCP clients that send whole CI-V frames themselves (Logger32, HRD, etc.). Bytes are passed through exactly as received. Protected by self._wlock to avoid colliding with commands from the server's own UI. """ if self.sim or not self._ser: return with self._wlock: try: self._ser.write(data) self._write_fail_count = 0 except Exception as e: self.log(f"[civ] write_bytes_raw error: {e}") # Same dead-port counter as _write() - a bridge client # (Logger32/HRD) polling over TCP would otherwise hit this # every time too and never recover on its own either. self._write_fail_count = getattr(self, "_write_fail_count", 0) + 1 if self._write_fail_count >= 3: self._handle_dead_port() def add_bridge_listener(self, callback): """Register a callback called for every chunk of bytes from the radio. Callback: fn(bytes). Used by civ_bridge to broadcast to TCP. Note: the callback is called from the _reader_loop thread, it must be thread-safe.""" with self._bridge_lock: if callback not in self._bridge_listeners: self._bridge_listeners.append(callback) def remove_bridge_listener(self, callback): with self._bridge_lock: if callback in self._bridge_listeners: self._bridge_listeners.remove(callback) def _transact(self, payload: bytes, expect, timeout: float, sub: int = None): """Send a command and wait for a response (a cmd code in 'expect' or ACK 0xFB/0xFA). sub: FIX — the whole "15 xx" meter family (S-meter=02, ALC=13, PWR/PO=11, SWR=12, VOLT=15) replies with the SAME top-level cmd 0x15, differentiated only by the first payload byte. The poller fires these back-to-back (all 5 during active PTT), each with its own short 0.25s timeout. Without 'sub', the dispatch in _handle_frame matched on cmd alone — so a LATE reply to one 15-xx query (e.g. ALC, delayed under heavy CI-V+audio load) could get consumed by whichever 15-xx transact happened to be waiting at that moment. The "stolen" transact got a payload with the wrong first byte (its own caller-side check, e.g. pp[0]==0x11, catches THAT and discards it) — but the reply that should have satisfied it is now gone too, so it also times out. Net effect live: ALC and PWR readings went stale/missing independently of each other, making them look completely uncorrelated when compared side by side (reported: "ALC 58%/PWR 0%" back to back with "ALC 3%/PWR 40%" — each half was a stale leftover from a DIFFERENT instant, not a real simultaneous pair). When 'sub' is given, only a frame whose payload[0] == sub can satisfy this wait; a mismatched 15-xx frame is left for whichever OTHER transact is actually waiting for it. """ if not self._ser: return None if isinstance(expect, int): expect = {expect} with self._txlock: return self._transact_locked(payload, expect, timeout, sub) def _transact_locked(self, payload: bytes, expect, timeout: float, sub: int = None): """Core of _transact(), assuming _txlock is ALREADY held by the caller. Exists so a caller that needs several transactions to happen as one atomic, uninterruptible unit (nothing else on the bus in between) can hold _txlock ONCE across all of them instead of once per transaction - see _read_freq_b_via_vfo_swap, which would otherwise risk another _transact() (the background poller, or a concurrent user freq/PTT action) landing exactly between its "switch to VFO-B" and "switch back to VFO-A" steps.""" self._resp_cmd = expect self._resp_sub = sub self._resp_payload = None self._resp_matched_cmd = None self._resp_ev.clear() try: self._write(payload) except Exception as e: self.log(f"[civ] write error: {e}") self._resp_cmd = None self._resp_sub = None return None got = self._resp_ev.wait(timeout) self._resp_cmd = None self._resp_sub = None # self._resp_matched_cmd is left set (not cleared here) so a # caller that cares which of 'expect' actually matched (e.g. # 0xFB ok vs 0xFA reject) can read it right after this call # returns - see set_scope_span for why this matters (this was # the ONLY CI-V "set" command in the file that never checked # for a response at all before this fix, so a rejected span # value looked identical to a silently-ignored one). return self._resp_payload if got else None def _read_freq_b_via_vfo_swap(self) -> int | None: """Fallback VFO-B (inactive VFO) frequency read for radios that don't support CI-V 0x25/0x26 at all (has_vfo_b_read=False in the model's civ_profiles.py entry - e.g. IC-746, confirmed live 2026-09-02 via Hamlib's own startup log: "icom_set_x25x26_ability: Hamlib thinks rig does not support x25/x26 command"). Same approach Hamlib itself falls back to for pre-0x25 radios: briefly select VFO-B (07 01), read its frequency with the normal 03 command, then switch back to VFO-A (07 00). SAFETY: never called while self.ptt is True (checked by the caller - webapp.py's _is_split_cross_band awaits this only when about to ALLOW a split TX, never during one). The whole 3-step sequence holds _txlock for its entire duration (via _transact_locked, not the public _transact) so no other CI-V transaction - the background poller's own freq read, or a concurrent user action - can land in between and either read VFO-B's frequency believing it's VFO-A's, or (worse) SET a new frequency onto VFO-B while the radio is momentarily parked there. The VFO-A restore step always runs (even if the VFO-B read failed) so the radio is never left parked on VFO-B. Returns the VFO-B frequency in Hz, or None if the radio didn't cooperate (caller must treat None as "still unknown" - fail safe, not silently trust a stale value). """ if self.ptt or not self._ser: return None self._tx_priority_active = True try: with self._txlock: sel = self._transact_locked(bytes([0x07, 0x01]), {0xFB, 0xFA}, 0.4) if sel is None or self._resp_matched_cmd == 0xFA: self.log(f"[civ] VFO-B swap-read: couldn't select VFO-B " f"({'no ACK' if sel is None else 'NG'})") return None fb = None try: fp = self._transact_locked(bytes([0x03]), {0x03, 0x00}, 0.4) if fp and len(fp) >= 4: fb = bcd_to_freq(fp[:5]) except Exception as e: self.log(f"[civ] VFO-B swap-read: freq read error: {e}") # ALWAYS restore VFO-A, even if the read above failed - # never leave the radio parked on VFO-B. back = self._transact_locked(bytes([0x07, 0x00]), {0xFB, 0xFA}, 0.4) if back is None or self._resp_matched_cmd == 0xFA: self.log("[civ] VFO-B swap-read: WARNING - couldn't switch back " "to VFO-A, radio may be left on VFO-B") # ALWAYS log the outcome (not just failures) - live-seen # 2026-09-02: with only failure logging, "no log line at # all" was ambiguous between "silently succeeded" and "ran # but hit an unlogged edge case", making the cross-band # guard impossible to verify from a live log alone. self.log(f"[civ] VFO-B swap-read: result={fb}Hz " f"({'no freq in response' if fb is None else 'OK'})") return fb finally: self._tx_priority_active = False async def get_freq_b_fresh(self) -> int | None: """Async wrapper for _read_freq_b_via_vfo_swap (see its docstring) - for webapp.py's cross-band-split safety check to call right before allowing a split TX on a radio with has_vfo_b_read=False. Not for routine/background polling - each call briefly swaps the radio's active VFO.""" return await asyncio.to_thread(self._read_freq_b_via_vfo_swap) def _reader_loop(self): buf = bytearray() while self._running: try: data = self._ser.read(512) except Exception as e: self.log(f"[civ] reader read error: {e}") # Same recovery path as _handle_dead_port() (repeated write # failures) - without this the reader thread just died here # forever, self.sim/self.connected never flipped, and # _reconnect_loop (which only runs while self.sim is True) # never started. _poller_loop kept running against the dead # self._ser, silently failing every transact() in a loop, # with no visible sign beyond one log line until a manual # restart. One bad read (e.g. a momentary USB/power drop) # must not be a permanent loss of radio control. self._handle_dead_port() break if not data: continue if self._rx_logged < 10: self.log(f"[civ] RX: {data.hex()}") self._rx_logged += 1 # Bridge: forward raw bytes to listeners (civ_bridge TCP clients). # This is done BEFORE parsing so TCP clients get exactly what # our own parsing logic gets (same events, same order). if self._bridge_listeners: with self._bridge_lock: listeners = list(self._bridge_listeners) for cb in listeners: try: cb(bytes(data)) except Exception as e: self.log(f"[civ] bridge listener error: {e}") buf += data # cut out FE FE ... FD frames while True: i = buf.find(b"\xfe\xfe") if i < 0: if len(buf) > 4096: del buf[:-2] break j = buf.find(b"\xfd", i + 2) if j < 0: if i > 0: del buf[:i] break frame = bytes(buf[i:j + 1]) del buf[:j + 1] try: self._handle_frame(frame) except Exception as e: self.log(f"[civ] handle_frame error: {e}") def _handle_frame(self, frame: bytes): if len(frame) < 6: return to, frm, cmd = frame[2], frame[3], frame[4] payload = frame[5:-1] # echo of our own command (to the radio) — ignore if to == self.addr: return # only accept frames addressed to the controller (0xE0) or transceive broadcast (0x00) if to not in (CTRL_ADDR, 0x00): return # SCOPE: 0x27 0x00 if cmd == 0x27 and len(payload) >= 1 and payload[0] == 0x00: # Scope frame counter - used by the scope-restart-after-power-ON # logic (webapp: _verify_radio_awake_and_start_scope checks # whether the counter is increasing = frames are flowing). self._scope_rx_count = getattr(self, "_scope_rx_count", 0) + 1 _hs_t0 = time.monotonic() self._handle_scope(payload[1:]) _hs_ms = (time.monotonic() - _hs_t0) * 1000.0 # Diagnostics: how long processing ONE scope frame takes. If # it's hundreds of ms, the culprit is _handle_scope; if it's # microseconds — the problem is in frame rate/core contention, # not the processing itself. if _hs_ms > 20: print(f"[scopetime] _handle_scope: {_hs_ms:.1f}ms", flush=True) return # frequency (response 0x03 or transceive 0x00) if cmd in (0x00, 0x03) and len(payload) >= 4: f = bcd_to_freq(payload[:5]) if f and abs(f - self.freq) >= 1: # If we recently (< 0.8s) set the frequency locally (e.g. # click-to-tune) and the radio returns a DIFFERENT value, # it's probably a "stale" reading that started before 0x05 # took effect — ignore it, so it doesn't overwrite the UI # with the old freq right after the click. grace = (time.time() - self._local_freq_set_at) < 0.8 if grace and self._local_freq_set_val is not None and f != self._local_freq_set_val: pass # ignore the stale reading else: # Protection against a corrupted BCD reading from the # radio (seen when a bridge client sent a bad frame - # the poller read freq=1 Hz and the UI showed # 0.000001 MHz). IC-7300 min freq = 30kHz, max = 74MHz. # We reject readings below 100kHz as invalid (very few # people use LF). This only guards against drastic # corruption, not subtle few-Hz errors. if f < 100_000: self.log(f"[civ] IGNORING absurd freq={f} Hz (BCD corruption?)") else: self.freq = f with self._telemetry_lock: self._telemetry_pending["freq"] = {"type": "freq", "freq": f} # mode (0x04 response / 0x01 transceive) — payload[0]=mode, # payload[1]=filter (01/02/03=FIL1/2/3, when the radio reports it) # # CRITICAL NOTE: payload[0] is ALWAYS the base CI-V mode (e.g. # USB=1), it NEVER carries DATA-mode information — that's a # separate command (1A 06), which the radio sends INDEPENDENTLY. # An earlier version of this handler did self.mode = nm DIRECTLY # from mode_map (which only knows base modes), which WIPED the # "-D" suffix every time the radio echoed/transceived its mode # (which happens often — including on every set_mode() we send # ourselves, since the IC-7300 sends back a confirmation through # the same transceive channel). Effect observed live: the UI # showed USB-D for a fraction of a second after clicking, and # polling every ~2s (rig_poll in server.py) reverted it back to # USB. Fix: append "-D" based on the ALREADY-TRACKED self.data_mode, # exactly the way set_mode does it when sending. elif cmd in (0x01, 0x04) and len(payload) >= 1: base_nm = self.mode_map.get(payload[0]) nm = f"{base_nm}-D" if (base_nm and self.data_mode) else base_nm if nm and nm != self.mode: self.mode = nm with self._telemetry_lock: self._telemetry_pending["mode"] = {"type": "mode", "mode": self.mode, "bandwidth": self.bw, "filterNum": self.filter_num} if len(payload) >= 2 and payload[1] in (1, 2, 3) and payload[1] != self.filter_num: self.filter_num = payload[1] with self._telemetry_lock: self._telemetry_pending["mode"] = {"type": "mode", "mode": self.mode, "bandwidth": self.bw, "filterNum": self.filter_num} # PTT elif cmd == 0x1C and len(payload) >= 2 and payload[0] == 0x00: self.ptt = bool(payload[1]) # forward to a waiting transaction — ONLY when the cmd code is # expected (set commands explicitly give {0xFB,0xFA}, so an # ACK/NAK won't falsely complete a freq/mode/S-meter read under # mixed CI-V traffic). When the waiter also gave 'sub' (the 15-xx # meter family — see _transact's docstring), the first payload # byte must match too, otherwise this frame belongs to a DIFFERENT # 15-xx query that's still in flight — leave it unconsumed so the # transact actually waiting for it can still catch it. ec = self._resp_cmd if ec and cmd in ec: es = self._resp_sub # An NG (0xFA) reply carries NO payload at all, so it can never # match a specific 'sub' (there's nothing to compare) - without # this, a real "Command rejected" response from the radio for a # 15-xx meter read (live-confirmed 2026-09-02 via rigctl on a # real IC-746: TX "15 13", RX "fa" NG) fell through to the # cross-talk branch below and looked identical to a plain # timeout (p=None) in our own logs. Safe to accept unconditionally # here because _txlock serializes _transact() - only ONE # query can ever be waiting at a time, so an ambiguous # payload-less frame can only belong to it. if es is None or not payload or payload[0] == es: self._resp_payload = payload self._resp_matched_cmd = cmd self._resp_ev.set() elif getattr(self, "_crosstalk_logged", 0) < 20: # DIAGNOSTIC: proves live whether 15-xx cross-talk actually # happens (see _transact docstring) — first 20 occurrences only. self._crosstalk_logged = getattr(self, "_crosstalk_logged", 0) + 1 self.log(f"[civ] 15-xx cross-talk avoided: waiting for sub=0x{es:02X}, " f"got 0x{payload[0]:02X} (belongs to another in-flight query)") def _handle_scope(self, d: bytes): if self._scope_logged < 4: self.log(f"[scope] raw frame ({len(d)}B): {d.hex()}") self._scope_logged += 1 if len(d) < 3: return seq = d[1] total = d[2] if seq <= 1: # First frame: full header per CI-V Ref. p.19-14 # d[3] = spectrum scope mode (00=Center,01=Fixed,02=SCROLL-C,03=SCROLL-F) # d[4:9] = center freq (Center mode) OR lower edge (Fixed/SCROLL) — 5B BCD # d[9:14] = span (Center mode, 5B "BCD-like" per the SPAN table) # OR higher edge (Fixed/SCROLL) — 5B BCD # d[14] = out-of-range (00=in range, 01=out of range) # d[15:] = waveform data (only when total==1 and oor==00, or # skipped in the 1st frame when total>1 — see the note in the docs) self._scope_mode_code = d[3] if len(d) > 3 else 0 self._scope_oor = d[14] if len(d) > 14 else 0 if self._scope_mode_code == 0x00: # Center mode: center freq + span try: self._scope_center = bcd_to_freq(d[4:9]) except Exception: self._scope_center = self.freq try: span_raw = bcd_to_freq(d[9:14]) self._scope_span_hz = span_raw if span_raw else self._scope_span_hz except Exception: pass self._scope_lo = self._scope_center - self._scope_span_hz // 2 self._scope_hi = self._scope_center + self._scope_span_hz // 2 else: # Fixed / SCROLL-C / SCROLL-F: lower + higher edge try: self._scope_lo = bcd_to_freq(d[4:9]) self._scope_hi = bcd_to_freq(d[9:14]) self._scope_center = (self._scope_lo + self._scope_hi) // 2 self._scope_span_hz = self._scope_hi - self._scope_lo except Exception: pass hdr_len = self.scope_header_len data = d[hdr_len:] if len(d) > hdr_len else b"" self._scope_acc = bytearray(data) self._scope_total = total else: self._scope_acc += d[3:] if total and seq >= total and self._scope_acc: # Rate limit: max 20 frames/s (50ms interval). The IC-7300 # sends the scope at ~30-60fps but the WS client doesn't need # that much - 20fps is a smooth waterfall + half the # asyncio/JSON/WebSocket load. Note: this is OPTIONAL, it # helps a lot with many clients. now = time.time() if (now - self._scope_last) < 0.050: # Too early - drop this frame, wait for the next one. # We do NOT reset _scope_acc so accumulation keeps going. # But it needs to start fresh on the next packet: self._scope_acc = bytearray() self._scope_total = 0 return smax = self.scope_max # Scaling scope data to 0-255. Used to be: a Python loop over # ~475 spectrum points, called a DOZEN-PLUS times per second — # this is what was blocking the event loop for 100-250ms # (confirmed: lag grows with the number of scope frames). numpy # does the same thing vectorized, ~100x faster. _raw = np.frombuffer(bytes(self._scope_acc), dtype=np.uint8) arr = np.minimum(255, (_raw.astype(np.uint32) * 255) // max(smax, 1)).astype(np.uint8).tolist() # In Center mode (scopeMode==0) we use self.freq (from the # 0x03 poller, confirmed accurate) as centerHz — instead of the # value from the scope header (_scope_center), which in # testing showed a discrepancy against the radio's actual # frequency (up to a few kHz, possibly a BCD field parsing # artifact in this particular firmware). Lo/Hi are recomputed # relative to self.freq and the span from the header, so the # frequency axis and click-to-tune stay consistent with the # actual VFO. if self._scope_mode_code == 0x00: center = self.freq lo = center - self._scope_span_hz // 2 hi = center + self._scope_span_hz // 2 else: center = self._scope_center lo, hi = self._scope_lo, self._scope_hi # DECOUPLING the scope from the reader thread. # # Previously the reader called broadcast_sync # (run_coroutine_threadsafe) on every frame — this thread->loop # sync, done a dozen-plus times per second, collided with the # asyncio loop and blocked it (confirmed: lag grows linearly # with the number of frames, while processing ONE frame is # fast). Now the reader ONLY writes the freshest complete frame # to a variable — zero synchronization with the loop. A # separate, lightweight asyncio loop (_scope_pump in webapp) # reads that variable and sends it at a steady 15fps. The # reader hands off the data and goes back to reading the port, # without waiting for the loop. self._latest_scope = { "type": "scope_frame", "data": arr, "source": "radio", "centerHz": center, "spanHz": self._scope_span_hz, "loHz": lo, "hiHz": hi, "mode": self.mode, "dataMode": self.data_mode, "filterHz": self._filter_width_hz, "scopeMode": self._scope_mode_code, "outOfRange": bool(self._scope_oor), } self._scope_acc = bytearray() self._scope_last = time.time() return def _poller_loop(self): """Periodically poll freq/mode/S-meter/filter (responses update state in the reader).""" n = 0 _off_logged = False while self._running and self._ser: try: # A powered-off radio (power OFF via CI-V) doesn't respond # to any transactions. Without this check every loop cycle # ended in a read error and spammed the log ("S-metr # transact fail" repeatedly). if not self.powered: if not _off_logged: self.log("[civ] Radio off — telemetry paused") _off_logged = True time.sleep(1.0) continue if _off_logged: self.log("[civ] Radio on — telemetry resumed") _off_logged = False # Yield the bus entirely to an in-flight CW/SSB PTT # command - see the comment on self._tx_priority_active in # __init__. Skip this whole cycle (no freq/mode/meter # polling at all) rather than trying to interleave. if self._tx_priority_active: time.sleep(0.02) continue self._transact(bytes([0x03]), {0x03, 0x00}, 0.3) # freq if n % 4 == 0: self._transact(bytes([0x04]), {0x04, 0x01}, 0.3) # mode # VFO B (inactive VFO): CI-V 25 01 -> response [01, 5B freq BCD] # (per doc p.19-13: "Selected or unselected VFO frequency # settings", 00=selected/active, 01=unselected). Polled # every ~1.2s — changes rarely (split/A-B swap). Older # Icoms (e.g. IC-746, see civ_profiles.py's # "has_vfo_b_read") don't support 0x25/0x26 at all - radio # NGs it every time, so skip the query entirely instead of # spamming the bus/log with a command we already know fails. if n % 4 == 2 and self.profile.get("has_vfo_b_read", True): bp = self._transact(bytes([0x25, 0x01]), {0x25}, 0.3) if bp and len(bp) >= 6 and bp[0] == 0x01: fb = bcd_to_freq(bp[1:6]) if fb and abs(fb - self.freq_b) >= 1: self.freq_b = fb with self._telemetry_lock: self._telemetry_pending["freqB"] = {"type": "freqB", "freqB": fb} # Split status: CI-V 0F with NO data byte = query (WITH a # data byte = set, see set_split() above) -> response 0F # [00=off, 01=on]. FIX (reported live 2026-08-24: "button # split nie aktualizuje sie podswietleniem i rzeczywistym # odczytem"): split was previously only ever SENT, never # polled - our own self.split (and the button highlight # every client derives from it) only changed when a client # clicked the button itself. If split got toggled on the # radio's own front panel, or missed by a client, nothing # ever corrected the drift. Same class of bug as the CW # stale-mode fix earlier today. Polled at the same ~1.2s # cadence as freq_b - changes rarely, no need for every cycle. if n % 4 == 1: sp = self._transact(bytes([0x0F]), {0x0F}, 0.3) if sp and len(sp) >= 1: is_split = bool(sp[0]) if is_split != self.split: self.split = is_split with self._telemetry_lock: self._telemetry_pending["split"] = {"type": "split", "split": self.split, "freqB": self.freq_b} # S-meter: 0x15 0x02 -> response 0x15 ... p = self._transact(bytes([0x15, 0x02]), {0x15}, 0.3, sub=0x02) if p and len(p) >= 3 and p[0] == 0x02: raw = bcd2(p[1:3]) lvl = smeter_to_sunit(raw) if abs(lvl - self.s_meter) > 0.2: self.s_meter = lvl # The frontend (ws.js) listens for msg.value - NOT msg.smeter! with self._telemetry_lock: self._telemetry_pending["smeter"] = {"type": "smeter", "value": round(lvl, 1)} elif getattr(self, '_smeter_fail_logged', 0) < 5: self.log(f"[civ] S-metr transact fail: p={p.hex() if p else None}") self._smeter_fail_logged = getattr(self, '_smeter_fail_logged', 0) + 1 # TX Meters: ALC (15 13), PWR (15 11), SWR (15 12), VOLT (15 15) # Polled only while PTT is active (or when the meter is # selected by the user) # 2-byte BCD -> values (calibration points from the # official IC-7300MK2 CI-V Reference Guide, see the # comments at each meter below): # ALC: 0..120 -> 0..100% (linear) # PWR: 0=0%, 143=50%, 213=100% (nonlinear) # SWR: 0=1.0, 48=1.5, 80=2.0, 120=3.0 (nonlinear) # VOLT: 0=0V, 13=10V, 241=16V (strongly nonlinear) if self.ptt or n % 8 == 0: _dbg_alc = None _dbg_pwr = None _dbg_alc_raw = None _dbg_pwr_raw = None _dbg_swr = None _dbg_swr_raw = None _dbg_volt = None _dbg_volt_raw = None _dbg_rfpower_raw = None # ALC. FIX: command "15 11" used to be used here, which # per the official IC-7300 CI-V documentation is # actually "PO" (output power), NOT ALC — the ALC/PWR # labels in the code were swapped relative to the # actual commands. The correct command for ALC is # "15 13", with a documented range of 0000=Min to # 0120=Max (NOT 0-241). ap = self._transact(bytes([0x15, 0x13]), {0x15, 0xFA}, 0.25, sub=0x13) if ap and len(ap) >= 3 and ap[0] == 0x13: alc_raw = bcd2(ap[1:3]) alc_pct = min(100.0, max(0.0, alc_raw / 120 * 100)) _dbg_alc = alc_pct _dbg_alc_raw = alc_raw if alc_pct > self._alc_peak: self._alc_peak = alc_pct self._alc_peak_raw = alc_raw with self._telemetry_lock: self._telemetry_pending["txmeter_ALC"] = { "type": "txmeter", "meter": "ALC", "raw": alc_raw, "value": round(alc_pct, 1), "pct": min(1.0, alc_raw / 120), "peak": round(self._alc_peak, 1), "peakRaw": self._alc_peak_raw} elif getattr(self, '_alc_fail_logged', 0) < 5: _why = "REJECTED by radio (NG 0xFA)" if self._resp_matched_cmd == 0xFA else "timeout, no response" self.log(f"[civ] ALC transact fail ({_why}): p={ap.hex() if ap else None}") self._alc_fail_logged = getattr(self, '_alc_fail_logged', 0) + 1 # PWR output. FIX: command "15 14" used to be used here # (that's actually COMP — the speech compressor in dB, # NOT power). The correct command for PO (output power) # is "15 11", with a documented NONLINEAR scale: # 0000=0%, 0143=50%, 0213=100% (full scale reached # already at raw=213, not 241) — piecewise # interpolation between these points. pp = self._transact(bytes([0x15, 0x11]), {0x15, 0xFA}, 0.25, sub=0x11) if pp and len(pp) >= 3 and pp[0] == 0x11: pwr_raw = bcd2(pp[1:3]) _po_points = [(0, 0.0), (143, 50.0), (213, 100.0)] if pwr_raw <= _po_points[0][0]: pwr_pct = _po_points[0][1] elif pwr_raw >= _po_points[-1][0]: pwr_pct = _po_points[-1][1] else: pwr_pct = _po_points[-1][1] for _i in range(len(_po_points) - 1): _x0, _y0 = _po_points[_i] _x1, _y1 = _po_points[_i + 1] if _x0 <= pwr_raw <= _x1: _t = (pwr_raw - _x0) / (_x1 - _x0) pwr_pct = round(_y0 + _t * (_y1 - _y0), 1) break _dbg_pwr = pwr_pct _dbg_pwr_raw = pwr_raw if pwr_pct > self._pwr_peak: self._pwr_peak = pwr_pct self._pwr_peak_raw = pwr_raw with self._telemetry_lock: self._telemetry_pending["txmeter_PWR"] = { "type": "txmeter", "meter": "PWR", "raw": pwr_raw, "value": pwr_pct, "pct": pwr_pct / 100, "peak": round(self._pwr_peak, 1), "peakRaw": self._pwr_peak_raw} elif getattr(self, '_pwr_fail_logged', 0) < 5: _why = "REJECTED by radio (NG 0xFA)" if self._resp_matched_cmd == 0xFA else "timeout, no response" self.log(f"[civ] PWR transact fail ({_why}): p={pp.hex() if pp else None}") self._pwr_fail_logged = getattr(self, '_pwr_fail_logged', 0) + 1 # SWR sp = self._transact(bytes([0x15, 0x12]), {0x15, 0xFA}, 0.25, sub=0x12) if sp and len(sp) >= 3 and sp[0] == 0x12: swr_raw = bcd2(sp[1:3]) # IC-7300 SWR: 0=1.0, 48=1.5, 80=2.0, 120=3.0, 241=50 # FIX: the previous formula was linear over the # WHOLE 0-241 range, which didn't match the # calibration points above from the IC-7300 # documentation (SWR rises quickly to 3.0 already # at half the raw=120 range, then much more slowly # to 50 at raw=241) — this gave heavily inflated # values (e.g. raw=48 -> 10.76 instead of the # documented 1.5). Piecewise interpolation between # the actual calibration points. _swr_points = [(0, 1.0), (48, 1.5), (80, 2.0), (120, 3.0), (241, 50.0)] swr_val = _swr_points[-1][1] for _i in range(len(_swr_points) - 1): _x0, _y0 = _swr_points[_i] _x1, _y1 = _swr_points[_i + 1] if _x0 <= swr_raw <= _x1: _t = (swr_raw - _x0) / (_x1 - _x0) swr_val = round(_y0 + _t * (_y1 - _y0), 2) break _dbg_swr = swr_val _dbg_swr_raw = swr_raw with self._telemetry_lock: self._telemetry_pending["txmeter_SWR"] = { "type": "txmeter", "meter": "SWR", "raw": swr_raw, "value": swr_val, "pct": min(1.0, swr_raw / 120)} # scale to SWR=3 elif getattr(self, '_swr_fail_logged', 0) < 5: _why = "REJECTED by radio (NG 0xFA)" if self._resp_matched_cmd == 0xFA else "timeout, no response" self.log(f"[civ] SWR transact fail ({_why}): p={sp.hex() if sp else None}") self._swr_fail_logged = getattr(self, '_swr_fail_logged', 0) + 1 # Supply voltage (Vd — drain voltage / power amp supply # voltage). Command "15 15" (NOT "15 16" — that's Id, # current, a different meter). Calibration verified # directly against the official IC-7300MK2 CI-V # Reference Guide (icomuk.co.uk), command table, entry # "15 15": 0000=0V, 0013=10V, 0241=16V. # FIX: the previous points (0=0V, 151=10V, 211=16V) were # wrong — the scale is actually strongly nonlinear (the # first 10V is only 13 raw units, the rest, 10-16V, # stretches over the remaining 228 units, since that's # the range where the 12-13.8V supply actually # operates). The wrong points gave, e.g. for a real # 13.8V (raw~157), a reading of ~10.6V — exactly the # kind of low reading that was reported live. vp = self._transact(bytes([0x15, 0x15]), {0x15, 0xFA}, 0.25, sub=0x15) if vp and len(vp) >= 3 and vp[0] == 0x15: v_raw = bcd2(vp[1:3]) _vd_points = [(0, 0.0), (13, 10.0), (241, 16.0)] if v_raw <= _vd_points[0][0]: volt = _vd_points[0][1] elif v_raw >= _vd_points[-1][0]: volt = _vd_points[-1][1] else: volt = _vd_points[-1][1] for _i in range(len(_vd_points) - 1): _x0, _y0 = _vd_points[_i] _x1, _y1 = _vd_points[_i + 1] if _x0 <= v_raw <= _x1: _t = (v_raw - _x0) / (_x1 - _x0) volt = round(_y0 + _t * (_y1 - _y0), 2) break _dbg_volt = volt _dbg_volt_raw = v_raw with self._telemetry_lock: self._telemetry_pending["txmeter_VOLT"] = { "type": "txmeter", "meter": "VOLT", "raw": v_raw, "value": volt, "pct": volt / 16.0} # scale to nominal 16V elif getattr(self, '_volt_fail_logged', 0) < 5: # Not necessarily a bug - VOLT (15 15) isn't listed # in every model's CI-V command set (e.g. Hamlib's # own IC-746 caps don't include it) - an older radio # may simply not have this meter at all. _why = "REJECTED by radio (NG 0xFA)" if self._resp_matched_cmd == 0xFA else "timeout, no response" self.log(f"[civ] VOLT transact fail ({_why}, may be unsupported on this model): p={vp.hex() if vp else None}") self._volt_fail_logged = getattr(self, '_volt_fail_logged', 0) + 1 # Supply current (Id — drain current draw). Command # "15 16" (the OTHER meter in the same pair as VOLT/"15 # 15" above). Calibration points straight from the # official IC-7300MK2 CI-V Reference Guide, command # table entry "15 16": 0000=0, 0097=10, 0146=15, 0241=25 # (Amps, same nonlinear-piecewise style as VOLT/SWR). idp = self._transact(bytes([0x15, 0x16]), {0x15, 0xFA}, 0.25, sub=0x16) if idp and len(idp) >= 3 and idp[0] == 0x16: id_raw = bcd2(idp[1:3]) _id_points = [(0, 0.0), (97, 10.0), (146, 15.0), (241, 25.0)] amps = _id_points[-1][1] for _i in range(len(_id_points) - 1): _x0, _y0 = _id_points[_i] _x1, _y1 = _id_points[_i + 1] if _x0 <= id_raw <= _x1: _t = (id_raw - _x0) / (_x1 - _x0) amps = round(_y0 + _t * (_y1 - _y0), 2) break with self._telemetry_lock: self._telemetry_pending["txmeter_ID"] = { "type": "txmeter", "meter": "ID", "raw": id_raw, "value": amps, "pct": amps / 25.0} # scale to nominal 25A elif getattr(self, '_id_fail_logged', 0) < 5: # Same "may simply not exist on this model" caveat as VOLT above. _why = "REJECTED by radio (NG 0xFA)" if self._resp_matched_cmd == 0xFA else "timeout, no response" self.log(f"[civ] ID transact fail ({_why}, may be unsupported on this model): p={idp.hex() if idp else None}") self._id_fail_logged = getattr(self, '_id_fail_logged', 0) + 1 # OVF (ADC overflow / front-end overdrive indicator). # Command "15 07": 00=Not displayed, 01=Displayed. Pure # boolean status, no calibration needed - the radio # itself decides when the receiver front-end is being # overdriven, which a remote operator has no other way # to see (no physical panel to glance at). ovp = self._transact(bytes([0x15, 0x07]), {0x15, 0xFA}, 0.25, sub=0x07) if ovp and len(ovp) >= 2 and ovp[0] == 0x07: ovf = bool(ovp[1]) if ovf != getattr(self, "ovf", False): self.ovf = ovf self.bcast({"type": "ovf", "value": ovf}) elif getattr(self, '_ovf_fail_logged', 0) < 5: _why = "REJECTED by radio (NG 0xFA)" if self._resp_matched_cmd == 0xFA else "timeout, no response" self.log(f"[civ] OVF transact fail ({_why}, may be unsupported on this model): p={ovp.hex() if ovp else None}") self._ovf_fail_logged = getattr(self, '_ovf_fail_logged', 0) + 1 # RFPOWER SET level (14 0A) — direct read-back of the # radio's CONFIGURED power ceiling, independent of the # UI slider display. DIAGNOSTIC: with PO/PWR staying # flat around ~25% across a big txVolume sweep # (0.7x-1.5x) and ALC pinned at 0 the whole time, one # remaining explanation is that the "100%" the UI shows # never actually reached raw=255 on the radio itself — # this confirms or rules that out directly instead of # guessing again. rfp = self._transact(bytes([0x14, 0x0A]), {0x14}, 0.25, sub=0x0A) if rfp and len(rfp) >= 3 and rfp[0] == 0x0A: _dbg_rfpower_raw = bcd2(rfp[1:3]) if VERBOSE and self.ptt and n % 4 == 0: self.log(f"[txmeter] TX read: ALC={_dbg_alc if _dbg_alc is not None else '?'}% " f"(raw={_dbg_alc_raw}, peak={self._alc_peak:.1f}% raw={self._alc_peak_raw}) " f"PWR={_dbg_pwr}% (raw={_dbg_pwr_raw}, peak={self._pwr_peak:.1f}% raw={self._pwr_peak_raw}) " f"SWR={_dbg_swr} (raw={_dbg_swr_raw}) VOLT={_dbg_volt} (raw={_dbg_volt_raw}) " f"RFPOWER_set_raw={_dbg_rfpower_raw} (0-255, 255=100%) " f"freq={self.freq/1e6:.4f}MHz mode={self.mode} (bcast sent)") # Filter width: 1A 03 -> response [03, idx_bcd_2B] (00..49) # DATA mode: 1A 06 -> response [06, data_mode(0/1), filter(1-3)] # (per doc p.19-10) — affects the interpretation of # idx>31 in _SSB_FILTER_TABLE (SSB-D range up to 3600Hz, # DATA mode only). # Polled less often (every ~1.2s) — both values rarely change if n % 4 == 1: # FIX: same "shared response cmd, no sub-command check" # bug as the 15-xx meter family (see _transact's # docstring) — data mode (1A 06) and filter width # (1A 03) BOTH reply with cmd 0x1A. A late/crossed # response here is worse than a stale meter reading: if # a stale "1A 06" answer (e.g. from BEFORE the radio was # ever switched into DATA mode) gets consumed here, the # code below actively WRITES data_mode=False back and # strips "-D" from self.mode — silently kicking a radio # that IS in USB-D back to being TRACKED as plain USB # (symptom reported live: mode=USB in the FT8 TX log, # PWR stuck low regardless of drive level, because plain # USB takes modulation from the physical MIC input, not # from our USB audio at all). dp = self._transact(bytes([0x1A, 0x06]), {0x1A}, 0.3, sub=0x06) if dp and len(dp) >= 2 and dp[0] == 0x06: new_dm = bool(dp[1]) if new_dm != self.data_mode: self.data_mode = new_dm # CRITICAL: also update self.mode (the -D # suffix). Without this, a radio in USB-D had # self.mode="USB" -> the UI held "USB" -> every # t='mode' from the UI (e.g. a filter change) # did set_mode("USB") -> 1A 06 00 -> TURNED OFF # data mode in the radio. Symptom: "keeps # switching from USB-D back to USB in a digital mode". base = self.mode[:-2] if self.mode.endswith("-D") else self.mode if base in ("USB", "LSB"): self.mode = f"{base}-D" if new_dm else base self.bcast({"type": "mode", "mode": self.mode, "bandwidth": self.bw}) self._filter_width_hz = filter_width_hz(self.mode, self._filter_idx, self.data_mode) self.bcast({"type": "filter_width", "hz": self._filter_width_hz, "idx": self._filter_idx, "mode": self.mode, "dataMode": self.data_mode}) fp = self._transact(bytes([0x1A, 0x03]), {0x1A}, 0.3, sub=0x03) # FIX: the response is [0x03, idx_bcd_1B] - ONE byte # holding a 2-digit BCD index (00..49), not two bytes - # confirmed live (fp.hex()=='0336', i.e. only 2 bytes # total). The old code required len(fp)>=3 and read a # 2-byte BCD via bcd2(fp[1:3]) - always failed the # length check, so _filter_idx/_filter_width_hz never # updated past their __init__ defaults (2400Hz) no # matter what filter was actually selected on the radio. if fp and len(fp) >= 2 and fp[0] == 0x03: idx = (fp[1] >> 4) * 10 + (fp[1] & 0x0F) if idx != self._filter_idx: self._filter_idx = idx self._filter_width_hz = filter_width_hz(self.mode, idx, self.data_mode) self.bcast({"type": "filter_width", "hz": self._filter_width_hz, "idx": idx, "mode": self.mode, "dataMode": self.data_mode}) elif getattr(self, "_filter_poll_fail_logged", 0) < 10: self._filter_poll_fail_logged = getattr(self, "_filter_poll_fail_logged", 0) + 1 self.log(f"[civ] filter poll FAILED: fp={fp.hex() if fp else None}") # Round-robin read of the Set Level sliders (CI-V 14 ) — # one level per cycle (~0.3s), a full sweep of 14 levels # takes ~4.2s. Lets the UI notice changes made manually on # the radio's front panel (e.g. turning the AF/RF knob on # the IC-7300). level_items = list(self.profile.get("levels", {}).items()) if level_items: name, lvl = level_items[n % len(level_items)] new_val = self._read_level(name, lvl) if new_val is not None: old_val = self.level_values.get(name) if old_val is None or abs(new_val - old_val) > 1e-6: self.level_values[name] = new_val self.bcast({"type": "level_value", "id": f"level_{name.lower()}", "value": new_val}) # Preamp (16 02 -> response [02, 00|01|02]) and Attenuator # (11 -> response [00|20]) — polled every ~1.2s (rarely # change, usually set manually on the radio's panel). if n % 4 == 3: pp = self._transact(bytes([0x16, 0x02]), {0x16}, 0.3) if pp and len(pp) >= 2 and pp[0] == 0x02: new_pre = pp[1] if new_pre in (0, 1, 2) and new_pre != self.preamp: self.preamp = new_pre self.bcast({"type": "preamp", "value": new_pre}) ap = self._transact(bytes([0x11]), {0x11}, 0.3) if ap and len(ap) >= 1: new_att = (ap[0] != 0x00) if new_att != self.attenuator: self.attenuator = new_att self.bcast({"type": "attenuator", "value": new_att}) # Tuner ON/OFF: 1C 01 -> response [01, 00|01]. We don't # poll the "START" state (1C 01 02 is a one-shot # command, the radio doesn't report it as a persistent # state). tp = self._transact(bytes([0x1C, 0x01]), {0x1C}, 0.3) if tp and len(tp) >= 2 and tp[0] == 0x01: new_tuner = (tp[1] != 0x00) if new_tuner != self.tuner: self.tuner = new_tuner self.bcast({"type": "tuner", "value": new_tuner}) except Exception as e: self.log(f"[civ] poller error: {e}") n += 1 time.sleep(0.3) # ════════════════════════════════════════════════════════════════════════ # SIMULATION (no hardware / Replit) # ════════════════════════════════════════════════════════════════════════ def _start_sim(self) -> bool: self.sim = True self.connected = False if self._disconnected_since is None: self._disconnected_since = time.time() self.bcast({"type": "rig_disconnected", "since": self._disconnected_since}) self.scope_running = True if not self.last_msg: self.last_msg = "Tryb bezposredni CI-V — SYMULACJA scope (brak sprzetu COM)" self._running = True if not (self._sim_th and self._sim_th.is_alive()): self._sim_th = threading.Thread(target=self._sim_loop, daemon=True) self._sim_th.start() self.log("[civ] Scope SIMULATION started") # Background reconnect-attempt thread (e.g. the radio was off when # the server started, the user will power it on later physically # or via the Power button after an earlier wakeup) if not (self._reconnect_th and self._reconnect_th.is_alive()): self._reconnect_th = threading.Thread(target=self._reconnect_loop, daemon=True) self._reconnect_th.start() return False def _reconnect_loop(self): """ Every RECONNECT_INTERVAL seconds (in simulation mode) try to reopen the port and poll 0x03. If the radio responds — switch from simulation to a real connection (analogous to a successful _open()). """ RECONNECT_INTERVAL = 10.0 while self._running and self.sim: time.sleep(RECONNECT_INTERVAL) if not self.sim or not self._running: return # someone already connected manually / it's been closed if not HAS_SERIAL: continue try: test_ser = serial.Serial(self.port, self.speed, timeout=0.3, write_timeout=1.0) except Exception: continue # port still unavailable — try again later try: # Send 0x03 (get freq) and check the response frame = bytes([0xFE, 0xFE, self.addr, CTRL_ADDR, 0x03, 0xFD]) test_ser.write(frame) time.sleep(0.3) resp = test_ser.read(64) test_ser.close() except Exception: try: test_ser.close() except: pass continue if not resp or 0xFD not in resp: continue # no response — radio not ready yet self.log("[civ] Reconnect: radio responded — switching out of simulation") # Stop the simulation, open a real connection self.sim = False self._running = False # stops _sim_loop time.sleep(0.2) self._running = True self._open() return # _open() either connects for real, or falls back to # _start_sim (which spawns a new _reconnect_th if still in sim) def _sim_loop(self): N = 320 t = 0.0 # a few "signals" that drift across the band peaks = [{"pos": random.uniform(0.15, 0.85), "amp": random.uniform(0.5, 1.0), "w": random.uniform(0.004, 0.02), "drift": random.uniform(-0.0008, 0.0008)} for _ in range(4)] # STABILITY: try INSIDE the loop. Without this, a single error in # bcast() (e.g. the hub in a bad state while disconnecting clients) # kills the simulation thread permanently - the waterfall in SIM # mode stops working until the server restarts. _errs = 0 while self._running and self.sim: try: arr = [] noise_base = 18 + 6 * math.sin(t * 0.7) for i in range(N): x = i / N v = noise_base + random.uniform(0, 10) for pk in peaks: d = x - pk["pos"] v += pk["amp"] * 230 * math.exp(-(d * d) / (2 * pk["w"])) arr.append(max(0, min(255, int(v)))) for pk in peaks: pk["pos"] += pk["drift"] if pk["pos"] < 0.1 or pk["pos"] > 0.9: pk["drift"] *= -1 pk["amp"] = max(0.25, min(1.0, pk["amp"] + random.uniform(-0.05, 0.05))) self.bcast({"type": "scope_frame", "data": arr, "source": "sim", "centerHz": self.freq, "spanHz": self._scope_span_hz, "loHz": self.freq - self._scope_span_hz // 2, "hiHz": self.freq + self._scope_span_hz // 2, "mode": self.mode, "dataMode": self.data_mode, "filterHz": self._filter_width_hz, "scopeMode": self._scope_mode_code, "outOfRange": False}) _errs = 0 except Exception as e: _errs += 1 if _errs <= 3: print(f"[civ] _sim_loop error ({_errs}): {e}", flush=True) elif _errs == 4: print("[civ] _sim_loop: further errors suppressed in the log", flush=True) time.sleep(0.5) # backoff to avoid flooding the log t += 0.08 time.sleep(0.08)