flex_sim.py

#!/usr/bin/env python3
#
# flex-sim — synthetic FlexRadio-6000 emulator / test bench for AetherSDR.
# Copyright (C) 2026 Nigel Fenton (G0JKN)
#
# This program is free software: you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation, either version 3 of the License, or (at your option) any later
# version. Distributed WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public
# License for more details. You should have received a copy of the GNU GPL along
# with this program; if not, see <https://www.gnu.org/licenses/>.
#
# Code generated by Claude (Anthropic) via Claude Code, directed and tested by
# the copyright holder.
"""flex-sim — a synthetic FlexRadio-6000 emulator for testing AetherSDR.

flex-sim impersonates a FlexRadio 6000's data plane: it answers SmartSDR
discovery + the FlexLib control handshake, then streams synthetic VITA-49
panadapter / waterfall / meter data so AetherSDR renders a waterfall, S-meter and
TX meters from a programmable signal engine — a deterministic, hardware-free test
bench for AE's spectrum / waterfall / CW code. (Wire format reverse-engineered
from AE's own decoder; see PROTOCOL.md.)

It serves a live browser control panel (default http://<ip>:8731/) with a pattern
picker, dBm/S-unit level controls, TX keying and CW/CWX — each pattern shows what
it exercises in AE.

NETWORKING: by default flex-sim binds UDP :4992 like a real radio, so give it its
own IP (second machine / VM / container / WSL2) — OR use --port to run it on the SAME
host as AE on a different control port. Pass --ae <AE-ip> to also unicast discovery to AE.

Pure Python 3.8+ stdlib — no dependencies.

Usage:  python3 flex_sim.py [--ip IP] [--ae AE_IP] [--pattern NAME] [--port P]
                            [--bins N] [--fps N] [--width-khz K] [--ctl-port P]
  --port lets flex-sim run on the SAME host as AE: it binds/advertises that port
  (e.g. 5992) for control+data while still announcing to AE's fixed :4992 listener.
Patterns (each is a test signal; the control panel explains what to look for):
  noise_floor   flat floor (baseline; auto-black #3586)
  ramp          whole-span level swept min->max (auto-black / level mapping)
  cal_tones     fixed carriers at known dBm (verify the dB scale)
  carrier       steady tone on the VFO (S-meter / dBm calibration)
  cw            20-wpm Morse 'CQ CQ CQ TST' (sim-keyed)
  swept_carrier one tone sweeping the span (freq mapping / tile decode)
  comb          N evenly-spaced tones (dynamic range)
  two_tone      two equal tones spaced 2 kHz about the VFO (calibrated ruler / linearity / IMD)
  step          whole-span square wave in time (temporal response)
  impulse       brief full-span flash each period (transient response)
  staircase     center carrier stepping floor->max (colormap ladder)
  test_card     17 carriers S1..S9 (1/2-S steps) on a steppable noise floor (auto-black)
  noise         random noise band, white->pink tilt
  noise_cal     flat noise at EXACT dBm for filter testing (output = filter response); width narrows -> band-limited
  tx_blank      periodic real TX gap -> repro #2126 / #1916
Also handles AE's own CWX keyer (cwx send/wpm/qsk_enabled) for authentic CW TX.
"""
import argparse, http.server, json, math, os, glob, random, socket, struct, subprocess, sys, threading, time, urllib.parse, uuid, wave

HERE = os.path.dirname(os.path.abspath(__file__))
FIXTURES_DIR = os.path.join(HERE, "fixtures")
REPORTS_DIR  = os.path.join(HERE, "reports")

FLEX_SIM_VERSION = "0.2.0"

DISCOVERY_PORT = 4992        # AE always listens for discovery here (fixed); we broadcast TO it
DEFAULT_PORT = 4992          # flex-sim's own control (TCP) + VITA-prime (UDP) port; override with --port
def _default_serial_prefix():
    """Serial prefix defaults to THIS MACHINE'S hostname.

    AE matches `connect local serial <s>`, so two sims sharing a prefix are
    indistinguishable to that verb and AE silently takes whichever it
    discovered first — a real wrong-radio bug on a network with more than one
    sim. Deriving from the hostname makes the collision structurally impossible
    rather than merely avoidable-if-you-remember-the-flag: every box is
    distinct for free, including ones nobody has thought of yet.

    Sanitised to A-Z0-9 (a serial travels in a discovery string and a
    `chassis_serial=` reply; punctuation has no business in either) and capped
    so the `:02d` suffix keeps rack serials readable. Falls back to the old
    FLEXSIM constant if the hostname is unusable, so this can never leave a
    radio without an identity.
    """
    try:
        raw = "".join(c for c in socket.gethostname().upper() if c.isalnum())
    except Exception:
        raw = ""
    return raw[:12] or "FLEXSIM"


SERIAL_PREFIX = _default_serial_prefix()   # serial = <prefix><radio_id:02d>; --serial overrides
                                           # (pass --serial FLEXSIM for the pre-2026-07-31 default)
                             # -> set e.g. 5992 to run on the SAME host as AE without a :4992 clash
HANDLE = 0x1A2B3C4D
HANDLE_HEX = f"{HANDLE:08X}"
PAN_ID = 0x40000000          # FFT stream id   (PCC 0x8003)
WF_ID  = 0x42000000          # waterfall id    (PCC 0x8004)
SERIAL, MODEL, VERSION = "EMULATE01", "FLEX-6600", "3.3.28.0"
MODELS = {                       # model -> caps; slice count drives the multi-slice axis later
    "FLEX-6300": {"slices": 2, "scu": 1}, "FLEX-6400": {"slices": 2, "scu": 1},
    "FLEX-6500": {"slices": 4, "scu": 1}, "FLEX-6600": {"slices": 4, "scu": 2},
    "FLEX-6700": {"slices": 8, "scu": 2},
    # FLEX-8000 series per flexradio.com/comparison: 8400 = 2 slice / 1 SCU, 8600 = 4 slice / 2 SCU
    "FLEX-8400": {"slices": 2, "scu": 1}, "FLEX-8600": {"slices": 4, "scu": 2},
}

CENTER_MHZ, SPAN_MHZ, BINS, FPS = 14.100, 0.250, 1024, 20
# Band -> default panadapter centre (MHz). AE's band buttons send
# "display pan set <pan> band=N"; map the band id to a sensible centre so a pan can be
# retuned by band, and so stacked pans can sit on different bands (20m + 6m) at once.
BAND_CENTERS_MHZ = {
    "160": 1.900, "80": 3.750, "60": 5.357, "40": 7.150, "30": 10.125,
    "20": 14.150, "17": 18.118, "15": 21.225, "12": 24.940, "10": 28.850,
    "6": 50.150, "4": 70.200, "2": 145.000, "wwv": 10.000, "gen": 14.150,
}
PCC_FFT, PCC_WF, PCC_METER = 0x8003, 0x8004, 0x8002
METER_SID = 0x46000000       # meter VITA stream id (AE routes meters by PCC, not by sid)
DAXTX_SID_BASE = 0x84000000  # dax_tx stream id base — matches the id a real 6x00
                             # hands out (0x84000000 observed in the #4510 report log)
DAXTX_AUDIO_PORT = int(os.environ.get("FLEXSIM_DAXTX_PORT", "4991"))
                             # AE hardcodes DAX TX audio to <radio>:4991
                             # (PanadapterStream.cpp: m_radioPort = 4991), so this is only
                             # overridable for TESTS: one process per box can bind it, and a
                             # running sim would otherwise make the txchain suite unrunnable.
_DAXTX_LISTENER_STARTED = False  # one :4991 observer per process (see Radio.__init__)
SLC_LEVEL_ID = 1             # (legacy) single-slice S-meter id; superseded by SLICE_METER_BASE+index
FWDPWR_ID, SWR_ID = 2, 3     # TX meters: forward power (src=TX nam=FWDPWR, dBm->W) + SWR
# Radio-published AMPLIFIER meters (src=AMP). AE creates the Amp applet's
# gauges from THESE, not from the PGXL's own :9008 telemetry -- without them a
# connected PGXL has nowhere to land and the applet stays empty (found on the
# bench 2026-07-30). Routing rule in MeterModel::defineMeter(): src=AMP with
# nam=FWD (unit dBm) / RL / TEMP, and `num` is matched against the TGXL handle
# -- num != tgxlHandle => PGXL (Amp applet); num == tgxlHandle => TGXL (Tuner
# applet). We publish under the AMPLIFIER handle so they route to the Amp side.
# ⚠ nam=RL is RETURN LOSS in dB (AE converts to SWR), NOT an SWR ratio.
AMP_FWD_ID, AMP_RL_ID, AMP_TEMP_ID = 4, 5, 6
SLICE_METER_BASE = 10        # per-slice S-meter: slice k -> meter id SLICE_METER_BASE+k (def num=k)

# pattern timing/shape defaults (seconds / counts)
RAMP_PERIOD = 10.0
SWEEP_PERIOD = 8.0
STEP_DT = 1.0
IMPULSE_PERIOD, IMPULSE_WIDTH = 2.0, 0.15
COMB_TONES = 8
TWO_TONE_SPACING_HZ = 2000.0                       # two-tone ruler: tone separation (700/1900 Hz pair = 2 kHz is the SSB IMD convention's close cousin; whole-span-independent)
NOISE_CAL_RIPPLE_DB = 1.5                           # noise_cal: ± bounded ripple about the exact mean (small enough the level reads true)
TXBLANK_PERIOD, TXBLANK_GAP = 6.0, 2.0
STAIRCASE_STEPS, STAIRCASE_DWELL = 10, 0.6        # 0->max amplitude ladder
SIGNAL_WIDTH_KHZ = 10.0                            # synthesized carrier/signal width (kHz)
CW_TEXT, CW_WPM, CW_TAIL_GAP = "CQ CQ CQ TST CQ CQ CQ TST", 20, 2.0   # CW keyer (Morse) pattern
SSB_LO_HZ, SSB_HI_HZ = 300.0, 2700.0          # SSB-lookalike occupied band, offset from the (suppressed) carrier
SSB_SYLLABLE_HZ = 4.0                          # ~4 syllables/sec — the voice envelope swell/dip rate
SSB_WORD_GAP_S = 0.45                          # brief inter-word pauses (signal drops toward the floor)
CWX_TAIL_S = 0.6            # hold TX this long after the last CWX element so AE's keyer drains first
WF_FLOOR_VAL, WF_PEAK_VAL = 70.0, 235.0   # AE int16/128 intensity for our [min_dbm,max_dbm] range:
                                          # low dBm -> ~black, high dBm -> bright. (Waterfall colour
                                          # still depends on AE's black_level/color_gain; panadapter
                                          # + S-meter read the exact dBm regardless.)
AMP_GAIN_DB = 11.0   # exciter -> amp gain in OPERATE (100 W in => ~1.25 kW out)
S9_DBM = -73.0                            # HF S-meter convention: S9 = -73 dBm, 6 dB per S-unit

# remote_audio_rx VITA-49 constants (sourced from AetherSDR PanadapterStream.cpp)
AUDIO_OUI  = 0x00001C2D   # FlexRadio Systems IEEE OUI
AUDIO_ICC  = 0x534C       # FlexRadio information class code ("SL")
AUDIO_PCC       = 0x03E3  # PCC_IF_NARROW:         float32 stereo @ 24 kHz (full BW)
AUDIO_PCC_MONO  = 0x0123  # PCC_IF_NARROW_REDUCED: int16   mono   @ 24 kHz (reduced BW)
AUDIO_FRAMES = 128        # frames per packet (stereo: 128×2 floats; mono: 128 int16)
AUDIO_RATE   = 24000      # Hz
AUDIO_INTERVAL = AUDIO_FRAMES / AUDIO_RATE   # ~5.333 ms / packet → ~187.5 Hz
AUDIO_SID_BASE = 0x48000010  # base stream id for remote_audio_rx (per-radio: + radio_id)
DAX_SID_BASE   = 0x48000040  # base stream id for dax_rx (per-radio: + radio_id*4 + (ch-1))
                             # AE routes dax_rx packets to its RADE/digital decoder ONLY when the
                             # stream id was registered via a 'stream <id> type=dax_rx ...' status
                             # line (TciServer::statusReceived -> registerDaxStream). The PCC/format
                             # is the SAME as remote_audio_rx (PCC_IF_NARROW float32 stereo), so the
                             # existing audio_loop streams the golden clip unchanged on the DAX id.

AUDIO_SRC_SILENCE = "silence"  # DEFAULT: send silence (no audio) — a spectrum bench shouldn't blast a tone
AUDIO_SRC_TONE  = "tone"    # sine tone at audio_tone_hz (opt-in)
AUDIO_SRC_NOISE = "noise"   # gaussian band noise
AUDIO_SRC_WAV   = "wav"     # WAV file playback (looped)


def log(*a):
    # Encoding-safe: on Windows the console is often cp1252, which raises
    # UnicodeEncodeError on non-ASCII (→, —, …). A crash in log() from inside a
    # worker thread (e.g. audio_loop) silently kills that thread and its stream,
    # so NEVER let a log character take down a thread — fall back to ASCII.
    try:
        print(time.strftime("%H:%M:%S"), *a, flush=True)
    except UnicodeEncodeError:
        safe = [str(x).encode("ascii", "replace").decode("ascii") for x in a]
        print(time.strftime("%H:%M:%S"), *safe, flush=True)


def local_ip():
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        s.connect(("8.8.8.8", 80)); return s.getsockname()[0]
    except OSError:
        return "127.0.0.1"
    finally:
        s.close()


def parse_kvs(text):
    return dict(tok.split("=", 1) for tok in text.split() if "=" in tok)


# ---- VITA-49 builders (PROTOCOL.md §5, big-endian, 28-byte header) ----
def vita_header(stream_id, pcc, seq, payload_len):
    total_words = (28 + payload_len + 3) // 4
    word0 = (0x3 << 28) | (1 << 27) | ((seq & 0xF) << 16) | (total_words & 0xFFFF)
    return struct.pack(">IIIIIII", word0, stream_id, 0x001C2D00, pcc & 0xFFFF, 0, 0, 0)


def fft_packet(stream_id, seq, pixels, frame_index):
    n = len(pixels)
    sub = struct.pack(">HHHHI", 0, n, 2, n, frame_index)
    payload = sub + struct.pack(">%dH" % n, *pixels)
    return vita_header(stream_id, PCC_FFT, seq, len(payload)) + payload


def wf_packet(stream_id, seq, intens, low_hz, binbw_hz, timecode, auto_black=20):
    # auto_black = the tile's AutoBlackLevel field (raw uint, same domain as the
    # intensity samples). A real Flex puts the radio-measured noise-floor level
    # here; AE's #3586 auto-black path uses it as the waterfall black/low point.
    # Pass the frame's measured floor (min raw) for faithful emulation.
    w = len(intens)
    # FrameLowFreq/BinBandwidth are FlexLib "VitaFrequency" = Hz * 2^20. AE >= #4412
    # decodes that format UNCONDITIONALLY (VitaTileFrequency.h) — the old magnitude
    # heuristic that let plain Hz through is gone, so plain Hz now lands ~2^20 low
    # and every row maps off-screen (waterfall black while the pan stays correct).
    low_raw = int(round(low_hz * 1048576))
    binbw_raw = int(round(binbw_hz * 1048576))
    sub = struct.pack(">qqIHHIIHH", low_raw, binbw_raw, 100, w, 1, timecode, auto_black, w, 0)
    payload = sub + struct.pack(">%dh" % w, *intens)       # signed int16, AE reads /128.0
    return vita_header(stream_id, PCC_WF, seq, len(payload)) + payload


def meter_packet(stream_id, seq, meter_id, value, scale=128.0):
    # PCC 0x8002 payload: N x (uint16 meter_id, int16 raw).
    # ⚠ THE SCALE IS PER-UNIT, not universal (MeterModel::convertRaw):
    #     dBm / dB / dBFS / SWR -> raw / 128
    #     degC / degF           -> raw / 64      <-- half!
    #     Volts / Amps          -> raw / 256
    # Encoding a temperature at 128 makes AE read exactly DOUBLE: a 38 C
    # heatsink displayed as 76 C, which is what caught this on the bench
    # (2026-07-30). Pass the matching scale for the meter's declared unit.
    raw = max(-32768, min(32767, int(round(value * scale))))
    payload = struct.pack(">Hh", meter_id, raw)
    return vita_header(stream_id, PCC_METER, seq, len(payload)) + payload


def audio_packet(stream_id, seq, samples, reduced_bw=False):
    """VITA-49 remote_audio_rx packet.
    reduced_bw=False: PCC 0x03E3 — float32 stereo big-endian, 128 frames (1024 B payload).
    reduced_bw=True:  PCC 0x0123 — int16   mono  big-endian, 128 samples (256 B payload).
    AE selects the decoder by PCC; use reduced_bw to match what AE requested via
    'client set send_reduced_bw_dax=1'. OUI/ICC sourced from AetherSDR PanadapterStream.cpp.
    """
    if reduced_bw:
        pcc = AUDIO_PCC_MONO
        raw = [max(-32767, min(32767, int(v * 32767))) for v in samples[:AUDIO_FRAMES]]
        payload = struct.pack(f">{AUDIO_FRAMES}h", *raw)   # 256 bytes = 64 words
    else:
        pcc = AUDIO_PCC
        payload = struct.pack(f">{AUDIO_FRAMES * 2}f", *samples)  # 1024 bytes = 256 words
    total_words = 7 + len(payload) // 4
    word0 = (0x3 << 28) | (1 << 27) | ((seq & 0xF) << 16) | total_words
    header = struct.pack(">IIIIIII",
                         word0, stream_id,
                         AUDIO_OUI, (AUDIO_ICC << 16) | pcc,
                         0, 0, 0)
    return header + payload


class _FloatWavReader:
    """Minimal WAV reader for IEEE-float (fmt tag 3) files, exposing the slice of
    the stdlib `wave.Wave_read` interface WavPlayer needs. Used as a fallback when
    `wave.open` rejects a float WAV (e.g. RADE Tap-E clips: 24k stereo float32)."""

    def __init__(self, path):
        with open(path, 'rb') as f:
            data = f.read()
        if data[:4] != b'RIFF' or data[8:12] != b'WAVE':
            raise wave.Error("not a RIFF/WAVE file")
        # walk chunks to find 'fmt ' and 'data'
        pos, fmt, dpos, dlen = 12, None, None, 0
        while pos + 8 <= len(data):
            cid = data[pos:pos+4]
            clen = struct.unpack('<I', data[pos+4:pos+8])[0]
            body = pos + 8
            if cid == b'fmt ':
                fmt = struct.unpack('<HHIIHH', data[body:body+16])
            elif cid == b'data':
                dpos, dlen = body, clen
            pos = body + clen + (clen & 1)         # chunks are word-aligned
        if fmt is None or dpos is None:
            raise wave.Error("missing fmt/data chunk")
        tag, ch, sr, _byterate, _ba, bits = fmt
        self.is_float = (tag == 3)
        if not self.is_float:
            raise wave.Error(f"_FloatWavReader only handles float (tag 3), got tag {tag}")
        self._ch, self._sr, self._sw = ch, sr, bits // 8
        self._frame_bytes = self._ch * self._sw
        self._data = data[dpos:dpos+dlen]
        self._cur = 0                              # byte cursor into _data

    def getnchannels(self): return self._ch
    def getsampwidth(self):  return self._sw
    def getframerate(self):  return self._sr

    def readframes(self, n):
        nb = n * self._frame_bytes
        chunk = self._data[self._cur:self._cur + nb]
        self._cur += len(chunk)
        return chunk

    def rewind(self): self._cur = 0
    def close(self):  pass


class WavPlayer:
    """Reads a WAV file and returns resampled mono float samples at AUDIO_RATE.

    Supports 8/16/32-bit PCM **and 32-bit IEEE float** (e.g. RADE Tap-E golden
    clips), any channel count (mixed to mono), any sample rate (linear-interp
    resample to 24 kHz).  Loops at EOF.
    """

    def __init__(self, path):
        # Python's wave module handles PCM but raises on IEEE-float (fmt tag 3) —
        # which is exactly what RADE Tap E emits. Try wave first (keeps PCM
        # behaviour identical); on failure, fall back to a manual float-aware reader.
        self._is_float = False
        try:
            self._wf = wave.open(path, 'rb')
            self.n_ch = self._wf.getnchannels()
            self.sw   = self._wf.getsampwidth()
            self.rate = self._wf.getframerate()
        except (wave.Error, EOFError):
            self._wf = _FloatWavReader(path)
            self.n_ch = self._wf.getnchannels()
            self.sw   = self._wf.getsampwidth()
            self.rate = self._wf.getframerate()
            self._is_float = self._wf.is_float
        self._ratio = self.rate / AUDIO_RATE   # src samples per output sample
        self._buf   = []     # decoded mono floats not yet consumed
        self._phase = 0.0    # fractional carry-over into next call
        log(f"[wav] opened: {self.rate}Hz {self.n_ch}ch {self.sw*8}bit"
            f"{' float' if self._is_float else ''} -> {AUDIO_RATE}Hz mono "
            f"(ratio {self._ratio:.4f})")

    # ------------------------------------------------------------------
    def _sample_to_float(self, raw):
        if self._is_float:                                     # 32-bit IEEE float
            return struct.unpack('<f', raw)[0]
        if self.sw == 2:
            return int.from_bytes(raw, 'little', signed=True) / 32768.0
        if self.sw == 4:
            return int.from_bytes(raw, 'little', signed=True) / 2147483648.0
        return (int.from_bytes(raw, 'little') - 128) / 128.0   # uint8

    def _fill(self, n_src):
        """Append at least n_src decoded mono floats to self._buf (looping WAV)."""
        frame_bytes = self.n_ch * self.sw
        needed = n_src
        while needed > 0:
            raw = self._wf.readframes(needed)
            if not raw:
                self._wf.rewind(); continue
            got = len(raw) // frame_bytes
            for i in range(got):
                frame = raw[i * frame_bytes:(i + 1) * frame_bytes]
                s = sum(self._sample_to_float(frame[c * self.sw:(c + 1) * self.sw])
                        for c in range(self.n_ch)) / self.n_ch
                self._buf.append(s)
            needed -= got

    def read(self, n_out):
        """Return n_out resampled mono float samples, continuing phase from last call."""
        n_src_needed = int(math.ceil(self._phase + n_out * self._ratio)) + 1
        self._fill(max(0, n_src_needed - len(self._buf)))
        out = []
        p = self._phase
        for _ in range(n_out):
            lo = int(p)
            frac = p - lo
            hi = min(lo + 1, len(self._buf) - 1)
            out.append(self._buf[lo] * (1 - frac) + self._buf[hi] * frac)
            p += self._ratio
        consumed = int(p)
        self._buf = self._buf[consumed:]
        self._phase = p - consumed
        return out

    def close(self):
        try: self._wf.close()
        except Exception: pass


# ---- pattern engine: each pattern returns a per-bin dBm list (or None = TX gap) ----
class PatternCtx:
    def __init__(self, n, center_bin, min_dbm, max_dbm):
        self.n, self.center = n, center_bin
        self.min_dbm, self.max_dbm = min_dbm, max_dbm
        self.floor = min_dbm + 10.0           # visible noise floor (live: noise-floor slider)
        self.sig_level = max_dbm - 5.0        # signal peak level (live: signal-level slider)
        self.sig_half = 2                     # half-width of a synthesized signal, in bins
        self.two_tone_half_bins = 1           # two-tone ruler: half the tone spacing, in bins
        self.noise_cal_full = True            # noise_cal: full-span bed (the standard); False = band-limit by width
        self.noise_color = 0.0                # noise tilt dB across band (0=white, >0=pink)
        self.cw_keydown = False               # CW keyer: element key-down this frame
        self.cw_in_message = False            # CW keyer: inside the message (TX over), not the tail gap
        self.cw_qsk = False                   # CW/CWX: full break-in (RX peeks between elements)


def _flat(ctx, dbm):
    return [dbm] * ctx.n


def _hash01(n):
    # Deterministic [0,1) from an int (cheap integer hash). Lets noise_cal be
    # reproducible per (frame, bin) so a golden capture repeats, while looking random.
    n &= 0xFFFFFFFF
    n = (n ^ 61) ^ (n >> 16)
    n = (n + (n << 3)) & 0xFFFFFFFF
    n ^= (n >> 4)
    n = (n * 0x27d4eb2d) & 0xFFFFFFFF
    n ^= (n >> 15)
    return (n & 0xFFFFFFFF) / 4294967296.0


def pat_noise_floor(ctx, t):
    return _flat(ctx, ctx.floor)


def pat_ramp(ctx, t):
    frac = (t % RAMP_PERIOD) / RAMP_PERIOD
    return _flat(ctx, ctx.min_dbm + frac * (ctx.max_dbm - ctx.min_dbm))


def pat_cal_tones(ctx, t):
    out = _flat(ctx, ctx.floor)
    for frac, dbm in [(0.2, -100), (0.4, -80), (0.6, -60), (0.8, -40)]:
        b = int(frac * ctx.n)
        for d in (-1, 0, 1):
            if 0 <= b + d < ctx.n:
                out[b + d] = max(out[b + d], dbm)
    return out


def pat_swept_carrier(ctx, t):
    out = _flat(ctx, ctx.floor)
    c = int(((t % SWEEP_PERIOD) / SWEEP_PERIOD) * (ctx.n - 1))
    for d in range(-ctx.sig_half, ctx.sig_half + 1):
        b = c + d
        if 0 <= b < ctx.n:
            out[b] = ctx.sig_level
    return out


def pat_comb(ctx, t):
    out = _flat(ctx, ctx.floor)
    step = max(1, ctx.n // COMB_TONES)
    for b in range(step // 2, ctx.n, step):
        out[b] = ctx.sig_level - 10
    return out


def pat_two_tone(ctx, t):
    # Calibrated two-tone "golden ruler": two EQUAL tones symmetric about the VFO,
    # spaced TWO_TONE_SPACING_HZ apart, each at exactly ctx.sig_level dBm. Single-bin
    # placement (no level-corrupting smear) so the level you measure IS the level set.
    # Classic linearity/IMD card: in a pure sim path the band between/around the tones
    # stays at the noise floor (no intermod products) — so it also proves AE isn't
    # inventing spurs. Spacing is rounded to whole bins (ctx.two_tone_half_bins, set
    # per-frame from the live span); the true delivered spacing + the expected per-tone
    # dBm/S-unit are reported in the control-panel hint so the ruler reads exactly.
    out = _flat(ctx, ctx.floor)
    for sign in (-1, +1):
        b = ctx.center + sign * ctx.two_tone_half_bins
        if 0 <= b < ctx.n:
            out[b] = ctx.sig_level
    return out


def pat_step(ctx, t):
    hi = int(t / STEP_DT) % 2 == 0
    return _flat(ctx, ctx.sig_level if hi else ctx.floor)


def pat_impulse(ctx, t):
    bright = (t % IMPULSE_PERIOD) < IMPULSE_WIDTH
    return _flat(ctx, ctx.sig_level if bright else ctx.floor)


def pat_noise(ctx, t):
    # A band of random noise, `sig_width_khz` wide, centered at the VFO. noise_color
    # tilts the band: 0 = white (flat), higher = pink/red (more power on the low side).
    out = [ctx.floor] * ctx.n
    lo = max(0, ctx.center - ctx.sig_half)
    hi = min(ctx.n - 1, ctx.center + ctx.sig_half)
    span = max(1, hi - lo)
    for b in range(lo, hi + 1):
        frac = (b - lo) / span                            # 0 low edge -> 1 high edge
        env = ctx.sig_level - ctx.noise_color * frac      # spectral tilt
        out[b] = env - random.uniform(0.0, 12.0)          # per-bin random texture
    return out


def pat_ssb(ctx, t):
    # An SSB-VOICE LOOKALIKE — not a calibration ruler, just something that *reads*
    # like real on-air USB on the waterfall: a band offset above the (suppressed)
    # carrier, ~300-2700 Hz wide, with a syllabic envelope (swells + word gaps) and
    # formant-like energy that drifts within the band. Deterministic-ish via t so it
    # animates smoothly frame to frame; uses random for live texture (not for golden
    # captures — this pattern is for looks, not measurement).
    out = [ctx.floor] * ctx.n
    # Map the SSB band (Hz offset) to bins. The stream loop sets ctx.sig_half from the
    # width slider, but SSB has its OWN fixed ~2.4 kHz band, so derive bins from the
    # span the caller encoded into sig_half's reference. We approximate using the same
    # bins-per-Hz the caller used for two_tone (ctx carries two_tone_half_bins for 2 kHz).
    hz_per_bin = (TWO_TONE_SPACING_HZ / 2.0) / max(1, ctx.two_tone_half_bins)
    lo_bin = ctx.center + int(SSB_LO_HZ / hz_per_bin)        # USB: band sits ABOVE the carrier
    hi_bin = ctx.center + int(SSB_HI_HZ / hz_per_bin)
    lo = max(0, min(ctx.n - 1, lo_bin))
    hi = max(0, min(ctx.n - 1, hi_bin))
    if hi <= lo:
        return out
    span = hi - lo

    # --- voice envelope: syllabic swell, occasional word gaps (drops toward floor) ---
    syl = 0.5 + 0.5 * math.sin(2 * math.pi * SSB_SYLLABLE_HZ * t)      # 0..1 syllable swell
    syl *= 0.6 + 0.4 * math.sin(2 * math.pi * 1.3 * t + 1.1)           # slower amplitude drift
    word_phase = (t % (1.0 / 0.7)) * 0.7                               # ~0.7 "words"/s cycle
    in_gap = (word_phase % 1.0) < (SSB_WORD_GAP_S * 0.7)               # brief inter-word silence
    env = 0.12 if in_gap else (0.55 + 0.45 * max(0.0, syl))           # overall loudness 0..1

    # `peak_dbm` is where a loud syllable on a formant should sit (the signal-level slider).
    peak_dbm = ctx.sig_level
    # How far DOWN from the peak the quiet parts of the band sag (dB of speech dynamic range).
    SSB_DR = 24.0

    # --- two drifting "formant" centres so energy clumps + moves like speech ---
    f1 = 0.30 + 0.18 * math.sin(2 * math.pi * 0.9 * t)                 # 0..1 within band
    f2 = 0.62 + 0.20 * math.sin(2 * math.pi * 0.6 * t + 2.0)
    for b in range(lo, hi + 1):
        x = (b - lo) / span                                           # 0..1 across the band
        edge = max(0.0, min(x / 0.12, (1 - x) / 0.12, 1.0))          # taper first/last ~12% (filter skirts)
        g1 = math.exp(-((x - f1) ** 2) / (2 * 0.06 ** 2))            # formant clumps
        g2 = 0.8 * math.exp(-((x - f2) ** 2) / (2 * 0.08 ** 2))
        shape = max(0.0, min(1.0, (0.35 + 0.85 * (g1 + g2)))) * edge  # 0..1 spectral shape
        # Level in dB BELOW the peak: loud-syllable * on-a-formant = at the peak; else sags.
        below = SSB_DR * (1.0 - env * shape)
        level = peak_dbm - below - random.uniform(0.0, 4.0)          # live per-bin texture
        out[b] = max(ctx.floor, level)
    return out


def pat_noise_cal(ctx, t):
    # Calibrated noise bed for FILTER testing. Unlike `noise` (textured ±12 dB random,
    # for looks), this holds an EXACT mean of ctx.sig_level dBm across the band with only
    # a small bounded ripple (±NOISE_CAL_RIPPLE_DB) so the level reads true. Feed it
    # through AE's filter and the output IS the filter's frequency response: bins inside
    # the passband sit at sig_level, bins outside drop to the floor, the edges trace the
    # roll-off — the whole curve in one frame (a tone shows one point; flat noise the lot).
    #
    # Full-span by default (noise_cal_full=True) — the STANDARD bed: noise everywhere,
    # so AE's filter carves the visible shape. Clear the flag (panel slider) to BAND-LIMIT
    # to sig_width_khz for ad-hoc/unknown tests — itself a filter test signal (watch AE's
    # filter chew the edges as you cross its width). noise_color tilts white->pink.
    # Deterministic per (frame, bin) so a golden capture is reproducible, yet looks noisy.
    if ctx.noise_cal_full:                                 # standard: fill the whole span
        lo, hi = 0, ctx.n - 1
    else:                                                  # ad-hoc: band-limit to the width
        lo = max(0, ctx.center - ctx.sig_half)
        hi = min(ctx.n - 1, ctx.center + ctx.sig_half)
    out = [ctx.floor] * ctx.n
    span = max(1, hi - lo)
    frame = int(t * 1000)                                  # stable within a frame, varies across
    for b in range(lo, hi + 1):
        frac = (b - lo) / span                             # 0 low edge -> 1 high edge
        mean = ctx.sig_level - ctx.noise_color * frac      # exact mean (tilt if pink)
        # bounded zero-mean ripple, seeded by (frame,bin) -> reproducible, noise-like
        ripple = (_hash01(frame * 131071 + b) - 0.5) * 2.0 * NOISE_CAL_RIPPLE_DB
        out[b] = mean + ripple
    return out


def pat_tx_blank(ctx, t):
    # Periodic TX gap. During the gap return None (no pan/wf data); stream_loop
    # ALSO asserts real TX state (interlock TRANSMITTING + forward power) so AE
    # runs its waterfall-during-TX path -> repro #2126 (blank/flicker during TX)
    # and #1916 (waterfall gone after TX).
    if (t % TXBLANK_PERIOD) < TXBLANK_GAP:
        return None                                    # TX gap: send nothing
    out = _flat(ctx, ctx.floor)
    for d in range(-ctx.sig_half, ctx.sig_half + 1):   # honour the width slider, like carrier/staircase
        b = ctx.center + d
        if 0 <= b < ctx.n:
            out[b] = ctx.sig_level
    return out


def pat_staircase(ctx, t):
    # A center carrier whose amplitude steps floor->max in STAIRCASE_STEPS discrete
    # levels (each held STAIRCASE_DWELL s), then repeats. Watch the waterfall carrier
    # climb out of black in even brightness steps -> shows the colormap/black threshold.
    cycle = STAIRCASE_STEPS * STAIRCASE_DWELL
    step = int((t % cycle) / STAIRCASE_DWELL)              # 0 .. STEPS-1
    frac = step / max(1, STAIRCASE_STEPS - 1)
    level = ctx.floor + frac * (ctx.sig_level - ctx.floor)   # floor (0) -> signal level
    out = [ctx.floor] * ctx.n
    for d in range(-ctx.sig_half, ctx.sig_half + 1):
        b = ctx.center + d
        if 0 <= b < ctx.n:
            out[b] = level
    return out


def pat_carrier(ctx, t):
    # Steady carrier at the VFO at sig_level — for S-meter / dBm calibration.
    out = [ctx.floor] * ctx.n
    for d in range(-ctx.sig_half, ctx.sig_half + 1):
        b = ctx.center + d
        if 0 <= b < ctx.n:
            out[b] = ctx.sig_level
    return out


def pat_test_card(ctx, t):
    # Auto-black test card: 17 carriers evenly spaced across the band at FIXED
    # levels S1..S9 in 1/2-S (3 dB) steps (S9 = -73 dBm, 6 dB per S-unit), on a
    # flat noise floor at ctx.floor. The floor is the swept variable (noise-floor
    # slider / /set?floor=): as it rises, carriers below it submerge, and because
    # flex-sim advertises auto_black = ctx.floor the receiver's black point tracks
    # the floor up with it. One pattern that shows the level->colour gradient,
    # auto-black tracking, and signal/noise contrast (PR #3586 / sblanchard).
    out = [ctx.floor] * ctx.n
    s9 = -73.0
    levels = [s9 - k * 3.0 for k in range(16, -1, -1)]   # -121 (S1) .. -73 (S9)
    m = len(levels)
    for i, dbm in enumerate(levels):
        b = int((i + 0.5) / m * ctx.n)
        for d in range(-ctx.sig_half, ctx.sig_half + 1):
            bb = b + d
            if 0 <= bb < ctx.n:
                out[bb] = max(out[bb], dbm)
    return out


# ---- CW keyer (Morse) ----
_MORSE = {
    "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": "----.", "/": "-..-.", "?": "..--..", ".": ".-.-.-",
    ",": "--..--", "=": "-...-",
}


def _schedule_keydown(schedule, tt):
    acc = 0.0
    for on, dur in schedule:
        if tt < acc + dur:
            return on
        acc += dur
    return False


def _morse_schedule(text, wpm):
    # [(key_down, seconds), ...] for `text` at `wpm` (PARIS timing: dit = 1.2/wpm s).
    dit = 1.2 / wpm
    seg = []
    for wi, word in enumerate(text.split(" ")):
        if wi:
            seg.append((False, 7 * dit))                # inter-word gap
        for ci, ch in enumerate(word):
            code = _MORSE.get(ch.upper())
            if not code:
                continue
            if ci:
                seg.append((False, 3 * dit))            # inter-character gap
            for ei, el in enumerate(code):
                if ei:
                    seg.append((False, dit))            # intra-character gap
                seg.append((True, (3 if el == "-" else 1) * dit))
    return seg


def _morse_char_ends(text, wpm):
    # Cumulative time (s) at which each keyed character finishes — for 'cwx sent=N'.
    dit = 1.2 / wpm
    t, ends = 0.0, []
    for wi, word in enumerate(text.split(" ")):
        if wi:
            t += 7 * dit
        for ci, ch in enumerate(word):
            code = _MORSE.get(ch.upper())
            if not code:
                continue
            if ci:
                t += 3 * dit
            for ei, el in enumerate(code):
                if ei:
                    t += dit
                t += (3 if el == "-" else 1) * dit
            ends.append(t)
    return ends


CW_SCHEDULE = _morse_schedule(CW_TEXT, CW_WPM) + [(False, CW_TAIL_GAP)]
CW_CYCLE = sum(d for _, d in CW_SCHEDULE)


def pat_cw(ctx, t):
    # Sim-driven Morse keyer at CW_WPM (the panel "Send CW" button — the sim fakes
    # the keying). key-down = TX burst (RX muted -> None); key-up = brief RX (carrier);
    # tail gap = RX; repeat. (For an AE-initiated over, use CWX — see handle_cwx.)
    tt = t % CW_CYCLE
    keydown = _schedule_keydown(CW_SCHEDULE, tt)
    in_msg = tt < (CW_CYCLE - CW_TAIL_GAP)
    ctx.cw_keydown = keydown
    ctx.cw_in_message = in_msg
    if keydown or (in_msg and not ctx.cw_qsk):          # TX burst, or non-QSK over: RX muted
        return None
    out = [ctx.floor] * ctx.n                            # QSK key-up peek, or the tail RX gap
    for d in range(-ctx.sig_half, ctx.sig_half + 1):
        b = ctx.center + d
        if 0 <= b < ctx.n:
            out[b] = ctx.sig_level
    return out


PATTERNS = {
    "carrier": pat_carrier, "cw": pat_cw,
    "noise_floor": pat_noise_floor, "ramp": pat_ramp, "cal_tones": pat_cal_tones,
    "swept_carrier": pat_swept_carrier, "comb": pat_comb, "two_tone": pat_two_tone,
    "step": pat_step, "impulse": pat_impulse, "tx_blank": pat_tx_blank,
    "staircase": pat_staircase, "noise": pat_noise, "noise_cal": pat_noise_cal,
    "test_card": pat_test_card, "ssb": pat_ssb,
}
AUTO_TX_PATTERNS = {"tx_blank", "cw"}     # patterns that drive TX state themselves


# ===========================================================================
# HF NOISE BENCH — additive audio mixer for testing AE's noise reduction.
#
# AE's NR (RN2/NR2/NR4/DFNR/BNR/MacNR) runs on the remote_audio_rx audio path
# (AudioEngine.cpp processMixedRxAudioData -> NR block). This bench builds the
# audio it hears as a LINEAR SUM of independently-toggleable channels:
#
#   frame = Σ enabled signal channels  +  Σ enabled noise channels   (then clip)
#
# Everything is GENERATED ON THE SPOT — no recordings for the noise (its whole
# point is non-repeating randomness). Only the voice channel plays a WAV.
#
# Each generator has signature gen(st, ch, n) -> list[float] of length n:
#   st  = the Radio's NoiseMixer state (rate, phase counters, RNG helpers)
#   ch  = this channel's dict (level_db + any per-channel knobs)
#   n   = frames to produce (== AUDIO_FRAMES)
# It returns samples at UNITY reference; the mixer scales by 10**(level_db/20).
# level_db is dBFS-ish: 0 = full reference amplitude, -20 = /10, etc.
# ===========================================================================

NOISE_REF_AMP = 0.18          # unity-reference RMS-ish amplitude (headroom for summing)
_PINK_ROWS    = 16            # Voss-McCartney octave rows


def _db2lin(db):
    return 10.0 ** (db / 20.0)


def _tnf_width_hz(raw):
    """AE sends TNF width in MHz (TnfModel.cpp:86, width/1e6). Convert to Hz,
    tolerating an already-Hz value defensively (matches AE's own parser)."""
    v = float(raw)
    return int(round(v * 1.0e6)) if 0.0 < v < 1.0 else max(10, int(round(v)))


# ---- signal channels (the WANTED signal NR must PRESERVE) -----------------

def sig_voice(st, ch, n):
    """Voice from a looped WAV (the only recorded source). Falls silent if no
    WAV is loaded — the mixer just skips it then."""
    wav = st.voice_wav
    if wav is None:
        return [0.0] * n
    return wav.read(n)          # WavPlayer already returns mono float at AUDIO_RATE


def sig_cw(st, ch, n):
    """A separate keyed CW tone at cw_hz, sending a repeating id in REAL Morse
    so NR has a tonal signal to preserve and a decoder reads actual content.
    Envelope-shaped keying (no clicks).

    Was an element table [1,3,1,1,3,0,0] that conflated duration with key
    state — every non-zero entry keyed, so the id rendered as one fused
    ~600 ms tone ('L' feel, not L). Found via the C++ port of this table in
    AetherSDR demo mode (aethersdr#4593/#4594); now built from the same
    _morse_schedule the panel keyer and CWX already use."""
    hz   = ch.get("hz", 700.0)
    wpm  = ch.get("wpm", 18.0)
    text = ch.get("id", "L")
    cache = ch.get("_sched")
    if not cache or cache[0] != (text, wpm):
        # schedule + 7-dit word gap before the loop restarts
        sched = _morse_schedule(text, wpm) + [(False, 7 * 1.2 / wpm)]
        cache = ((text, wpm), sched, sum(d for _, d in sched))
        ch["_sched"] = cache
    _, sched, total = cache
    edge = 0.005                                       # raised-cosine, 5 ms
    out = []
    for i in range(n):
        tt = (st.cw_phase + i) / st.rate
        pos = tt % total
        acc = 0.0; key = 0.0
        for on, dur in sched:
            if pos < acc + dur:
                if on:
                    key = 1.0
                    into = pos - acc; left = acc + dur - pos
                    if into < edge: key = 0.5 - 0.5 * math.cos(math.pi * into / edge)
                    elif left < edge: key = 0.5 - 0.5 * math.cos(math.pi * left / edge)
                break
            acc += dur
        out.append(key * math.sin(2 * math.pi * hz * tt))
    st.cw_phase += n
    return out


# ---- noise channels (all live DSP, non-repeating) -------------------------

def noi_white(st, ch, n):
    """Flat AWGN — the thermal baseline."""
    g = st.rng.gauss
    return [g(0.0, NOISE_REF_AMP) for _ in range(n)]


def noi_pink(st, ch, n):
    """1/f pink noise (Voss-McCartney) — the atmospheric band-hiss floor."""
    rows = st._pink_rows
    key  = st._pink_key
    out  = []
    for _ in range(n):
        st._pink_ctr = (st._pink_ctr + 1) & ((1 << _PINK_ROWS) - 1)
        diff = key ^ st._pink_ctr
        for r in range(_PINK_ROWS):
            if diff & (1 << r):
                rows[r] = st.rng.uniform(-1.0, 1.0)
        key = st._pink_ctr
        out.append(sum(rows) / _PINK_ROWS * NOISE_REF_AMP * 3.0)
    st._pink_key = key
    return out


def noi_qrn(st, ch, n):
    """Atmospheric impulse crackle (lightning static) — Poisson-timed sharp
    impulses with fast exponential decay. rate = impulses/sec."""
    rate = ch.get("rate", 12.0)
    p    = rate / st.rate                              # per-sample trigger prob
    out  = []
    for _ in range(n):
        if st.rng.random() < p:
            st._qrn_env = 1.0
            st._qrn_sign = 1.0 if st.rng.random() < 0.5 else -1.0
        v = st._qrn_sign * st._qrn_env * (0.6 + 0.4 * st.rng.random())
        st._qrn_env *= 0.86                            # ~ a few ms tail at 24 k
        out.append(v * NOISE_REF_AMP * 5.0)
    return out


def noi_powerline(st, ch, n):
    """Mains buzz — fundamental + odd harmonics (the classic power-line raster).
    freq = 50 or 60 Hz."""
    f0 = ch.get("freq", 60.0)
    out = []
    for i in range(n):
        tt = (st.pl_phase + i) / st.rate
        v = 0.0
        for h in (1, 3, 5, 7, 9, 11):
            v += (1.0 / h) * math.sin(2 * math.pi * f0 * h * tt)
        out.append(v * NOISE_REF_AMP * 0.5)
    st.pl_phase += n
    return out


def noi_crashes(st, ch, n):
    """Static crashes — big correlated bursts of band-limited noise (storm
    front). Rare, long-tailed. rate = crashes/sec."""
    rate = ch.get("rate", 0.4)
    p    = rate / st.rate
    out  = []
    for _ in range(n):
        if st.rng.random() < p:
            st._crash_env = 1.0
        # low-passed noise (one-pole) for a rushing rather than hissing texture
        st._crash_lp = 0.6 * st._crash_lp + 0.4 * st.rng.gauss(0.0, 1.0)
        out.append(st._crash_lp * st._crash_env * NOISE_REF_AMP * 4.0)
        st._crash_env *= 0.9995                        # long decay (~150 ms)
    return out


def noi_birdie(st, ch, n):
    """A steady carrier heterodyne — a pure tone birdie for notch/auto-notch
    testing. hz = audio pitch of the het."""
    hz = ch.get("hz", 1000.0)
    out = []
    for i in range(n):
        tt = (st.birdie_phase + i) / st.rate
        out.append(math.sin(2 * math.pi * hz * tt) * NOISE_REF_AMP * 1.2)
    st.birdie_phase += n
    return out


def noi_hash(st, ch, n):
    """Switching-supply / digital hash — broadband noise gated into clumps with
    a fast repetition rate (SMPS whine + broadband splatter)."""
    prf = ch.get("prf", 120.0)                         # clump repetition (Hz)
    out = []
    for i in range(n):
        tt = (st.hash_phase + i) / st.rate
        gate = 1.0 if (tt * prf) % 1.0 < 0.4 else 0.15
        out.append(st.rng.gauss(0.0, 1.0) * gate * NOISE_REF_AMP * 2.0)
    st.hash_phase += n
    return out


def noi_woodpecker(st, ch, n):
    """Pulsed wideband rasp (over-the-horizon-radar flavour) — noise bursts at a
    low PRF. prf = pulses/sec."""
    prf = ch.get("prf", 10.0)
    out = []
    for i in range(n):
        tt = (st.wood_phase + i) / st.rate
        on = (tt * prf) % 1.0 < 0.5                     # ~50% duty rasp
        out.append((st.rng.gauss(0.0, 1.0) if on else 0.0) * NOISE_REF_AMP * 3.0)
    st.wood_phase += n
    return out


# Registry: name -> (generator, is_signal, default channel dict).
# default dict carries level_db (starts DISABLED via enabled=False) + knobs.
NOISE_CHANNELS = {
    # signal channels (wanted)
    "voice":      (sig_voice,      True,  {"enabled": False, "level_db": -14.0}),
    "cw":         (sig_cw,         True,  {"enabled": False, "level_db": -16.0, "hz": 700.0, "wpm": 18.0}),
    # noise channels
    "white":      (noi_white,      False, {"enabled": False, "level_db": -26.0}),
    "pink":       (noi_pink,       False, {"enabled": False, "level_db": -24.0}),
    "qrn":        (noi_qrn,        False, {"enabled": False, "level_db": -18.0, "rate": 12.0}),
    "powerline":  (noi_powerline,  False, {"enabled": False, "level_db": -22.0, "freq": 60.0}),
    "crashes":    (noi_crashes,    False, {"enabled": False, "level_db": -16.0, "rate": 0.4}),
    "birdie":     (noi_birdie,     False, {"enabled": False, "level_db": -28.0, "hz": 1000.0}),
    "hash":       (noi_hash,       False, {"enabled": False, "level_db": -24.0, "prf": 120.0}),
    "woodpecker": (noi_woodpecker, False, {"enabled": False, "level_db": -22.0, "prf": 10.0}),
}


class NoiseMixer:
    """Per-radio additive mixer state. Holds each channel's live dict, phase
    counters for the deterministic tone/buzz generators, and RNG scratch for the
    stochastic ones. mix(n) returns a summed, soft-clipped mono float block."""

    def __init__(self, voice_wav=None):
        import copy
        self.rate = AUDIO_RATE
        self.channels = {name: copy.deepcopy(defs)
                         for name, (_g, _s, defs) in NOISE_CHANNELS.items()}
        self.voice_wav = voice_wav
        self.rng = random.Random(0x5EED)   # own stream; reproducible per launch
        # phase counters
        self.cw_phase = self.pl_phase = self.birdie_phase = 0
        self.hash_phase = self.wood_phase = 0
        # generator scratch
        self._pink_rows = [0.0] * _PINK_ROWS
        self._pink_ctr = 0; self._pink_key = 0
        self._qrn_env = 0.0; self._qrn_sign = 1.0
        self._crash_env = 0.0; self._crash_lp = 0.0
        # TNF (tracking notch) — audio Hz offsets to remove + per-notch biquad state
        self._notch_hz = []
        self._notch_width_hz = {}   # round(f_hz) -> width Hz (for the display dip)
        self._notch_state = {}

    def any_enabled(self):
        return any(c["enabled"] for c in self.channels.values())

    def mix(self, n):
        acc = [0.0] * n
        for name, (gen, _is_sig, _d) in NOISE_CHANNELS.items():
            ch = self.channels[name]
            if not ch["enabled"]:
                continue
            g = _db2lin(ch["level_db"])
            block = gen(self, ch, n)
            for i in range(n):
                acc[i] += block[i] * g
        # Tracking Notch Filters: remove each notched AUDIO frequency in the time
        # domain (biquad notch), so a TNF placed on the birdie kills the tone you
        # hear -- matching the display notch. self.notch_hz is a list of audio Hz.
        for f_hz in self._notch_hz:
            acc = self._apply_notch(acc, f_hz)
        # soft clip (tanh) to keep within float range without hard edges
        for i in range(n):
            acc[i] = math.tanh(acc[i])
        return acc

    def set_notches(self, notches):
        """Set the notch list (TNF + ANF). `notches` = list of (audio_hz, width_hz)
        offsets from the VFO; only those inside the audio band are kept."""
        self._notch_hz = []
        self._notch_width_hz = {}
        for f, w in notches:
            if abs(f) < self.rate / 2:
                self._notch_hz.append(f)
                self._notch_width_hz[round(f)] = w

    def auto_notch_tones(self):
        """ANF detection: the audio-Hz offsets of steady TONAL channels currently
        active (birdie carrier, CW tone). A real Auto-Notch Filter finds these by
        their narrow spectral signature; the sim knows them directly. Broadband /
        impulse noise is NOT returned -- ANF can't catch those, same as hardware.
        Returned as (hz, width_hz) so they slot straight into set_notches()."""
        tones = []
        for name in ("birdie", "cw"):
            ch = self.channels[name]
            if ch["enabled"]:
                tones.append((ch.get("hz", 1000.0), 120.0))   # ~120 Hz ANF notch
        return tones

    def _apply_notch(self, buf, f_hz, q=8.0):
        """One-pole-pair biquad notch at f_hz, with per-notch state carried across
        blocks (keyed by rounded freq) so there are no block-edge clicks."""
        w0 = 2.0 * math.pi * f_hz / self.rate
        alpha = math.sin(w0) / (2.0 * q)
        cw = math.cos(w0)
        b0, b1, b2 = 1.0, -2.0 * cw, 1.0
        a0, a1, a2 = 1.0 + alpha, -2.0 * cw, 1.0 - alpha
        b0, b1, b2 = b0 / a0, b1 / a0, b2 / a0
        a1, a2 = a1 / a0, a2 / a0
        st = self._notch_state.setdefault(round(f_hz), [0.0, 0.0, 0.0, 0.0])
        x1, x2, y1, y2 = st
        out = [0.0] * len(buf)
        for i, x0 in enumerate(buf):
            y0 = b0 * x0 + b1 * x1 + b2 * x2 - a1 * y1 - a2 * y2
            out[i] = y0
            x2, x1 = x1, x0
            y2, y1 = y1, y0
        st[0], st[1], st[2], st[3] = x1, x2, y1, y2
        return out

    def snapshot(self):
        """JSON-safe view for /state readback + Claude/QA hooks."""
        return {name: dict(ch) for name, ch in self.channels.items()}

    def spectrum(self, n, floor_dbm, span_hz, center_bin):
        """Render the ACTIVE noise scene as a dBm-per-bin display spectrum, so
        the waterfall SHOWS what the audio mixer is producing.

        This is an analytic render of each channel's known spectral signature
        (not a live FFT of the audio — that would be pure-Python-slow and only
        span the audio bandwidth). The audio path and this display path both
        read the SAME channel levels, so a change in either is reflected in both.

        n           bins in the row
        floor_dbm   the configured display floor (quiet baseline)
        span_hz     total display span in Hz (maps audio Hz -> bins)
        center_bin  the VFO bin (audio 0 Hz sits here; +Hz to the right)
        """
        out = [floor_dbm] * n
        hz_per_bin = span_hz / n if n else 1.0

        def add_db(b, dbm):                       # power-sum a contribution into a bin
            if 0 <= b < n:
                a, x = out[b], dbm
                hi, lo = (a, x) if a >= x else (x, a)
                out[b] = hi + 10.0 * math.log10(1.0 + 10.0 ** ((lo - hi) / 10.0))

        def bin_of(hz):                           # audio Hz -> display bin (USB: +Hz above VFO)
            return int(center_bin + hz / hz_per_bin)

        # Map a channel's level_db (dBFS-ish) to an on-screen dBm. Reference the
        # display so a -20 dBFS channel sits a sensible distance above the floor.
        def scr(level_db, headroom=90.0):
            return floor_dbm + headroom + level_db

        # Per-bin dB deviation of REAL noise: each FFT bin's power is exponentially
        # distributed, so in dB it's 10*log10(exp) -- a spiky "grass" with a long
        # DOWNWARD tail (occasional +7, deep dropouts to -15 and below). Its low
        # tail must sit ON the display floor so noise rises FROM the floor upward,
        # not as a band floating above it. (G0JKN: noise was "around a level, not
        # from 0" -- the old scr()+90 anchoring left a big dead gap below it.)
        def noise_bin_db():
            u = self.rng.random()
            if u <= 0.0:
                u = 1e-9
            return 10.0 * math.log10(-math.log(u))

        # A noise channel BECOMES THE FLOOR: its grass BOTTOMS OUT on the display
        # floor and rises up from there, exactly like turning on noise lifts a real
        # radio's baseline. We anchor the noise's LOW tail at floor_dbm: the mean
        # sits `height` above the floor and the deep exponential dropouts reach down
        # to it. height scales with level_db (louder noise = higher grass).
        #   noise_bin_db() spans about -15..+7 (median ~-2.5); floor it at 0 so the
        #   dropouts land ON the floor rather than below it.
        def noise_grass(level_db):
            height = 8.0 + (level_db + 60.0) * 0.6      # -60->8dB .. 0->44dB tall
            return floor_dbm + max(0.0, height + noise_bin_db())

        for name, ch in self.channels.items():
            if not ch["enabled"]:
                continue
            lvl = ch["level_db"]
            if name == "white":
                for b in range(n):
                    add_db(b, noise_grass(lvl))
            elif name == "pink":
                # sloped: taller (louder) at the low edge, tapering up-band
                for b in range(n):
                    frac = b / max(1, n - 1)
                    add_db(b, noise_grass(lvl) - 14.0 * frac)
            elif name == "qrn":
                # impulse = broadband vertical streak, only on frames one fires
                if self.rng.random() < min(0.9, ch.get("rate", 12.0) / 20.0):
                    top = scr(lvl) + 6.0
                    for b in range(n):
                        add_db(b, top + self.rng.uniform(-6.0, 2.0))
            elif name == "crashes":
                if self.rng.random() < 0.15:
                    top = scr(lvl)
                    for b in range(n):
                        add_db(b, top + self.rng.uniform(-8.0, 2.0))
            elif name == "birdie":
                b = bin_of(ch.get("hz", 1000.0))
                for d in (-1, 0, 1):
                    add_db(b + d, scr(lvl) + (10.0 if d == 0 else -6.0))
            elif name == "powerline":
                f0 = ch.get("freq", 60.0)
                for h in range(1, 12, 2):
                    add_db(bin_of(f0 * h), scr(lvl) - 3.0 * (h // 2))
            elif name == "hash":
                # broadband clumps — real noise texture (grassy, not a ribbon)
                for b in range(n):
                    add_db(b, scr(lvl) - 6.0 + noise_bin_db())
            elif name == "woodpecker":
                if self.rng.random() < 0.5:
                    top = scr(lvl)
                    for b in range(n):
                        add_db(b, top + self.rng.uniform(-5.0, 3.0))
            elif name == "cw":
                b = bin_of(ch.get("hz", 700.0))
                for d in (-1, 0, 1):
                    add_db(b + d, scr(lvl) + (8.0 if d == 0 else -8.0))
            elif name == "voice":
                # SSB-ish band 300..2700 Hz above the VFO
                for b in range(bin_of(300), bin_of(2700)):
                    add_db(b, scr(lvl) - 6.0 + self.rng.uniform(-6.0, 3.0))
        # TNF display notch: carve a dip to the floor at each notched offset, so a
        # notch placed on the birdie removes its line from the waterfall too. Width
        # in bins from the notch's audio width (min a couple of bins to be visible).
        for f_hz in self._notch_hz:
            nb = bin_of(f_hz)
            half = max(1, int((self._notch_width_hz.get(round(f_hz), 100)) / (2 * hz_per_bin)))
            for b in range(nb - half, nb + half + 1):
                if 0 <= b < n:
                    out[b] = floor_dbm
        return out


# Query params understood by the noise bench (shared by the web panel and the
# Claude/QA hooks). Every channel is driven through the SAME /set surface:
#
#   /set?noise_<chan>=1|0                     enable / disable a channel
#   /set?noise_<chan>_level=<dBFS>            set its level (e.g. -20)
#   /set?noise_<chan>_<knob>=<val>            set a per-channel knob (rate/hz/freq/prf/wpm)
#   /set?noise_voice_wav=<path>               WAV file for the voice channel
#   /set?noise_reset=1                        disable every channel
#   /set?noise_preset=<name>                  load a named scene (see NOISE_PRESETS)
#
# and /state returns radio.noise_mixer.snapshot() for readback / assertions.
_NOISE_KNOBS = {"level", "rate", "hz", "freq", "prf", "wpm"}


def apply_noise_params(radio, q):
    mx = radio.noise_mixer
    getf = lambda k: float(q[k][0])
    if q.get("noise_reset", ["0"])[0] == "1":
        for ch in mx.channels.values():
            ch["enabled"] = False
        log("[noise] all channels disabled")
    if "noise_voice_wav" in q:
        radio.noise_wav_path = q["noise_voice_wav"][0] or None
        mx.voice_wav = None                       # force lazy re-open with new path
        log(f"[noise] voice WAV -> {radio.noise_wav_path}")
    if "noise_preset" in q:
        load_noise_preset(radio, q["noise_preset"][0])
    for key in list(q.keys()):
        if not key.startswith("noise_"):
            continue
        parts = key[len("noise_"):].split("_")     # <chan> or <chan>_<knob>
        chan = parts[0]
        if chan not in mx.channels:
            continue
        ch = mx.channels[chan]
        if len(parts) == 1:                        # noise_<chan> = enable toggle
            ch["enabled"] = q[key][0] == "1"
            log(f"[noise] {chan} {'ON' if ch['enabled'] else 'off'} "
                f"({ch.get('level_db', 0):.0f} dB)")
        elif parts[1] in _NOISE_KNOBS:
            knob = "level_db" if parts[1] == "level" else parts[1]
            try:
                ch[knob] = getf(key)
            except (ValueError, KeyError):
                pass


# Named scenes — reproducible for the QA suite, one-click in the panel.
NOISE_PRESETS = {
    "quiet-20m":  {"pink": -30},
    "night-40m":  {"pink": -22, "qrn": -24, "crashes": -22},
    "storm":      {"pink": -20, "qrn": -12, "crashes": -10},
    "noisy-qth":  {"pink": -24, "powerline": -16, "hash": -18},
    "birdie-hell":{"pink": -28, "birdie": -18, "powerline": -22},
    "voice-in-noise": {"voice": -14, "pink": -22, "qrn": -20},
    "cw-in-noise":    {"cw": -16, "pink": -24, "qrn": -22},
}


def load_noise_preset(radio, name):
    mx = radio.noise_mixer
    scene = NOISE_PRESETS.get(name)
    if scene is None:
        log(f"[noise] unknown preset '{name}'"); return
    for ch in mx.channels.values():                # presets are absolute: start clean
        ch["enabled"] = False
    for chan, level in scene.items():
        if chan in mx.channels:
            mx.channels[chan]["enabled"] = True
            mx.channels[chan]["level_db"] = level
    log(f"[noise] preset '{name}' -> {', '.join(scene)}")


class Radio:
    def __init__(self, ip, ae_ip, pattern="ramp", bins=BINS, fps=FPS, width_khz=SIGNAL_WIDTH_KHZ,
                 port=DEFAULT_PORT, radio_id=0, model=MODEL, serial_prefix=SERIAL_PREFIX):
        self.ip, self.ae_ip = ip, ae_ip
        self.port = port                # control/data port we bind + advertise (discovery dest stays 4992)
        self.model = model if model in MODELS else MODEL
        self.max_slices = MODELS.get(self.model, {}).get("slices", 4)   # advertised cap (multi-slice TODO)
        # SCU count gates 2-SCU-only features. AE decides diversity from the
        # MODEL it is told, so this is really a self-consistency guard: it stops
        # the sim inventing a diversity pair on a radio that could not do one.
        self.scu = MODELS.get(self.model, {}).get("scu", 1)
        # Per-radio identity, so many radios can live in one process (the "rack").
        # Each value must be unique per radio or AE cross-wires them; derived from
        # radio_id off the base constants below.
        self.radio_id = radio_id
        # Serial prefix is overridable (--serial) because AE's `connect local
        # serial <s>` matches on SERIAL: two sims advertising the default on one
        # network (e.g. a local one plus the hub's bench) are indistinguishable
        # to that verb, and AE silently takes whichever it found first. The
        # per-radio :02d suffix is preserved so rack mode stays unique.
        self.serial = f"{serial_prefix}{radio_id:02d}"
        self.client_handle = HANDLE + radio_id   # NB: NOT self.handle — that's the method!
        self.handle_hex = f"{self.client_handle:08X}"
        self.pan_id = PAN_ID + radio_id * 16     # FFT stream-id BASE; AE stacks a panadapter per +RX,
        self.wf_id = WF_ID + radio_id * 16       # waterfall base. pan k -> base+k (16 ids/radio, no overlap)
        self.pans = {}                           # pan_id -> {"wf_id":int,"slice":global_idx|None,"y_pixels":int}
        self.pan_seq = 0                          # next panadapter index for this radio
        self.meter_sid = METER_SID + radio_id    # meter VITA stream id (PCC 0x8002)
        self.pattern = pattern if pattern in PATTERNS else "ramp"
        self.bins, self.fps = bins, fps
        self.sig_width_khz = width_khz
        self.noise_floor_dbm = -120.0   # live (control panel, dBm; ~S1)
        self.sig_level_dbm = S9_DBM     # live (control panel, dBm; default S9 = -73)
        self.noise_color = 0.0          # live (noise tilt: 0=white, >0=pink)
        self.noise_cal_full = True      # live: noise_cal full-span (standard) vs width-limited
        self.vita_dest = None
        self.run = True
        self.send_lock = threading.Lock()   # serialize TCP writes: stream thread (status) vs command thread (replies)
        self.streaming = False
        self.paused = False             # Stop/Go: when True, stay connected but send NO pan/wf/meter data
                                        # (AE keeps the radio; panadapter + waterfall go dead). Go resumes.
        self.ae_peer_ip = None          # only accept the UDP prime from the connected AE
        self.y_pixels, self.min_dbm, self.max_dbm = 700, -130.0, -20.0
        self.center_mhz, self.span_mhz = CENTER_MHZ, SPAN_MHZ   # track AE's actual pan view
        self.slice_freq = CENTER_MHZ                            # active slice freq (= VFO); mirrors slices[active]
        self.slice_mode = "USB"
        self.slices = {}            # index -> {"freq":MHz,"mode":str,"active":bool}; AE adds via +RX (slice create)
        # Tracking Notch Filters (waterfall notch). AE sends 'tnf create/set/remove'
        # (radio-side, like ANF) and DRAWS the marker only when we echo status back.
        # Unlike ANF, we emulate it: notch the mix audio + spectrum render so the
        # tone vanishes from ear AND eye. id -> {"freq":MHz,"width":Hz,"depth":n,"permanent":0|1}
        self.tnfs = {}
        self._next_tnf_id = 1
        self.active_slice = 0       # the slice the pan/waterfall/primary carrier follow
        self.conn = None            # active TCP conn (so the control panel can push TX status)
        self.tx_mox = False         # live: TX keyed (panel toggle or AE's 'transmit set mox=1')
        # --txlog: characterise UDP arriving FROM AE (TX audio discovery).
        self.txlog = False
        self.txlog_interval = 5.0
        self._txlog_state = {"streams": {}, "total": 0, "last_report": time.time()}
        # ---- DAX TX chain observer (the #4510 "audio arrives but no RF" family) ----
        # "No RF" is a symptom-SINK: DAX TX is a six-stage chain and every
        # mid-chain failure presents identically at the far end, which is why
        # the family never converged on one fix. The sim tracks each stage it
        # can see (2..5; stage 1 is app-side capture, stage 6 is the verdict)
        # and names the FIRST missing one — turning the symptom into a
        # diagnosis. GET /txchain on the control panel returns the table.
        self.daxtx_id = None            # 'stream create type=dax_tx' id
        self.daxtx_tx = False           # radio-side TX-source flag ('stream set <id> tx=1')
        # multiFLEX CONTENTION, not a firmware switch (corrected on #4640 from
        # FlexLib: `stream set <id> tx=1` is RequestTX — ownership arbitration
        # between GUI clients, not a per-transmission step). On a single-client
        # radio the create GRANTS the TX bit implicitly and the status echoes
        # tx=1; nothing needs sending. When ANOTHER client holds it, the same
        # client code silently produces no RF — the real "works for many, fails
        # for some" split, and state-dependent rather than device-dependent, so
        # it can recur on the very radio that worked yesterday.
        self.daxtx_foreign_owner = 0    # 0 = we own it; else the handle holding TX
        self.tx_dax = False             # 'transmit set dax=1'
        self.daxtx_pkts = 0             # VITA audio packets seen on :4991
        self.daxtx_bytes = 0
        self.daxtx_last_pkt = 0.0       # monotonic-ish wall time of the last packet
        self._txchain_was_keyed = False # rising-edge latch for the verdict log line
        # One :4991 observer per process (rack mode: radio 0 owns the counters —
        # AE only runs one DAX TX stream at a time, so that is not a loss).
        global _DAXTX_LISTENER_STARTED
        if not _DAXTX_LISTENER_STARTED:
            _DAXTX_LISTENER_STARTED = True
            threading.Thread(target=self._daxtx_audio_listener,
                             daemon=True, name="daxtx-audio-observer").start()
        self.tx_tune = False
        self.tx_power_w = 100.0      # live: forward power (W) reported while TX
        # What the RADIO generates while tuning: "" (plain carrier) or "two_tone".
        # AE sets this via 'transmit set tune_mode='; see tx_envelope().
        self.tune_mode = ""
        self.tx_swr = 1.2           # live: SWR reported while TX
        self.amp_handle = 0xA5000000 + self.radio_id  # radio-side amplifier (PGXL) object handle
        self.amp_operate = True      # PGXL OPERATE(1)/STANDBY(0); AE toggles via 'amplifier set operate='
        self.amp_temp_c = 38.0       # amp heatsink; warms while keyed, cools when idle
        self.cwx_wpm = 20           # CWX keyer speed (AE drives via 'cwx wpm')
        self.cwx_active = False      # an AE-initiated CWX transmission is in progress
        self.cwx_schedule = []
        self.cwx_dur = 0.0
        self.cwx_t0 = 0.0
        self.cwx_char_ends = []
        self.cwx_sent = 0
        self.qsk = False             # full break-in (RX between elements); AE sets via cwx qsk_enabled
        self.enabled = True          # "power": when False, stop advertising (radio drops off AE)
        self.last_vfo_dbm = -130.0   # last level at the VFO — drives the rack strip's signal meter
        # remote_audio_rx stream state
        self.audio_stream_id = None
        self.audio_stop = threading.Event()
        self.audio_thread = None
        self.audio_tone_hz = 440.0 + radio_id * 220.0  # distinct tone per radio (A4, C#5, …)
        self.audio_reduced_bw = False  # set True when AE sends 'client set send_reduced_bw_dax=1'
        self.audio_source   = AUDIO_SRC_SILENCE  # default silent: SILENCE | TONE | NOISE | WAV
        self.audio_wav_path = None             # filesystem path to WAV file (used when source=wav)
        # HF noise bench — additive mixer for AE noise-reduction testing. When
        # ANY channel is enabled it OVERRIDES audio_source and streams the
        # summed voice+CW+noise mix instead. voice channel plays this WAV.
        self.noise_mixer    = NoiseMixer()
        self.noise_wav_path = None             # WAV for the mixer's 'voice' channel
        # dax_rx stream state (RADE / digital decode path). AE arms this via
        # 'stream create type=dax_rx dax_channel=N' when a slice goes RADE/DIGU;
        # we register an id back so AE feeds these packets to its host-side decoder.
        self.dax_channel = None                # the dax_channel AE requested (1..4), or None

    @property
    def tx_on(self):
        return self.tx_mox or self.tx_tune

    # ---- DAX TX chain observer (#4510) -----------------------------------
    def txchain(self):
        """The 'audio arrives but no RF' chain, radio-side view.

        Stage 1 (app-side audio capture) is invisible from here by design —
        this is the radio's testimony, which is exactly what the divergent
        no-RF family lacks: the operator sees only the two chain ends, so
        every mid-chain failure looks identical. The verdict names the FIRST
        missing stage instead."""
        audio_fresh = (self.daxtx_last_pkt > 0
                       and (time.time() - self.daxtx_last_pkt) < 2.0)
        stages = [
            ("2: dax_tx stream exists",           self.daxtx_id is not None),
            ("3: TX audio arriving on :4991",     audio_fresh),
            ("4: radio keyed with dax=1",         bool(self.tx_on) and self.tx_dax),
            ("5: stream claimed as TX source",    self.daxtx_tx),   # foreign owner => False
        ]
        first_missing = next((name for name, ok in stages if not ok), None)
        return {
            "stages": {name: ok for name, ok in stages},
            "stream_id": f"0x{self.daxtx_id:08X}" if self.daxtx_id else None,
            "audio_pkts": self.daxtx_pkts,
            "audio_bytes": self.daxtx_bytes,
            "tx_owner": ("us" if self.daxtx_tx
                         else f"0x{self.daxtx_foreign_owner:08X}"
                         if self.daxtx_foreign_owner else "unclaimed"),
            "contention": self.daxtx_foreign_owner != 0,
            "keyed": bool(self.tx_on),
            "dax": self.tx_dax,
            "verdict": ("RF WOULD BE PRODUCED" if first_missing is None
                        else f"NO RF — first missing stage: {first_missing}"),
        }

    def txchain_edge(self):
        """Log the chain verdict once per key-down while DAX is the TX source
        setting — the moment an operator would be transmitting silence."""
        keyed = bool(self.tx_on)
        if keyed and not self._txchain_was_keyed and self.tx_dax:
            log(f"[txchain] {self.txchain()['verdict']}")
        self._txchain_was_keyed = keyed

    def _daxtx_audio_listener(self):
        # AE hardcodes DAX TX audio to <radio>:4991 (PanadapterStream.cpp:
        # m_radioPort = 4991), which nothing else here binds — a dedicated
        # observer socket counts stage 3. Best-effort: if the port is taken
        # (SmartSDR DAX on this host?), log once and run without stage 3.
        try:
            s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            s.bind(("0.0.0.0", DAXTX_AUDIO_PORT))
        except OSError as e:
            log(f"[txchain] :{DAXTX_AUDIO_PORT} observer unavailable ({e}) — "
                f"stage-3 audio counting disabled")
            return
        log(f"[txchain] observing DAX TX audio on UDP :{DAXTX_AUDIO_PORT}")
        while True:
            try:
                data, _addr = s.recvfrom(4096)
            except OSError:
                return
            self.daxtx_pkts += 1
            self.daxtx_bytes += len(data)
            self.daxtx_last_pkt = time.time()

    def set_power(self, on):
        # Rack "power button": off -> stop streaming + drop AE's connection + stop
        # advertising (radio vanishes from AE). on -> discovery resumes, AE re-finds it.
        self.enabled = bool(on)
        if not on:
            self.streaming = False
            c = self.conn
            if c:
                try: c.shutdown(socket.SHUT_RDWR); c.close()
                except OSError: pass

    def state(self):
        # Snapshot for the rack strip (polled by the panel).
        return {"id": self.radio_id, "serial": self.serial, "model": self.model, "ip": self.ip,
                "on": self.enabled, "connected": self.conn is not None,
                "freq": round(self.slice_freq, 5), "mode": self.slice_mode, "tx": self.tx_on,
                "pattern": self.pattern, "power_w": round(self.tx_power_w),
                "meter_dbm": round(self.last_vfo_dbm, 1),
                "slices": len(self.slices), "max_slices": self.max_slices,
                "noise": self.noise_mixer.snapshot(),           # HF noise bench state
                "noise_active": self.noise_mixer.any_enabled(),
                "noise_wav": self.noise_wav_path}

    def dbm_to_pixel(self, dbm):
        p = (self.max_dbm - dbm) / (self.max_dbm - self.min_dbm) * (self.y_pixels - 1)
        return max(0, min(self.y_pixels - 1, int(round(p))))

    def dbm_to_wf_raw(self, dbm):
        frac = max(0.0, min(1.0, (dbm - self.min_dbm) / (self.max_dbm - self.min_dbm)))
        return int(round((WF_FLOOR_VAL + frac * (WF_PEAK_VAL - WF_FLOOR_VAL)) * 128))

    def discovery_loop(self):
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        s.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
        dests = [("255.255.255.255", DISCOVERY_PORT)] + ([(self.ae_ip, DISCOVERY_PORT)] if self.ae_ip else [])
        while self.run:
            if self.enabled:                       # "powered off" -> stop advertising
                msg = (f"name=flex-sim model={self.model} serial={self.serial} version={VERSION} "
                       f"ip={self.ip} port={self.port} status=Available mf_enable=1 "
                       f"slices={self.max_slices} pans={self.max_slices} "   # capacity -> AE enables +RX
                       f"max_licensed_version=3").encode()   # rebuilt each tick -> live model change propagates
                for d in dests:
                    try: s.sendto(msg, d)
                    except OSError: pass
            time.sleep(1.0)

    def _log_inbound_datagram(self, data, addr):
        """Characterise UDP arriving FROM AetherSDR (--txlog).

        Answers, from observation rather than from reading AE's source, the
        questions the TX-audio/parrot design has to settle first (see
        TX_AUDIO_DESIGN.md): which stream AE actually uses for a given mode,
        what the packet geometry is, and at what rate they arrive.

        Prints a one-line summary per distinct stream id, then a periodic
        rate/size report — NOT a line per packet, which at ~375 packets/s for
        24 kHz audio would bury everything else.

        VITA-49 header (see vita_header()): word 0 packs type/count/size,
        word 1 is the stream id, then OUI/ICC/PCC, then timestamps. The PCC
        (packet class code) is what AE uses to pick a decoder, so it is the
        single most useful field for identifying the payload format.
        """
        # Ignore our OWN discovery broadcast looping back on the same host --
        # it is plain ASCII key=value ("name=flex-sim model=..."), not VITA, and
        # in same-host mode it lands right back here.
        if data[:5] == b"name=":
            return
        now = time.time()
        st = self._txlog_state
        st["total"] += 1

        sid = pcc = None
        if len(data) >= 28:
            try:
                sid = struct.unpack(">I", data[4:8])[0]
                pcc = struct.unpack(">H", data[14:16])[0]
            except struct.error:
                pass

        key = (sid, pcc, len(data))
        seen = st["streams"]
        if key not in seen:
            seen[key] = {"count": 0, "first": now}
            log(f"[txlog] NEW inbound stream: id=0x{sid:08X} pcc=0x{pcc:04X} "
                f"payload={len(data)}B from {addr[0]}  (mox={1 if self.tx_mox else 0})"
                if sid is not None else
                f"[txlog] NEW inbound datagram: {len(data)}B from {addr[0]} (non-VITA?)")
            # First few bytes help identify a format the header alone doesn't.
            log(f"[txlog]   head={data[:16].hex(' ')}")
        seen[key]["count"] += 1

        # Periodic summary: rate per stream + whether TX is keyed. One line per
        # stream per interval, so a long capture stays readable.
        if now - st["last_report"] >= self.txlog_interval:
            span = now - st["last_report"]
            log(f"[txlog] --- {st['total']} datagrams in {span:.1f}s  "
                f"(mox={1 if self.tx_mox else 0} tune={1 if self.tx_tune else 0}) ---")
            for (s_id, s_pcc, size), info in sorted(seen.items(), key=lambda kv: -kv[1]["count"]):
                delta = info["count"] - info.get("reported", 0)
                info["reported"] = info["count"]
                if delta:
                    ident = f"id=0x{s_id:08X} pcc=0x{s_pcc:04X}" if s_id is not None else "non-VITA"
                    log(f"[txlog]   {ident} {size}B  {delta} pkt  {delta / span:.1f}/s")
            st["last_report"] = now
            st["total"] = 0

    def prime_loop(self):
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        s.bind((self.ip, self.port))
        while self.run:
            try:
                data, addr = s.recvfrom(2048)
            except OSError:
                break
            if self.ae_peer_ip and addr[0] == self.ae_peer_ip and not self.vita_dest:
                self.vita_dest = addr
                log(f"[udp] captured VITA dest from prime: {addr} ({len(data)}B)")
            if self.txlog:
                self._log_inbound_datagram(data, addr)

    def serve(self):
        srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        srv.bind((self.ip, self.port)); srv.listen(1)   # bind our own IP (rack: each radio a distinct IP)
        log(f"[tcp] listening on :{self.port}  (radio ip {self.ip}, pattern={self.pattern})")
        while self.run:
            conn, addr = srv.accept()
            log(f"[tcp] AE connected from {addr}")
            self.ae_peer_ip = addr[0]; self.vita_dest = None
            try: self.handle(conn)
            except Exception as e: log("[tcp] conn error:", e)
            finally:
                conn.close()
                self.streaming = False; self.ae_peer_ip = None; self.conn = None; self.vita_dest = None
                self.audio_stop.set(); self.audio_stream_id = None; self.dax_channel = None
                # Free this client's receivers/panadapters on disconnect, like a real radio.
                # Without it, stale slices/pans pile up across reconnects and the slice
                # lettering drifts (A -> B -> C on each new connection).
                self.slices.clear(); self.pans.clear(); self.pan_seq = 0; self.active_slice = 0

    def handle(self, conn):
        self.conn = conn
        with self.send_lock:
            conn.sendall(f"V{VERSION}\n".encode())
            conn.sendall(f"H{self.handle_hex}\n".encode())
        log(f"[tcp] sent V{VERSION} / H{self.handle_hex}")
        buf = b""
        while self.run:
            chunk = conn.recv(4096)
            if not chunk: break
            buf += chunk
            while b"\n" in buf:
                line, buf = buf.split(b"\n", 1)
                line = line.decode(errors="replace").strip()
                if line: self.on_line(conn, line)

    def reply(self, conn, seq, body="", code=0):
        with self.send_lock:
            conn.sendall(f"R{seq}|{code:X}|{body}\n".encode())

    def status(self, conn, obj_kvs):
        with self.send_lock:
            conn.sendall(f"S{self.handle_hex}|{obj_kvs}\n".encode())

    def on_line(self, conn, line):
        if not line.startswith("C") or "|" not in line:
            log("[<-] (non-C)", line); return
        head, cmd = line.split("|", 1)
        seq, c = head[1:], cmd.strip()
        log("[<-]", f"C{seq}|{c}")
        if c.startswith("client gui"):
            self.reply(conn, seq, str(uuid.uuid4()).upper())
        elif c.startswith("client set"):
            if "send_reduced_bw_dax=1" in c:
                self.audio_reduced_bw = True
            elif "send_reduced_bw_dax=0" in c:
                self.audio_reduced_bw = False
            self.reply(conn, seq)
        elif c == "slice list":
            self.reply(conn, seq, "")
        elif c.startswith("display panafall create") or c.startswith("display pan create"):
            pid = self._new_pan()                          # AE's +RX = a NEW stacked panadapter
            self.reply(conn, seq, f"0x{pid:08X},0x{self.pans[pid]['wf_id']:08X}")
            self.emit_pan_status(conn, pid)
        elif c.startswith("display pan remove") or c.startswith("display panafall remove"):
            self.reply(conn, seq)
            for t in c.split():
                if t.lower().startswith("0x"):
                    pid = int(t, 16)
                    if self.pans.pop(pid, None):           # stop streaming the removed panadapter
                        for g, sl in list(self.slices.items()):
                            if sl.get("pan") == pid: self.slices.pop(g, None)
                        log(f"[pan] remove -> 0x{pid:08X}  ({len(self.pans)} left)")
                    break
        elif c.startswith("slice create"):
            kvs = parse_kvs(c)
            idx = self._next_slice_index()
            if idx is None:                                # model's slice budget exhausted
                self.reply(conn, seq, code=0x50000000)     # -> AE: "No available slices"
                log(f"[slice] create refused: all {self.max_slices} slices in use")
                return
            pid = self._pan_from_kvs(kvs)                  # the panadapter this receiver lands on
            freq = float(kvs["freq"]) if "freq" in kvs else self.slice_freq
            mode = kvs.get("mode", self.slice_mode)
            for s in self.slices.values(): s["active"] = False
            # A real radio always has exactly one TX slice; the first one created
            # takes it. Diversity children are RX-only and never claim it.
            is_tx = not any(s.get("tx") for s in self.slices.values())
            self.slices[idx] = {"freq": freq, "mode": mode, "active": True, "pan": pid,
                                "tx": is_tx}
            self.pans[pid]["slice"] = idx
            self.pans[pid]["center"] = freq                # pan frames on its slice at creation
            self.active_slice = idx
            self._sync_active_slice()                      # mirror active slice -> slice_freq/mode + pan centre
            self.reply(conn, seq, str(idx))
            log(f"[slice] create -> slice {idx} @ {freq:.5f} {mode} on pan 0x{pid:08X}  ({len(self.slices)}/{self.max_slices})")
            self.emit_slice_status(conn, idx)
            self.emit_meter_status(conn)                   # (re)define S-meters incl. the new slice
            self.emit_pan_status(conn, pid)                # centre this pan/waterfall on its slice
        elif c.startswith("display pan set"):
            kvs = parse_kvs(c)
            pid = None                                     # which panadapter this set targets
            for t in c.split():
                if t.lower().startswith("0x"):
                    try: pid = int(t, 16)
                    except ValueError: pid = None
                    break
            pan = self.pans.get(pid)
            if "ypixels" in kvs: self.y_pixels = max(2, int(kvs["ypixels"]))
            if "y_pixels" in kvs: self.y_pixels = max(2, int(kvs["y_pixels"]))
            if "min_dbm" in kvs: self.min_dbm = float(kvs["min_dbm"])
            if "max_dbm" in kvs: self.max_dbm = float(kvs["max_dbm"])
            if "bandwidth" in kvs: self.span_mhz = float(kvs["bandwidth"])
            # Retune THIS panadapter by absolute centre= or by band= (AE's band buttons).
            # Per-pan, not radio-global, so stacked pans can sit on different bands; move
            # the pan's slice with it (a band change retunes its receiver, like a real radio).
            new_center = None
            if "band" in kvs:
                bc = BAND_CENTERS_MHZ.get(str(kvs["band"]).strip().lower())
                if bc is not None: new_center = bc
            if "center" in kvs:
                try: new_center = float(kvs["center"])
                except ValueError: pass
            if new_center is not None:
                if pan is not None:
                    pan["center"] = new_center
                    sidx = pan.get("slice")
                    if sidx is not None and sidx in self.slices:
                        self.slices[sidx]["freq"] = new_center
                else:
                    self.center_mhz = new_center           # no pan id: legacy radio-global behaviour
                self._sync_active_slice()
            self.reply(conn, seq)
            if pan is not None and new_center is not None:
                self.emit_pan_status(conn, pid)            # re-announce the pan's new centre to AE
                sidx = pan.get("slice")
                if sidx is not None and sidx in self.slices:
                    self.emit_slice_status(conn, sidx)
        elif c == "sub radio all":
            self.reply(conn, seq)
            self.emit_radio_status(conn)                   # slice/pan capability -> AE enables +RX
        elif c.startswith("info"):
            self.reply(conn, seq, f"model={self.model},chassis_serial={self.serial},callsign=SDRSIM,"
                                  f"name=flex-sim,software_ver={VERSION},ip={self.ip}")
        elif c == "mic list":
            self.reply(conn, seq, "MIC,LINE")
        elif c.startswith("client udpport"):
            # AE announces the UDP port it listens on for VITA data. On a single
            # host (AE owns :4992), AE's prime packet to our advertised port may
            # never arrive — so seed vita_dest directly from this announcement +
            # the known peer IP, instead of relying solely on prime_loop. The
            # prime path still works as a fallback (e.g. WSL2 separate stacks).
            try:
                up = int(c.split()[-1])
                if self.ae_peer_ip and not self.vita_dest:
                    self.vita_dest = (self.ae_peer_ip, up)
                    log(f"[udp] VITA dest from client udpport: {self.vita_dest}")
                else:
                    log("[tcp] AE udpport =", up)
            except ValueError:
                pass
            self.reply(conn, seq)
        elif c.startswith("stream create"):
            kvs = parse_kvs(c)
            if kvs.get("type") == "remote_audio_rx":
                sid = AUDIO_SID_BASE + self.radio_id
                self.audio_stream_id = sid
                self.dax_channel = None
                self.reply(conn, seq, f"0x{sid:08X}")
                self._start_audio_thread()
            elif kvs.get("type") == "dax_rx":
                # AE arms a DAX RX channel for RADE / digital decode. It relies on a
                # 'stream <id> type=dax_rx dax_channel=N client_handle=<ours>' STATUS line
                # (TciServer::statusReceived -> registerDaxStream) to learn the id; the
                # command reply alone is NOT enough. Stream the golden clip on this id with
                # the same PCC_IF_NARROW format -> AE routes it to feedRxAudio -> RADE decode.
                try:
                    ch = int(kvs.get("dax_channel", "1"))
                except ValueError:
                    ch = 1
                ch = max(1, min(4, ch))
                sid = DAX_SID_BASE + self.radio_id * 4 + (ch - 1)
                self.audio_stream_id = sid
                self.dax_channel = ch
                self.reply(conn, seq, f"0x{sid:08X}")
                # The registration status line — must carry our client_handle so AE's
                # streamStatusBelongsToUs() accepts it, and the dax_channel it asked for.
                self.status(conn, f"stream 0x{sid:08X} type=dax_rx dax_channel={ch} "
                                  f"client_handle=0x{self.handle_hex}")
                log(f"[dax] registered dax_rx stream 0x{sid:08X} channel={ch}")
                self._start_audio_thread()
            elif kvs.get("type") == "dax_tx":
                # The TX half of the #4510 chain. Real firmware replies with an
                # id and a registration status; whether it ALSO marks the fresh
                # stream tx=1 unprompted is exactly the behaviour split behind
                # the "no RF" family — see daxtx_foreign_owner for the contention model.
                sid = DAXTX_SID_BASE + self.radio_id
                self.daxtx_id = sid
                # Create grants TX ownership implicitly — UNLESS a foreign
                # client holds it, which is the whole failure mode.
                self.daxtx_tx = (self.daxtx_foreign_owner == 0)
                self.daxtx_pkts = 0
                self.daxtx_bytes = 0
                self.daxtx_last_pkt = 0.0
                self.reply(conn, seq, f"0x{sid:08X}")
                owner = (self.handle_hex if self.daxtx_tx
                         else f"{self.daxtx_foreign_owner:08X}")
                self.status(conn, f"stream 0x{sid:08X} type=dax_tx "
                                  f"client_handle=0x{owner} "
                                  f"tx={1 if self.daxtx_tx else 0}")
                log(f"[txchain] dax_tx stream created id=0x{sid:08X} "
                    f"tx={int(self.daxtx_tx)}"
                    + ("" if self.daxtx_tx else
                       f" — TX bit held by FOREIGN client 0x{owner} (multiFLEX contention)"))
            else:
                self.reply(conn, seq, "0x48000000")
        elif c.startswith("stream set"):
            # AE sends 'stream set 0x… tx=1|0' to claim/release a dax_tx stream
            # as the transmit modulation source — the stage-5 command of the
            # chain, and the one the ordinary DAX/TCI path never sends (#4510:
            # only the WSPR path is wired to it). Echo a status so AE's
            # ownership-adoption path (m_daxTxActive) sees the radio's answer.
            kvs = parse_kvs(c)
            sid = None
            for tok in c.split():
                if tok.lower().startswith("0x"):
                    try:
                        sid = int(tok, 16)
                    except ValueError:
                        pass
                    break
            if sid is not None and sid == self.daxtx_id and "tx" in kvs:
                self.daxtx_tx = kvs["tx"] == "1"
                log(f"[txchain] stream 0x{sid:08X} tx={kvs['tx']} — TX source "
                    f"{'CLAIMED' if self.daxtx_tx else 'released'} by client")
                self.status(conn, f"stream 0x{sid:08X} type=dax_tx "
                                  f"client_handle=0x{self.handle_hex} tx={kvs['tx']}")
            self.reply(conn, seq)
        elif c.startswith("stream remove"):
            for tok in c.split():
                if tok.lower().startswith("0x"):
                    if int(tok, 16) == self.audio_stream_id:
                        self.audio_stop.set()
                        self.audio_stream_id = None
                        self.dax_channel = None
                    if self.daxtx_id is not None and int(tok, 16) == self.daxtx_id:
                        self.daxtx_id = None
                        self.daxtx_tx = False
                        log("[txchain] dax_tx stream removed")
            self.reply(conn, seq)
        elif c.startswith("transmit tune"):
            # POSITIONAL form: "transmit tune 1|0". AE's two-tone/ATU tune uses
            # this, NOT "transmit set tune=" -- measured on the wire 2026-07-22.
            # Handling only the 'set' form meant every tune was silently ignored:
            # the sim never keyed, so power stayed at the un-keyed floor.
            parts = c.split()
            self.tx_tune = (len(parts) > 2 and parts[2] == "1")
            self.reply(conn, seq)
            self.emit_transmit_status()
            self.txchain_edge()
            log(f"[tx] tune {'ON' if self.tx_tune else 'OFF'}"
                f"{' (' + self.tune_mode + ')' if self.tune_mode else ''}")
        elif c.startswith("transmit set"):                 # AE keys TX (MOX / TUNE)
            kvs = parse_kvs(c)
            if "mox" in kvs:  self.tx_mox = kvs["mox"] == "1"
            if "tune" in kvs: self.tx_tune = kvs["tune"] == "1"
            # Stage 4 of the #4510 chain: dax=1 tells the radio the TX audio
            # SOURCE is DAX — necessary but not sufficient; stage 5 (stream
            # tx=1) still has to name WHICH stream modulates.
            if "dax" in kvs:  self.tx_dax = kvs["dax"] == "1"
            # tune_mode selects what the RADIO generates while tuning. AE's
            # two-tone test sends 'tune_mode=two_tone' then 'transmit tune 1'
            # and sends NO audio of its own -- measured on the wire 2026-07-22.
            # A real Flex synthesises the tones in hardware, so we must too.
            if "tune_mode" in kvs:
                self.tune_mode = kvs["tune_mode"]
                log(f"[tx] tune_mode={self.tune_mode}")
            # Track the operator's power setting so the simulated envelope scales
            # with it -- turning RF Power down must visibly reduce reported watts.
            for k in ("rfpower", "tunepower"):
                if k in kvs:
                    try: self.tx_power_w = max(0.0, float(kvs[k]))
                    except ValueError: pass
            self.reply(conn, seq)
            self.emit_transmit_status()
            self.txchain_edge()
        elif c == "sub amplifier all":                      # AE subscribes to amplifier presence
            self.reply(conn, seq)
            self.emit_amplifier_status(conn)                # -> AE hasAmplifier=true; AMP applet un-gates
        elif c.startswith("amplifier set"):                 # AE toggles OPERATE / STANDBY
            kvs = parse_kvs(c)
            if "operate" in kvs: self.amp_operate = (kvs["operate"] == "1")
            self.reply(conn, seq)
            self.emit_amplifier_status(conn)
        elif c.startswith("cwx "):                          # AE's CW text keyer
            self.handle_cwx(c)
            self.reply(conn, seq)
        elif c.startswith("slice set"):
            kvs = parse_kvs(c)
            idx = self._slice_index_from(c)
            if idx in self.slices:
                if "audio_mute" in kvs:
                    self.slices[idx]["muted"] = (kvs["audio_mute"] == "1")
                    log(f"[slice] {idx} audio_mute={kvs['audio_mute']}")
                if "mode" in kvs:
                    self.slices[idx]["mode"] = kvs["mode"]
                    if self.slices[idx].get("active"):
                        self.slice_mode = kvs["mode"]
                # ANF (Auto-Notch Filter) — radio-side, like TNF. AE sends
                # 'slice set N anf=1'; the RADIO is expected to auto-detect steady
                # tonal interference and notch it. We emulate: with anf on, the sim
                # auto-notches its own tonal channels (birdie/cw carriers) out of
                # BOTH audio and display -- exactly what a real ANF does. Impulse
                # and broadband noise are untouched (ANF can't catch those either).
                if "anf" in kvs:
                    self.slices[idx]["anf"] = (kvs["anf"] == "1")
                    log(f"[slice] {idx} anf={kvs['anf']}")
                # Diversity RX (2-SCU radios only). AE sends 'slice set N
                # diversity=1'. The radio answers by creating a CHILD slice that
                # is RX-only and shares the PARENT's panadapter — the two SCUs
                # feed one combined receiver. Turning it off removes the child.
                if "diversity" in kvs:
                    if kvs["diversity"] == "1":
                        self._diversity_on(conn, idx)
                    else:
                        self._diversity_off(conn, idx)
            self.reply(conn, seq)
        elif c.startswith("slice remove"):
            idx = self._slice_index_from(c)
            gone = self.slices.pop(idx, None)
            if gone:
                # Free its panadapter slot — but ONLY if no other slice is still
                # on that pan. A diversity child deliberately SHARES its parent's
                # pan, so popping unconditionally would delete the pan out from
                # under the parent.
                pid = gone.get("pan")
                if pid is not None and not any(s.get("pan") == pid for s in self.slices.values()):
                    self.pans.pop(pid, None)
            self.reply(conn, seq)
            self.status(conn, f"slice {idx} in_use=0")
            log(f"[slice] remove -> slice {idx}  ({len(self.slices)}/{self.max_slices})")
            if idx == self.active_slice and self.slices:
                self.active_slice = next(iter(self.slices))
            self._sync_active_slice()
            self.emit_meter_status(conn)                   # drop the removed slice's S-meter def
        elif c.startswith("slice set") or c.startswith("slice tune"):
            kvs = parse_kvs(c)
            idx = self._slice_index_from(c)
            sl = self.slices.setdefault(idx, {"freq": self.slice_freq, "mode": self.slice_mode,
                                              "active": False, "pan": self._primary_pan()})
            if "mode" in kvs: sl["mode"] = kvs["mode"]
            for k in ("RF_frequency", "freq"):
                if k in kvs: sl["freq"] = float(kvs[k])
            if c.startswith("slice tune"):                 # "slice tune <idx> <freq>" (positional)
                parts = c.split()
                if len(parts) >= 4:
                    try: sl["freq"] = float(parts[3])
                    except ValueError: pass
            if kvs.get("active") == "1":
                for s in self.slices.values(): s["active"] = False
                sl["active"] = True
                self.active_slice = idx
            if kvs.get("autopan") == "1":                  # Pan Lock on: the pan re-centres on the slice
                pid = sl.get("pan")
                if pid in self.pans: self.pans[pid]["center"] = sl["freq"]
            self.reply(conn, seq)
            self._sync_active_slice()
            self.emit_slice_status(conn, idx)
        elif c.startswith("tnf "):
            self._handle_tnf(conn, seq, c)
        else:
            self.reply(conn, seq)

    def _handle_tnf(self, conn, seq, c):
        """Tracking Notch Filter: 'tnf create freq=<MHz>' (reply = new id),
        'tnf set <id> freq=|width=|depth=|permanent=', 'tnf remove <id>'.
        AE draws/holds the marker only when we echo 'tnf <id> ...' status back."""
        parts = c.split()
        sub = parts[1] if len(parts) > 1 else ""
        if sub == "create":
            kv = parse_kvs(c)
            tid = self._next_tnf_id
            self._next_tnf_id += 1
            self.tnfs[tid] = {
                "freq": float(kv.get("freq", self.slice_freq)),   # MHz
                "width": 100,    # Hz (AE's default; it will 'set width=' after)
                "depth": 1,
                "permanent": 0,
            }
            self.reply(conn, seq, str(tid))       # AE learns the id from the reply body
            self.emit_tnf_status(conn, tid)       # ...then draws the marker from status
            log(f"[tnf] create id={tid} freq={self.tnfs[tid]['freq']:.6f}")
        elif sub == "set" and len(parts) >= 3 and parts[2].isdigit():
            tid = int(parts[2])
            t = self.tnfs.get(tid)
            if t is not None:
                kv = parse_kvs(c)
                if "freq" in kv:      t["freq"] = float(kv["freq"])            # MHz
                if "width" in kv:     t["width"] = _tnf_width_hz(kv["width"])  # AE sends MHz
                if "depth" in kv:     t["depth"] = int(float(kv["depth"]))
                if "permanent" in kv: t["permanent"] = int(kv["permanent"] == "1")
                self.emit_tnf_status(conn, tid)
            self.reply(conn, seq)
        elif sub == "remove" and len(parts) >= 3 and parts[2].isdigit():
            tid = int(parts[2])
            self.tnfs.pop(tid, None)
            self.reply(conn, seq)
            self.status(conn, f"tnf {tid} removed=1")   # AE clears the marker
            log(f"[tnf] remove id={tid}")
        else:
            self.reply(conn, seq)

    def emit_tnf_status(self, conn, tid):
        t = self.tnfs.get(tid)
        if not t or not conn:
            return
        # Status shape AE parses (RadioModel.cpp:6320): width in Hz, freq in MHz.
        self.status(conn, f"tnf {tid} freq={t['freq']:.6f} width={t['width']} "
                          f"depth={t['depth']} permanent={t['permanent']}")

    def emit_pan_status(self, conn, pid=None):
        if pid is None: pid = self._primary_pan()
        pan = self.pans.get(pid)
        if not pan: return
        wid, cfreq = pan["wf_id"], self._pan_center(pid)
        self.status(conn,
            f"display pan 0x{pid:08X} client_handle=0x{self.handle_hex} waterfall=0x{wid:08X} "
            f"center={cfreq:.6f} bandwidth={self.span_mhz:.6f} min_dbm={self.min_dbm:.0f} "
            f"max_dbm={self.max_dbm:.0f} x_pixels={self.bins} y_pixels={self.y_pixels} fps={self.fps} "
            f"ant_list=ANT1 rxant=ANT1")
        self.status(conn,
            f"display waterfall 0x{wid:08X} client_handle=0x{self.handle_hex} panadapter=0x{pid:08X} "
            f"line_duration=100 center={cfreq:.6f} bandwidth={self.span_mhz:.6f} "
            f"auto_black=1 black_level=15 color_gain=50")
        log(f"[->] pan 0x{pid:08X} / wf 0x{wid:08X} status @ {cfreq:.5f}")
        if not self.streaming:
            self.streaming = True
            threading.Thread(target=self.stream_loop, daemon=True).start()

    def _new_pan(self):
        pid, wid = self.pan_id + self.pan_seq, self.wf_id + self.pan_seq
        self.pan_seq += 1
        self.pans[pid] = {"wf_id": wid, "slice": None, "y_pixels": self.y_pixels,
                          "center": self.slice_freq}     # framing centre; fixed unless AE pans (autopan)
        log(f"[pan] create -> 0x{pid:08X}/0x{wid:08X}  ({len(self.pans)} panadapter(s))")
        return pid

    def _primary_pan(self):
        return next(iter(self.pans)) if self.pans else self._new_pan()

    def _pan_center(self, pid):                            # a pan's own framing centre (NOT the slice freq:
        pan = self.pans.get(pid)                           # a slice can tune within a fixed pan, autopan=0)
        if pan and "center" in pan:
            return pan["center"]
        return self.slice_freq

    def _pan_from_kvs(self, kvs):                          # resolve "pan=0x...." to one of our panadapters
        if "pan" in kvs:
            try:
                pid = int(str(kvs["pan"]).lower().replace("0x", ""), 16)
                if pid in self.pans: return pid
            except ValueError: pass
        return self._primary_pan()

    def _next_slice_index(self):
        for k in range(self.max_slices):
            if k not in self.slices: return k
        return None                                        # budget exhausted

    def _slice_index_from(self, c):
        for t in c.split()[2:]:                            # after "slice set"/"slice tune"/"slice remove"
            if t.isdigit(): return int(t)
        return self.active_slice

    def _sync_active_slice(self):
        sl = self.slices.get(self.active_slice)
        if sl:
            self.slice_freq, self.slice_mode = sl["freq"], sl["mode"]
            self.center_mhz = self.slice_freq              # active slice drives the pan/waterfall centre

    def _diversity_on(self, conn, parent_idx):
        """2-SCU diversity: add an RX-only child slice on the PARENT's pan.

        Modelled on FlexLib: the child is a distinct slice tied to its parent by
        DiversitySlicePartner, it is never a transmit slice, and it shares the
        parent's panadapter rather than getting one of its own. That shape is
        what AE keys on (diversity_child / diversity_parent / diversity_index).
        """
        parent = self.slices.get(parent_idx)
        if not parent or parent.get("diversity"):
            return                                          # unknown slice, or already on
        if self.scu < 2:
            log(f"[slice] diversity refused: {self.model} is single-SCU")
            return
        child_idx = self._next_slice_index()
        if child_idx is None:
            log(f"[slice] diversity refused: no free slice ({self.max_slices} in use)")
            return
        pid = parent.get("pan") or self._primary_pan()
        parent.update(diversity=True, div_parent=True, div_child=False, div_index=0)
        self.slices[child_idx] = {
            "freq": parent["freq"], "mode": parent["mode"], "active": False,
            "pan": pid,                                     # SAME pan as the parent
            "tx": False,                                    # RX-only, always
            "diversity": True, "div_parent": False, "div_child": True, "div_index": 1,
        }
        self.emit_slice_status(conn, parent_idx)
        self.emit_slice_status(conn, child_idx)
        log(f"[slice] diversity ON -> parent {parent_idx} + child {child_idx} "
            f"share pan 0x{pid:08X}  ({len(self.slices)}/{self.max_slices})")

    def _diversity_off(self, conn, idx):
        """Turn diversity off from either half of the pair; removes the child."""
        sl = self.slices.get(idx)
        if not sl or not sl.get("diversity"):
            return
        # Normalise to the parent, whichever half AE addressed.
        parent_idx = idx if sl.get("div_parent") else next(
            (i for i, s in self.slices.items() if s.get("div_parent")), None)
        if parent_idx is None:
            return
        child_idx = next((i for i, s in self.slices.items() if s.get("div_child")), None)
        for i in (parent_idx, child_idx):
            if i is not None and i in self.slices:
                self.slices[i].update(diversity=False, div_parent=False,
                                      div_child=False, div_index=-1)
        if child_idx is not None:
            self.slices.pop(child_idx, None)
            self.status(conn, f"slice {child_idx} in_use=0")
            if self.active_slice == child_idx and self.slices:
                self.active_slice = next(iter(self.slices))
                self._sync_active_slice()
        self.emit_slice_status(conn, parent_idx)
        log(f"[slice] diversity OFF -> parent {parent_idx}, child {child_idx} removed")

    def emit_slice_status(self, conn, idx=None):
        if idx is None: idx = self.active_slice
        sl = self.slices.get(idx)
        if not sl: return
        pid = sl.get("pan") or self._primary_pan()
        # tx=1 marks the transmit slice. Without it AE never populates its
        # txByPan map, so anything keyed on "the TX slice" stays inert — the
        # SPLIT/SWAP badges never render at all, for instance.
        # diversity_* describes 2-SCU diversity RX: the child is an RX-only
        # slave that shares its PARENT's panadapter (see handle of
        # "slice set N diversity=").
        # Always send the diversity keys, including the all-zero case. Sending
        # them ONLY while diversity is on leaves the client holding the last
        # non-zero value after teardown — AE keeps diversity_parent=1 on a slice
        # whose diversity=0, which then misleads anything that tests the flags.
        div = (f" diversity={1 if sl.get('diversity') else 0}"
               f" diversity_parent={1 if sl.get('div_parent') else 0}"
               f" diversity_child={1 if sl.get('div_child') else 0}"
               f" diversity_index={sl.get('div_index', -1)}")
        self.status(conn, f"slice {idx} client_handle=0x{self.handle_hex} pan=0x{pid:08X} "
                          f"RF_frequency={sl['freq']:.6f} mode={sl['mode']} in_use=1 "
                          f"tx={1 if sl.get('tx') else 0} "
                          f"active={1 if sl['active'] else 0}{div}")

    def emit_radio_status(self, conn):
        # Capability advert. AE computes MaxSlices/SlicesRemaining from the radio status;
        # without it SlicesRemaining stays 0 and +RX is greyed. slices/max_slices = the model
        # cap; AE stacks one panadapter per receiver, so panadapters tracks the same cap.
        # slices = CURRENT in-use count (AE: free = max_slices - slices); max_slices = the cap.
        # (Sending slices=max told AE "all slices used" -> it refused every new receiver.)
        self.status(conn, f"radio slices={len(self.slices)} max_slices={self.max_slices} "
                          f"panadapters={self.max_slices} "
                          f"lineout_gain=50 lineout_mute=0 headphone_gain=50 headphone_mute=0 "
                          f"callsign=SDRSIM nickname=flex-sim")
        log(f"[->] radio status: slices(in-use)={len(self.slices)} max_slices={self.max_slices}")

    def emit_meter_status(self, conn):
        # Define the radio's meters. NB: meter status is '#'-separated
        # index.key=value tokens after "meter " (NOT space-separated, and NO
        # leading "meter N") -- RadioModel::handleMeterStatus.
        defs = [(FWDPWR_ID, 0, "TX", "FWDPWR", "dBm", 0.0, 60.0),    # TX fwd power (dBm -> W)
                (SWR_ID,    0, "TX", "SWR",    "SWR", 1.0, 10.0)]    # TX SWR
        # Amplifier meters, keyed to the amp object's handle so MeterModel
        # routes them to the PGXL/Amp applet rather than the TGXL/Tuner one.
        # ⚠ num must be NON-ZERO and distinct from any TGXL handle: MeterModel
        # routes AMP meters by `num == tgxlHandle ? TGXL : PGXL`, and the low 16
        # bits of amp_handle (0xA5000000+id) are 0 for radio 0 -- which would
        # collide with an unset/zero TGXL handle. Use the low byte of the handle
        # plus a fixed offset so it is always non-zero and stable per radio.
        amp_num = 0xA5 + self.radio_id
        defs += [(AMP_FWD_ID,  amp_num, "AMP", "FWD",  "dBm",  0.0, 70.0),
                 (AMP_RL_ID,   amp_num, "AMP", "RL",   "dB",   0.0, 60.0),
                 (AMP_TEMP_ID, amp_num, "AMP", "TEMP", "degC", 0.0, 100.0)]
        for k in (sorted(self.slices) or [0]):                       # one S-meter per slice (num = slice index)
            defs.append((SLICE_METER_BASE + k, k, "SLC", "LEVEL", "dBm", -150.0, 20.0))
        for i, num, src, nam, unit, lo, hi in defs:
            toks = "#".join([f"{i}.src={src}", f"{i}.num={num}", f"{i}.nam={nam}",
                             f"{i}.unit={unit}", f"{i}.low={lo}", f"{i}.hi={hi}"])
            self.status(conn, f"meter {toks}")
        log(f"[->] emitted meter defs: {len(self.slices) or 1} S-meter(s) + TX FWDPWR/SWR + AMP FWD/RL/TEMP")

    def tx_envelope(self, t):
        """Envelope (0..1) of what the RADIO is generating on transmit, at time t.

        Stage 1a of TX_AUDIO_DESIGN.md. AE's two-tone test is radio-side: it sends
        'transmit set tune_mode=two_tone' + 'transmit tune 1' and NO audio (measured
        2026-07-22). A real Flex synthesises the tones itself, so the sim does too --
        otherwise the sim sits silent through a two-tone and reports flat power,
        which is simply wrong emulation.

        two_tone : two equal sines, TWO_TONE_SPACING_HZ apart, summed. Their sum
                   BEATS -- envelope |cos(pi*delta*t)| -- which is exactly what makes
                   a two-tone the classic linearity/PEP test: peak power is 2x the
                   single-tone average. Returning the real beat (rather than a flat
                   line) is what lets the power meter swing the way an operator
                   expects to see.
        carrier  : steady dead-key, envelope 1.0 (what 'tune' does without two_tone).

        ⚠ This is a SIMULATION of what the radio would emit. It is not, and must not
        be presented as, a measurement -- see the honesty note in TX_AUDIO_DESIGN.md.
        """
        if self.tune_mode == "two_tone":
            # Beat frequency = the tone spacing; |cos| because power follows the
            # magnitude of the summed envelope, not its sign.
            return abs(math.cos(math.pi * TWO_TONE_SPACING_HZ * t))
        return 1.0

    def tx_drive(self, t):
        """(power_watts, alc_0_to_1) the radio is producing right now.

        Power tracks the envelope SQUARED (envelope is amplitude, power is amplitude^2),
        scaled by the operator's RF-power setting against the model's rated output.
        """
        env = self.tx_envelope(t)
        # tx_power_w already tracks AE's 'transmit set rfpower=' / 'tunepower='.
        watts = float(self.tx_power_w) * (env ** 2)
        # ALC follows the instantaneous envelope; >0.95 is the "you are driving it
        # hard" region an operator watches for.
        return watts, env

    def emit_transmit_status(self):
        # interlock state drives AE's m_radioTransmitting; transmit status drives
        # the MOX/TUNE UI. tx_client_handle = our handle => AE sees us as TX owner.
        if not self.conn:
            return
        on = self.tx_on
        try:
            self.status(self.conn,
                f"interlock state={'TRANSMITTING' if on else 'READY'} source=SW "
                f"tx_client_handle=0x{self.handle_hex}")
            self.status(self.conn,
                f"transmit mox={1 if self.tx_mox else 0} tune={1 if self.tx_tune else 0} "
                f"freq={self.slice_freq:.6f} rfpower={int(self.tx_power_w)} "
                f"tunepower={int(self.tx_power_w)}")
            log(f"[->] TX {'ON' if on else 'off'}  ({self.tx_power_w:.0f} W, SWR {self.tx_swr:.1f})")
        except OSError:
            pass

    def emit_amplifier_status(self, conn):
        # Radio-side amplifier advert (answers 'sub amplifier all'). This is what
        # flips AE's hasAmplifier true, so the AMP applet/button, the status-bar amp
        # indicator, and the peakfwd->S-meter feed all un-gate. Rich telemetry
        # (temp/current/fwd/swr) still arrives over the direct :9008 PGXL link
        # (pgxl_sim); this line establishes presence + OPERATE state.  model must be
        # exactly "PowerGeniusXL" for AE to classify it as a power amplifier.
        try:
            self.status(conn,
                f"amplifier 0x{self.amp_handle:08X} model=PowerGeniusXL serial=PGXL-SIM "
                f"operate={1 if self.amp_operate else 0} ant=ANT1 ip={self.ip} port=9008")
            log(f"[->] amplifier: PowerGeniusXL operate={1 if self.amp_operate else 0}")
        except OSError:
            pass

    def emit_cwx_status(self, **kv):
        if not self.conn:
            return
        body = " ".join(f"{k}={v}" for k, v in kv.items())
        try:
            self.status(self.conn, f"cwx {body}")
        except OSError:
            pass

    def handle_cwx(self, c):
        rest = c[4:].strip()                               # after "cwx "
        if rest.startswith("wpm"):
            try: self.cwx_wpm = max(5, int(rest.split()[1]))
            except (IndexError, ValueError): pass
        elif rest.startswith("qsk"):                        # cwx qsk_enabled 0/1
            self.qsk = rest.split()[-1] == "1"
            log(f"[cwx] QSK {'on (full break-in)' if self.qsk else 'off (normal)'}")
        elif rest.startswith("send"):                      # cwx send "<enc>" <block>
            q1 = c.find('"'); q2 = c.find('"', q1 + 1)
            if q1 >= 0 and q2 > q1:
                text = c[q1 + 1:q2].replace("\x7f", " ")   # 0x7f = AE-encoded space
                self.start_cwx(text)
        elif rest.startswith("clear") or rest.startswith("erase"):
            if self.cwx_active:
                self.cwx_active = False
                if self.tx_mox:
                    self.tx_mox = False
                    self.emit_transmit_status()
                self.emit_cwx_status(queue=0)

    def start_cwx(self, text):
        sched = _morse_schedule(text, self.cwx_wpm)
        if not sched:
            return
        self.cwx_schedule = sched
        self.cwx_char_ends = _morse_char_ends(text, self.cwx_wpm)
        self.cwx_sent = 0
        self.cwx_dur = sum(d for _, d in sched) + CWX_TAIL_S   # tail so AE's local keyer finishes before unkey
        self.cwx_t0 = time.time()
        self.cwx_active = True
        log(f"[cwx] AE sent {text!r} @ {self.cwx_wpm} wpm  ({self.cwx_dur:.1f}s)")

    def _cwx_frame(self, ctx, now):
        # One frame of an AE-initiated CWX transmission. Returns (levels, keyed).
        el = now - self.cwx_t0
        if el >= self.cwx_dur:                             # buffer drained
            self.cwx_active = False
            if self.tx_mox:
                self.tx_mox = False
                self.emit_transmit_status()
            self.emit_cwx_status(queue=0)                  # -> AE releases MOX (#2450)
            log("[cwx] complete")
            return [ctx.floor] * ctx.n, False
        if not self.tx_mox:                                # key the over on the first frame
            self.tx_mox = True
            self.emit_transmit_status()
        keydown = _schedule_keydown(self.cwx_schedule, el)
        ctx.cw_keydown = keydown
        if keydown or not self.qsk:                        # TX burst, or non-QSK over: RX muted
            return None, keydown
        out = [ctx.floor] * ctx.n                           # QSK: RX peeks between elements
        for d in range(-ctx.sig_half, ctx.sig_half + 1):
            b = ctx.center + d
            if 0 <= b < ctx.n:
                out[b] = ctx.sig_level
        return out, keydown

    def _start_audio_thread(self):
        """Start the shared audio_loop if not already running. Used by both the
        remote_audio_rx and dax_rx 'stream create' paths — the loop streams on
        self.audio_stream_id with PCC_IF_NARROW, which AE accepts for either route."""
        if not (self.audio_thread and self.audio_thread.is_alive()):
            self.audio_stop.clear()
            self.audio_thread = threading.Thread(
                target=self.audio_loop, daemon=True, name=f"audio-{self.radio_id}")
            self.audio_thread.start()

    def audio_loop(self):
        """Send remote_audio_rx / dax_rx VITA-49 packets at AUDIO_RATE.

        Three audio sources are supported (self.audio_source):
          AUDIO_SRC_TONE  — deterministic sine at audio_tone_hz (default)
          AUDIO_SRC_NOISE — gaussian band noise
          AUDIO_SRC_WAV   — WAV file at audio_wav_path, resampled + looped
        """
        t0 = time.monotonic()
        while self.run and not self.vita_dest and time.monotonic() - t0 < 12:
            time.sleep(0.05)
        if not self.vita_dest:
            log("[audio] no UDP dest — cannot stream"); return

        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        dest    = self.vita_dest
        src     = self.audio_source

        # Open WAV file if needed (fall back to tone on error)
        wav = None
        if src == AUDIO_SRC_WAV:
            try:
                wav = WavPlayer(self.audio_wav_path)
            except Exception as e:
                log(f"[audio] WAV open failed ({e}), falling back to tone")
                src = AUDIO_SRC_TONE

        log(f"[audio] streaming [{src}] → {dest}  (stream/format chosen live per route)")
        last_route = None   # log only when the route/format actually changes

        seq      = 0
        sample_t = 0   # phase counter for tone source
        t_start  = time.monotonic()
        amp      = 0.1  # −20 dBFS

        wav_path = self.audio_wav_path        # track which file is open, to detect live changes
        try:
            while self.run and not self.audio_stop.is_set():
                # Re-read the source each iteration so the control panel can hot-swap
                # tone/noise/wav (and the WAV path) live without recreating the stream.
                src = self.audio_source
                # Choose stream id + format LIVE: when a dax_rx (RADE/digital) channel is
                # active, stream on the DAX id and FORCE full-BW float32 stereo — the RADE
                # OFDM modem needs full bandwidth, so ignore send_reduced_bw_dax here. If AE
                # arms RADE after remote_audio_rx already started this loop, this switches the
                # running loop onto the dax stream without recreating it.
                dax_on   = self.dax_channel is not None
                reduced  = False if dax_on else self.audio_reduced_bw
                stream_id = self.audio_stream_id
                route = (f"dax_rx ch{self.dax_channel} (full-BW f32)" if dax_on
                         else f"remote_audio_rx ({'mono int16' if reduced else 'stereo f32'})")
                if route != last_route:
                    log(f"[audio] route -> {route}, sid 0x{stream_id:08X}")
                    last_route = route
                if src == AUDIO_SRC_WAV and (wav is None or self.audio_wav_path != wav_path):
                    try:
                        if wav: wav.close()
                        wav = WavPlayer(self.audio_wav_path)
                        wav_path = self.audio_wav_path
                        log(f"[audio] source -> wav: {wav_path}")
                    except Exception as e:
                        log(f"[audio] WAV open failed ({e}), tone"); src = AUDIO_SRC_TONE; wav = None

                any_muted = any(sl.get("muted") for sl in self.slices.values())
                mixer = self.noise_mixer
                if self.paused or any_muted:                       # Stop / slice mute: silence (stream stays
                    mono = [0.0] * AUDIO_FRAMES                     #   alive so Go resumes with continuous timing)
                elif mixer.any_enabled():                          # NOISE BENCH overrides audio_source: stream the
                    # additive voice+CW+noise mix. Keep the mixer's voice WAV in
                    # sync with the requested path (lazy open, hot-swappable).
                    if mixer.voice_wav is None and self.noise_wav_path:
                        try:
                            mixer.voice_wav = WavPlayer(self.noise_wav_path)
                        except Exception as e:
                            log(f"[noise] voice WAV open failed ({e})")
                            self.noise_wav_path = None
                    mono = mixer.mix(AUDIO_FRAMES)
                elif src == AUDIO_SRC_WAV and wav:
                    mono = wav.read(AUDIO_FRAMES)
                elif src == AUDIO_SRC_NOISE:
                    mono = [random.gauss(0, amp) for _ in range(AUDIO_FRAMES)]
                elif src == AUDIO_SRC_TONE:
                    mono = [amp * math.sin(2 * math.pi * self.audio_tone_hz * (sample_t + i) / AUDIO_RATE)
                            for i in range(AUDIO_FRAMES)]
                else:  # AUDIO_SRC_SILENCE (default) — no audio. The DEFAULT so the bench is quiet.
                    mono = [0.0] * AUDIO_FRAMES

                sample_t += AUDIO_FRAMES  # harmless for non-tone sources

                if reduced:
                    samples = mono
                else:
                    samples = []
                    for v in mono: samples.extend([v, v])

                try:
                    s.sendto(audio_packet(stream_id, seq & 0xF, samples, reduced_bw=reduced), dest)
                except OSError as e:
                    log("[audio] send error:", e); break

                seq += 1
                # Absolute deadline — prevents accumulated drift
                deadline = t_start + seq * AUDIO_INTERVAL
                wait = deadline - time.monotonic()
                if wait > 0:
                    time.sleep(wait)
        finally:
            s.close()
            if wav:
                wav.close()
            log("[audio] stopped")

    def stream_loop(self):
        t0 = time.time()
        while self.run and not self.vita_dest and time.time() - t0 < 12:
            time.sleep(0.1)
        if not self.vita_dest:
            log("[stream] no UDP prime in 12s — AE never opened the data path; cannot stream")
            return
        dest = self.vita_dest
        log(f"[stream] streaming '{self.pattern}' to {dest} @ {self.fps} fps (Ctrl-C to stop)")
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        ctx = PatternCtx(self.bins, self.bins // 2, self.min_dbm, self.max_dbm)
        gen = PATTERNS[self.pattern]
        fseq = wseq = mseq = fi = 0
        last_tc = -1                                                # throttle pan/wf to one per row
        start = time.time()
        while self.run and self.streaming:
            now = time.time()
            if self.paused:                                        # Stop: send nothing; AE's pan/wf go dead
                time.sleep(0.1)                                    #       but the radio stays connected. Go resumes.
                continue
            cw_mode = self.cwx_active or self.pattern == "cw"
            period = (1.0 / 50.0) if cw_mode else (1.0 / self.fps)  # 20-ms keying res for 20-wpm CW (60-ms dits)
            tc = int((now - start) * self.fps)                     # waterfall row index (loop-rate independent);
            #                                                        wall-clock so it JUMPS across a TX gap
            #                                                        (real-radio behaviour -> AE history reproj).
            ctx.min_dbm, ctx.max_dbm = self.min_dbm, self.max_dbm   # track AE's display-pan-set
            ctx.floor = self.noise_floor_dbm                        # live (control panel, dBm)
            ctx.sig_level = self.sig_level_dbm
            ctx.noise_color = self.noise_color
            ctx.cw_qsk = self.qsk
            binbw_hz = (self.span_mhz * 1e6) / self.bins            # track AE's live span
            ctx.sig_half = max(1, int(round((self.sig_width_khz * 1000.0) / binbw_hz / 2)))
            ctx.two_tone_half_bins = max(1, int(round((TWO_TONE_SPACING_HZ / binbw_hz) / 2)))  # two-tone ruler spacing in bins
            ctx.noise_cal_full = self.noise_cal_full        # noise_cal: full-span (standard) vs band-limited
            if self.cwx_active:                                    # AE-initiated CWX overrides the pattern
                levels, keyed = self._cwx_frame(ctx, now)
            else:
                gen = PATTERNS.get(self.pattern, gen)              # live pattern switch
                # While the radio is generating a two-tone (stage 1a), SHOW the
                # two tones rather than whatever RX pattern was configured --
                # what you see should match what the radio is emitting.
                if self.tx_on and self.tune_mode == "two_tone":
                    levels = pat_two_tone(ctx, now - start)
                else:
                    levels = gen(ctx, now - start)
                if self.pattern == "tx_blank":                    # faithful TX cycle: the gap == keyed.
                    want_tx = (levels is None)
                elif self.pattern == "cw":                        # Morse keyer: TX for the whole over
                    want_tx = ctx.cw_in_message
                else:
                    want_tx = None
                if want_tx is not None and want_tx != self.tx_mox:  # assert real TX -> repro #2126/#1916
                    self.tx_mox = want_tx
                    self.emit_transmit_status()
                keyed = ctx.cw_keydown if self.pattern == "cw" else self.tx_on
                # NOISE BENCH: when any channel is on, the DISPLAY shows the noise
                # scene (audio path already streams the mix). So the waterfall/pan
                # match what you hear. span_hz here is the audio Hz -> bin mapping;
                # the noise sits within a few kHz of the VFO like real RX audio.
                # Push live notches into the shared mixer -> applied to BOTH audio
                # (mix(), audio thread) and display (spectrum(), below), one source
                # of truth. TWO sources combine:
                #   TNF (manual, user-placed): each TNF's RF freq -> audio Hz offset.
                #   ANF (auto): if the active slice has anf on, auto-notch the mix's
                #     tonal channels (birdie/cw) -- the sim's emulation of a radio's
                #     Auto-Notch Filter finding steady carriers.
                vfo_mhz = self.slice_freq
                notches = [((t["freq"] - vfo_mhz) * 1.0e6, t["width"])
                           for t in self.tnfs.values()]
                active = self.slices.get(self.active_slice)
                if active and active.get("anf"):
                    notches += self.noise_mixer.auto_notch_tones()
                self.noise_mixer.set_notches(notches)
                if self.noise_mixer.any_enabled() and want_tx is None:
                    # Map audio Hz -> bins through AE's LIVE span (binbw_hz), not a
                    # fixed span, so a tone lands at its TRUE on-screen frequency.
                    # display_span_hz = binbw_hz * bins = AE's actual visible span.
                    # (At a wide span a 1 kHz-from-VFO tone correctly sits almost on
                    # the VFO; zoom AE in to a few kHz to see the tones spread out.)
                    display_span_hz = binbw_hz * ctx.n
                    levels = self.noise_mixer.spectrum(
                        ctx.n, self.noise_floor_dbm, display_span_hz, ctx.center)
            if levels is not None and tc != last_tc:               # one pan/wf row per waterfall tick
                pixels = [self.dbm_to_pixel(d) for d in levels]     # generated once; each stacked panadapter
                intens = [self.dbm_to_wf_raw(d) for d in levels]    # shows it, centred on ITS slice (low_hz).
                wf_ab = self.dbm_to_wf_raw(self.noise_floor_dbm)    # auto-black = the CONFIGURED noise floor,
                #   not min(intens): a flat pattern (step/impulse/ramp) has min==max, so min(intens) would
                #   set the black level AT the signal and AE blanks the whole waterfall row. The true floor
                #   keeps flat-high rows bright and still feeds AE's #3586 auto-black path correctly.
                self.last_vfo_dbm = levels[ctx.center]              # active slice (pan centre) -> rack strip
                try:
                    for pid, pan in list(self.pans.items()):        # AE stacks one panadapter per receiver
                        low_hz = (self._pan_center(pid) - self.span_mhz / 2) * 1e6
                        s.sendto(fft_packet(pid, fseq & 0xF, pixels, fi), dest); fseq += 1
                        s.sendto(wf_packet(pan["wf_id"], wseq & 0xF, intens, low_hz, binbw_hz, tc, auto_black=wf_ab), dest); wseq += 1
                    for g in list(self.slices):                     # per-slice S-meter (level at slice = pan centre)
                        s.sendto(meter_packet(self.meter_sid, mseq & 0xF, SLICE_METER_BASE + g, levels[ctx.center]), dest); mseq += 1
                except OSError as e:
                    log("[stream] send error:", e); break
                fi += 1
                last_tc = tc
            # TX meters every frame: real power/SWR while keyed, ~0 W / 1.0 SWR when not,
            # so AE's meter decays back to zero on de-key. CW/CWX key power per element.
            # Power follows the radio-generated envelope (stage 1a): a two-tone
            # BEATS, so the meter swings between ~0 and full instead of sitting
            # flat -- which is the whole point of watching a two-tone.
            pw = self.tx_drive(now)[0] if keyed else 0.0
            sw = self.tx_swr if keyed else 1.0
            fwd_dbm = 10.0 * math.log10(max(pw, 1e-6)) + 30.0
            # Amp heatsink thermal lag: warms toward a drive-dependent target
            # while keyed in OPERATE, coasts back to ambient otherwise. Updated
            # OUTSIDE the send try/except so a transient socket error cannot
            # freeze the model mid-climb.
            tgt = 38.0 + (26.0 * min(1.0, pw / 1200.0)
                          if (keyed and self.amp_operate) else 0.0)
            self.amp_temp_c += (tgt - self.amp_temp_c) * 0.02
            try:
                s.sendto(meter_packet(self.meter_sid, mseq & 0xF, FWDPWR_ID, fwd_dbm), dest); mseq += 1
                s.sendto(meter_packet(self.meter_sid, mseq & 0xF, SWR_ID, sw), dest); mseq += 1
                # Amplifier meters. In OPERATE the amp is amplifying, so it
                # reports GAIN_DB more forward power than the exciter; in
                # STANDBY it is a through-path and reports the exciter level.
                # RL is return loss in dB (>0), derived from the same SWR --
                # AE converts it back, so the two agree by construction.
                # Heatsink thermal lag: warms toward a drive-dependent target
                # while keyed, coasts back to ambient when not. Gives the
                # applet's temp gauge something real to track.
                amp_dbm = fwd_dbm + (AMP_GAIN_DB if self.amp_operate else 0.0)
                rho = max(1e-6, (sw - 1.0) / (sw + 1.0))
                amp_rl = -20.0 * math.log10(rho)      # 1.0 SWR -> ~120 dB, 3:1 -> 6 dB
                s.sendto(meter_packet(self.meter_sid, mseq & 0xF, AMP_FWD_ID, amp_dbm), dest); mseq += 1
                s.sendto(meter_packet(self.meter_sid, mseq & 0xF, AMP_RL_ID, min(60.0, amp_rl)), dest); mseq += 1
                s.sendto(meter_packet(self.meter_sid, mseq & 0xF, AMP_TEMP_ID,
                                      self.amp_temp_c, scale=64.0), dest); mseq += 1
            except OSError:
                pass
            dt = period - (time.time() - now)
            if dt > 0: time.sleep(dt)
        log("[stream] stopped")


# ---- test-bench helpers: list fixtures + reports, run the ruler suite ----------
# These back the panel's "Audio source" fixture dropdown and the "Tests & reports"
# section. All read-only / sandboxed to the known dirs; the runner is a subprocess
# (run_tests.py spins up its OWN mock-AE loopback, so it never disturbs the live sim).

# Available ruler names — kept in sync with run_tests.py RULERS (the picker offers these).
RULER_NAMES = ["two_tone", "noise_cal", "cal_tones", "carrier", "comb"]

# In-flight test runs: id -> {proc, stamp, done, passed, report}
_test_runs = {}
_test_lock = threading.Lock()


def list_fixtures():
    """WAV fixtures available to the audio-source dropdown (name + size + duration)."""
    out = []
    for p in sorted(glob.glob(os.path.join(FIXTURES_DIR, "*.wav"))):
        name = os.path.basename(p)
        info = {"name": name, "path": p, "size": os.path.getsize(p)}
        try:
            with wave.open(p, "rb") as w:
                info["dur"] = round(w.getnframes() / w.getframerate(), 2)
        except Exception:
            info["dur"] = None        # float WAVs (RADE Tap-E) — wave can't read fmt 3; that's fine
        out.append(info)
    return out


def list_reports():
    """Ruler reports on disk, newest first, with a pass/fail summary from the JSON sidecar."""
    out = []
    for p in sorted(glob.glob(os.path.join(REPORTS_DIR, "ruler-report_*.html")),
                    key=os.path.getmtime, reverse=True):
        name = os.path.basename(p)
        rec = {"name": name, "mtime": os.path.getmtime(p), "passed": None, "total": None}
        jp = p[:-5] + ".json"
        try:
            with open(jp, encoding="utf-8") as fp:
                d = json.load(fp)
            res = d.get("results", [])
            rec["total"] = len(res)
            rec["passed"] = sum(1 for r in res if r.get("passed"))
            rec["stamp"] = d.get("meta", {}).get("timestamp")
        except Exception:
            pass
        out.append(rec)
    return out


def start_test_run(only=None):
    """Spawn run_tests.py as a subprocess; return a run-id to poll via /run-status."""
    stamp = time.strftime("%Y-%m-%d_%H%M%S")
    rid = f"run-{stamp}"
    cmd = [sys.executable, os.path.join(HERE, "run_tests.py"),
           "--out", REPORTS_DIR, "--stamp", stamp]
    if only:
        cmd += ["--only", ",".join(only)]
    proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    with _test_lock:
        _test_runs[rid] = {"proc": proc, "stamp": stamp, "done": False,
                           "report": f"ruler-report_{stamp}.html"}
    log(f"[test] started ruler run {rid}" + (f" (only {','.join(only)})" if only else " (all)"))
    return rid


def test_run_status(rid):
    """Poll a spawned run: running / done + the report filename and pass count."""
    with _test_lock:
        r = _test_runs.get(rid)
    if not r:
        return {"id": rid, "found": False}
    rc = r["proc"].poll()
    if rc is None:
        return {"id": rid, "found": True, "done": False}
    # finished — read pass/fail from the report json if present
    rep = {"id": rid, "found": True, "done": True, "rc": rc, "report": r["report"]}
    jp = os.path.join(REPORTS_DIR, r["report"][:-5] + ".json")
    try:
        with open(jp, encoding="utf-8") as fp:
            res = json.load(fp).get("results", [])
        rep["passed"] = sum(1 for x in res if x.get("passed"))
        rep["total"] = len(res)
    except Exception:
        pass
    return rep


# ---- LIVE test mode: drive the REAL connected AE through each ruler, visibly --------
# Unlike start_test_run (hidden mock-AE subprocess), this sets the live sim to each
# ruler's pattern so AE's panadapter/waterfall actually SHOW it, holds ~5 s so it's
# unmistakable, and runs the SAME numeric check (run_tests.check_*) in parallel so the
# measured-vs-expected lines stream alongside what Nigel sees on the radio. Auto-advances
# (pausable). The numbers come from run_tests so they never diverge from the standalone runner.

# Per-ruler order + the level/floor to set + the "what to look for on AE" guidance.
LIVE_RULERS = [
    ("carrier",   {"level": -73, "floor": -120},
     "One sharp peak at the VFO centre. AE's S-meter should read ~-73 dBm (S9)."),
    ("two_tone",  {"level": -73, "floor": -120},
     "TWO equal peaks symmetric about the VFO, ~2 kHz apart, both at -73 dBm. Nothing between/outside."),
    ("cal_tones", {"level": -73, "floor": -130},
     "Four tones across the span at -100/-80/-60/-40 dBm. Check AE's dB scale reads each step."),
    ("noise_cal", {"level": -73, "floor": -130},
     "A flat noise shelf sitting AT -73 dBm (±1.5 dB ripple). Put AE's RX filter on it to trace its shape."),
    ("comb",      {"level": -73, "floor": -120},
     "Eight evenly-spaced tones across the span, all well above the floor."),
]

_live = {"id": None, "thread": None, "steps": [], "i": -1, "paused": False,
         "running": False, "dwell": 5.0, "abort": False}
_live_lock = threading.Lock()


def _live_run(radio, dwell):
    import run_tests as rt
    prev_pattern = radio.pattern
    prev_level, prev_floor = radio.sig_level_dbm, radio.noise_floor_dbm
    try:
        for idx, (pattern, cfg, look) in enumerate(LIVE_RULERS):
            with _live_lock:
                if _live["abort"]:
                    break
                _live["i"] = idx
            # 1. Drive the REAL AE: set pattern + level/floor so the expected reading holds.
            radio.sig_level_dbm = float(cfg["level"])
            radio.noise_floor_dbm = float(cfg["floor"])
            radio.pattern = pattern
            radio.paused = False                      # make sure output is live so AE shows it
            log(f"[live-test] {idx+1}/{len(LIVE_RULERS)} -> {pattern} (watch AE)")
            with _live_lock:
                _live["steps"][idx]["state"] = "showing"
            # 2. Hold the dwell so it's visibly on screen (pausable).
            t_end = time.monotonic() + dwell
            while time.monotonic() < t_end:
                with _live_lock:
                    if _live["abort"]:
                        break
                    paused = _live["paused"]
                    _live["steps"][idx]["remaining"] = max(0.0, round(t_end - time.monotonic(), 1))
                if paused:
                    t_end = time.monotonic() + 0.2     # freeze the countdown while paused
                time.sleep(0.1)
            # 3. Run the SAME numeric check (mock-loopback) and stream the lines.
            try:
                bins, mdbm = rt.capture_frame(pattern)
                check = next(c for p, c in rt.RULERS if p == pattern)
                r = check(bins, mdbm)
                with _live_lock:
                    _live["steps"][idx]["measurements"] = r.get("measurements", [])
                    _live["steps"][idx]["passed"] = r.get("passed")
                    _live["steps"][idx]["state"] = "done"
            except Exception as e:
                with _live_lock:
                    _live["steps"][idx]["state"] = "done"
                    _live["steps"][idx]["passed"] = False
                    _live["steps"][idx]["error"] = str(e)
    finally:
        radio.pattern = prev_pattern
        radio.sig_level_dbm, radio.noise_floor_dbm = prev_level, prev_floor
        with _live_lock:
            _live["running"] = False
            _live["i"] = len(LIVE_RULERS)
        log("[live-test] finished — restored prior pattern")


def live_test_start(radio, dwell=5.0):
    with _live_lock:
        if _live["running"]:
            return {"id": _live["id"], "already": True}
        rid = f"live-{int(time.monotonic()*1000)}"
        _live.update(id=rid, paused=False, running=True, abort=False, dwell=float(dwell), i=-1,
                     steps=[{"pattern": p, "look": look, "state": "pending",
                             "measurements": [], "passed": None, "remaining": None}
                            for p, _, look in LIVE_RULERS])
        th = threading.Thread(target=_live_run, args=(radio, float(dwell)), daemon=True)
        _live["thread"] = th
        th.start()
        return {"id": rid}


def live_test_status():
    with _live_lock:
        return {"id": _live["id"], "running": _live["running"], "paused": _live["paused"],
                "i": _live["i"], "dwell": _live["dwell"],
                "n": len(LIVE_RULERS), "steps": _live["steps"]}


def live_test_set(paused=None, abort=False):
    with _live_lock:
        if paused is not None:
            _live["paused"] = bool(paused)
        if abort:
            _live["abort"] = True
    return {"ok": True}


# ---- live control panel (web UI served by the sim; drive from the host browser) ----
CONTROL_HTML = """<!DOCTYPE html><html><head><meta charset=utf-8><title>flex-sim control</title>
<style>body{{font-family:sans-serif;background:#111;color:#ddd;padding:18px;max-width:440px}}
h2{{color:#5cf}} label{{display:block;margin:16px 0 4px}} select,input[type=range]{{width:380px}}
.v{{color:#5cf;font-weight:bold}}</style></head><body>
<h2 style="display:flex;align-items:baseline;gap:12px">flex-sim &mdash; live control
<a href="/qa/view" style="font-size:13px;font-weight:normal;color:#5cf;text-decoration:none;border:1px solid #243;border-radius:6px;padding:3px 9px">QA results &rsaquo;</a></h2>
<div id=statusbar style="display:flex;align-items:center;gap:12px;background:#15202b;border:1px solid #243;border-radius:6px;padding:10px 12px;margin:0 0 14px">
  <button id=gobtn onclick="togglePause()" style="font-size:15px;font-weight:bold;padding:7px 16px;border:none;border-radius:5px;cursor:pointer;background:#c33;color:#fff">&#9632; Stop</button>
  <div style="line-height:1.4">
    <div><span id=dot style="display:inline-block;width:10px;height:10px;border-radius:50%;background:#888;margin-right:7px"></span><span id=conn style="font-weight:bold">…</span></div>
    <div id=streamstat style="font-size:12px;color:#9ab;margin-top:2px">&nbsp;</div>
  </div>
</div>
<label>Pattern</label><select id=pat onchange=upd()>{options}</select>
<div id=hint style="background:#15202b;border-left:3px solid #5cf;padding:8px 11px;margin:8px 0;font-size:13px;color:#bcd;line-height:1.45"></div>
<label>Noise floor: <span class=v id=floorv>-120</span> dBm <span class=v id=floors>(S1)</span></label>
<input type=range id=floor min=-130 max=-20 step=1 value=-120 oninput=upd()>
<label>Signal level: <span class=v id=levelv>-73</span> dBm <span class=v id=levels>(S9)</span></label>
<input type=range id=level min=-130 max=-10 step=1 value=-73 oninput=upd()>
<label>Signal width: <span class=v id=widthv>10</span> kHz</label>
<input type=range id=width min=1 max=250 step=1 value=10 oninput=upd()>
<label>Noise color (white&rarr;pink): <span class=v id=tiltv>0</span> dB tilt</label>
<input type=range id=tilt min=0 max=30 step=1 value=0 oninput=upd()>
<hr style="border-color:#333;margin:20px 0">
<label><input type=checkbox id=tx onchange=upd()> <b>TX (key) &mdash; forward-power meter</b></label>
<label>TX power: <span class=v id=txpwv>100</span> W</label>
<input type=range id=txpw min=0 max=100 step=1 value=100 oninput=upd()>
<label>SWR: <span class=v id=swrv>1.2</span> : 1</label>
<input type=range id=swr min=10 max=50 step=1 value=12 oninput=upd()>
<div style="margin-top:14px">
<button onclick="sendcw(0)">&#9654; CW normal</button>
<button onclick="sendcw(1)">&#9654; CW full break-in</button>
<button onclick="setp('carrier')">&#9632; back to carrier</button>
</div>
<div style="font-size:12px;color:#888;margin-top:6px">sim-keyed CQ CQ CQ TST @ 20 wpm. Authentic path: send from AE's CWX &mdash; the sim follows AE's QSK. (Use the Stop button up top to pause all output.)</div>
<hr style="border-color:#333;margin:20px 0">
<h3 style="color:#5cf;margin:0 0 10px">Audio source</h3>
<div style="display:flex;gap:16px;align-items:center;flex-wrap:wrap">
<label style="margin:0"><input type=radio name=asrc value=silence id=r_silence checked onchange=updAudio()> <b>Silence</b> <span style="color:#888;font-size:12px">(default — no audio)</span></label>
<label style="margin:0"><input type=radio name=asrc value=tone id=r_tone onchange=updAudio()> Tone <span style="color:#888;font-size:12px">(440 Hz sine)</span></label>
<label style="margin:0"><input type=radio name=asrc value=noise id=r_noise onchange=updAudio()> Noise <span style="color:#888;font-size:12px">(gaussian)</span></label>
<label style="margin:0"><input type=radio name=asrc value=wav id=r_wav onchange=updAudio()> WAV file</label>
</div>
<div id=wav_row style="margin-top:10px;display:none">
<label style="margin:0 0 4px;display:block;font-size:13px;color:#aaa">Fixture (from fixtures/)</label>
<select id=fix onchange="pickFixture()"
  style="width:386px;background:#1a2332;color:#ddd;border:1px solid #334;padding:4px 6px">
  <option value="">— choose a fixture —</option>
</select>
<label style="margin:10px 0 4px;display:block;font-size:13px;color:#aaa">…or custom path (server filesystem)</label>
<input type=text id=wav_path placeholder="/path/to/capture.wav"
  style="width:374px;background:#1a2332;color:#ddd;border:1px solid #334;padding:4px 6px;font-family:monospace"
  oninput=updAudio()>
<div style="font-size:12px;color:#666;margin-top:4px">Any sample rate / channel count. Resampled to 24&nbsp;kHz mono and looped.
<br>Takes effect on next audio stream start (reconnect AE if already connected).</div>
</div>
<hr style="border-color:#333;margin:20px 0">
<h3 style="color:#5cf;margin:0 0 4px">HF Noise Bench <span style="color:#888;font-size:12px;font-weight:normal">— additive audio for testing AE's noise reduction</span></h3>
<div style="font-size:12px;color:#888;margin-bottom:10px">Each channel is generated live (never a recording) and <b>sums into the audio AE hears</b> on its RX audio path — the same buffer RN2/NR2/NR4/BNR clean up. Toggle any combination; set levels in dBFS. When any channel is on it overrides the Audio-source picker above.</div>
<div style="margin-bottom:10px">
  <span style="font-size:13px;color:#aaa">Scene presets:</span>
  <span id=presets></span>
  <button onclick="fetch('/set?noise_reset=1');setTimeout(pollNoise,120)" style="margin-left:6px;background:#511;color:#fcc;border:none;border-radius:4px;padding:4px 9px;cursor:pointer">All off</button>
</div>
<div id=noisebench style="display:grid;grid-template-columns:auto 1fr auto;gap:6px 12px;align-items:center;max-width:560px"></div>
<div id=noise_voicewav style="margin-top:8px;font-size:12px;color:#888"></div>
<hr style="border-color:#333;margin:20px 0">
<details style="margin-top:4px">
<summary style="color:#5cf;font-size:18px;font-weight:bold;cursor:pointer">Tests &amp; reports</summary>
<div style="margin-top:12px">
<div style="font-size:13px;color:#aaa;margin-bottom:6px">Calibration rulers (run via run_tests.py — uses its own mock-AE loopback, doesn't touch the live sim)</div>
<div id=rulers style="display:flex;gap:14px;flex-wrap:wrap;margin-bottom:10px"></div>
<button id=runbtn onclick="runTests()">&#9654; Run selected rulers</button>
<span id=runstat style="margin-left:10px;font-size:13px;color:#5cf"></span>
<div style="font-size:12px;color:#888;margin:6px 0 0">(fast: hidden mock-AE loopback — proves the numbers, but your radio shows nothing)</div>
<div style="margin-top:16px;padding-top:14px;border-top:1px solid #243">
<div style="font-size:13px;color:#aaa;margin-bottom:6px">Live test — drives YOUR real AE, ~5 s per ruler, so you watch each signal appear</div>
<button id=livebtn onclick="liveStart()" style="background:#1a5;color:#fff;border:none;border-radius:5px;padding:7px 14px;font-weight:bold;cursor:pointer">&#9654; Live test on real AE</button>
<div id=livebox style="display:none;margin-top:12px;background:#0d1620;border:1px solid #243;border-radius:6px;padding:12px"></div>
</div>
<div style="font-size:13px;color:#aaa;margin:16px 0 6px">Recent reports</div>
<div id=reports style="font-size:13px"></div>
</div>
</details>
<script>
function sLabel(d){{var o=d+73;if(o>0)return '(S9+'+Math.round(o)+'dB)';return '(S'+Math.max(0,Math.round(9+o/6))+')';}}
var qskv=0;
var HINTS={{
carrier:"Steady carrier on the VFO. S-meter should read the signal level (dBm / S-unit); panadapter peak + solid waterfall stripe at centre.",
cw:"20-wpm Morse 'CQ CQ CQ TST'. Power meter keys the rhythm; waterfall flickers during keying — watch it recover cleanly after the over (#1916).",
noise_floor:"Flat, featureless noise floor (baseline). With AE auto-black on, the floor should sit even and dark (#3586).",
ramp:"Whole-span level swept min→max. Watch auto-black hold the floor even as level rises (#3586) and the colour map track it.",
test_card:"17 carriers S1→S9 (1/2-S steps) on a steppable noise floor. Raise the floor: weak carriers submerge and AE's black point tracks it — auto-black doing its job (<i>#3586</i>).",
cal_tones:"Tones at -100/-80/-60/-40 dBm. Verify the panadapter dB scale reads each one correctly (calibration card).",
swept_carrier:"Single tone sweeping the span. Check frequency mapping / waterfall tile decode (#3457).",
comb:"Evenly-spaced tones across the span. Dynamic range / many simultaneous bins.",
two_tone:"Two EQUAL tones {tt_khz} kHz apart, symmetric about the VFO, each at the signal level (dBm / S-unit shown). <b>Golden ruler:</b> both peaks read the SAME level; the band between &amp; outside stays at the noise floor — no intermod products (a pure sim path is perfectly linear, so any spur AE shows is AE's, not the signal's).",
step:"Whole-span square wave in time. Temporal response / waterfall scroll timing.",
impulse:"Brief full-span flash each period. Transient response / paced fallback (#3182).",
tx_blank:"Periodic real TX gap (keys interlock + power). Watch the waterfall during and after TX — #2126 (flicker) / #1916 (disappears).",
staircase:"Centre carrier stepping floor→max in 10 even steps. Reads AE's colormap / black-threshold as a ladder.",
noise:"Random noise band, white→pink tilt. Floor texture and colour mapping.",
ssb:"An <b>SSB voice lookalike</b> — a ~2.4 kHz band offset above the VFO (USB-style) with a syllabic swell, word-gap pauses and drifting formants, so the waterfall <i>reads</i> like someone talking on HF. For looks/demo, not a ruler. <b>Tip:</b> narrow AE's span (zoom to a few kHz) to see the speech detail — at a wide span it's only ~20 bins. Signal-level slider sets how loud it sits.",
noise_cal:"Flat noise at the EXACT signal level (small ±1.5 dB ripple) — a <b>filter-test bed</b>. Put AE's RX filter on it: the noise output traces the filter's response (passband sits at level, stopband drops to floor, edges show the roll-off — the whole curve at once). Narrow the width below the span for band-limited noise that the filter chews at the edges. noise_color tilts white→pink."
}};
function setp(p){{pat.value=p;upd();}}
function sendcw(q){{qskv=q;pat.value='cw';upd();}}
function updAudio(){{
  var src=document.querySelector('input[name=asrc]:checked').value;
  document.getElementById('wav_row').style.display=(src==='wav'?'block':'none');
  var p=new URLSearchParams({{audio_src:src}});
  if(src==='wav')p.append('wav_path',document.getElementById('wav_path').value);
  fetch('/set?'+p.toString());
}}
function upd(){{var h=HINTS[pat.value];if(h)hint.innerHTML=h;
floorv.textContent=floor.value;levelv.textContent=level.value;
floors.textContent=sLabel(+floor.value);levels.textContent=sLabel(+level.value);
widthv.textContent=width.value;tiltv.textContent=tilt.value;
txpwv.textContent=txpw.value;var sv=(swr.value/10).toFixed(1);swrv.textContent=sv;
fetch('/set?pattern='+pat.value+'&floor='+floor.value+'&level='+level.value+'&width='+width.value+'&tilt='+tilt.value+'&tx='+(tx.checked?1:0)+'&txpw='+txpw.value+'&swr='+sv+'&qsk='+qskv);}}

// ---- fixtures dropdown ----
function pickFixture(){{var f=document.getElementById('fix').value;if(f){{document.getElementById('wav_path').value=f;updAudio();}}}}
function loadFixtures(){{fetch('/fixtures').then(r=>r.json()).then(list=>{{
  var s=document.getElementById('fix');
  list.forEach(f=>{{var o=document.createElement('option');o.value=f.path;
    var d=f.dur?(' &middot; '+f.dur+'s'):'';o.textContent=f.name+' ('+(f.size/1024|0)+' KB'+d+')';s.appendChild(o);}});
}}).catch(()=>{{}});}}

// ---- tests & reports ----
var RULERS=['two_tone','noise_cal','cal_tones','carrier','comb'];
function loadRulers(){{var d=document.getElementById('rulers');
  RULERS.forEach(r=>{{var l=document.createElement('label');l.style.margin='0';l.style.fontSize='13px';
    l.innerHTML='<input type=checkbox value="'+r+'" checked> '+r;d.appendChild(l);}});}}
function selectedRulers(){{return Array.from(document.querySelectorAll('#rulers input:checked')).map(c=>c.value);}}
function runTests(){{
  var only=selectedRulers();
  if(!only.length){{document.getElementById('runstat').textContent='select at least one ruler';return;}}
  document.getElementById('runbtn').disabled=true;
  document.getElementById('runstat').textContent='running '+only.length+' ruler(s)…';
  fetch('/run-tests?only='+only.join(',')).then(r=>r.json()).then(j=>poll(j.id));
}}
function poll(id){{fetch('/run-status?id='+encodeURIComponent(id)).then(r=>r.json()).then(s=>{{
  if(!s.done){{setTimeout(()=>poll(id),700);return;}}
  document.getElementById('runbtn').disabled=false;
  var msg=(s.passed!=null)?(s.passed+'/'+s.total+' passed'):'done (rc '+s.rc+')';
  document.getElementById('runstat').innerHTML=msg+' &middot; <a style="color:#5cf" target=_blank href="/reports/'+s.report+'">open report</a>';
  loadReports();
}});}}
function loadReports(){{fetch('/reports').then(r=>r.json()).then(list=>{{
  var d=document.getElementById('reports');d.innerHTML='';
  if(!list.length){{d.innerHTML='<span style="color:#666">no reports yet</span>';return;}}
  list.slice(0,8).forEach(r=>{{
    var ok=(r.passed!=null&&r.passed===r.total);
    var dot=(r.passed==null)?'#888':(ok?'#3c6':'#e55');
    var sum=(r.passed!=null)?(r.passed+'/'+r.total):'';
    var when=r.stamp||r.name;
    d.innerHTML+='<div style="margin:3px 0"><span style="display:inline-block;width:9px;height:9px;border-radius:50%;background:'+dot+';margin-right:7px"></span>'
      +'<a style="color:#5cf" target=_blank href="/reports/'+r.name+'">'+when+'</a> <span style="color:#888">'+sum+'</span></div>';
  }});
}}).catch(()=>{{}});}}

// ---- Stop/Go + live status ----
var paused=false;
function togglePause(){{
  paused=!paused;
  fetch('/set?paused='+(paused?1:0));
  renderGo();pollStatus();
}}
function renderGo(){{
  var b=document.getElementById('gobtn');
  if(paused){{b.innerHTML='&#9654; Go';b.style.background='#3a7';}}
  else{{b.innerHTML='&#9632; Stop';b.style.background='#c33';}}
}}
function pollStatus(){{fetch('/status').then(r=>r.json()).then(s=>{{
  paused=s.paused;renderGo();
  var dot=document.getElementById('dot'),conn=document.getElementById('conn'),ss=document.getElementById('streamstat');
  if(!s.connected){{dot.style.background='#888';conn.textContent='waiting for AetherSDR…';ss.innerHTML='&nbsp;';}}
  else{{
    dot.style.background=s.paused?'#fa3':'#3c6';
    conn.textContent='connected to AE'+(s.peer?(' ('+s.peer+')'):'');
    var what=s.paused?'<b style="color:#fa3">output STOPPED</b> — spectrum &amp; audio silent'
                     :('streaming <b>'+s.pattern+'</b>'+(s.tx?' &middot; TX':'')+' &middot; '+s.meter_dbm+' dBm');
    ss.innerHTML=what;
  }}
}}).catch(()=>{{
  document.getElementById('dot').style.background='#888';
  document.getElementById('conn').textContent='sim unreachable';
}});}}

// ---- live test (drive the real AE) ----
var liveTimer=null, livePaused=false;
function liveStart(){{
  document.getElementById('livebox').style.display='block';
  document.getElementById('livebtn').disabled=true;
  livePaused=false;
  fetch('/live-test/start?dwell=5').then(()=>{{
    if(liveTimer)clearInterval(liveTimer);
    liveTimer=setInterval(livePoll,400);livePoll();
  }});
}}
function livePause(){{livePaused=!livePaused;fetch('/live-test/pause?paused='+(livePaused?1:0));}}
function liveStop(){{fetch('/live-test/stop');}}
function livePoll(){{fetch('/live-test/status').then(r=>r.json()).then(s=>{{
  var box=document.getElementById('livebox');
  var h='';
  s.steps.forEach((st,i)=>{{
    var cur=(i===s.i && s.running);
    var icon=(st.state==='done')?(st.passed?'<span style="color:#3c6">&#10004;</span>':'<span style="color:#e55">&#10007;</span>')
            :(st.state==='showing')?'<span style="color:#fa3">&#9654;</span>':'<span style="color:#666">&middot;</span>';
    h+='<div style="margin:6px 0;padding:6px 8px;border-radius:5px;'+(cur?'background:#15202b;border:1px solid #fa3':'')+'">';
    h+='<div style="font-weight:bold">'+icon+' '+(i+1)+'/'+s.n+'  '+st.pattern
       +(cur&&st.state==='showing'?(' <span style="color:#fa3;font-size:12px">— SHOWING on AE ('+(st.remaining!=null?st.remaining:'')+'s)</span>'):'')+'</div>';
    if(cur||st.state==='done')h+='<div style="font-size:12px;color:#9ab;margin:3px 0 0">'+st.look+'</div>';
    (st.measurements||[]).forEach(m=>{{
      var ok=m.ok?'<span style="color:#3c6">&#10004;</span>':'<span style="color:#e55">&#10007;</span>';
      h+='<div style="font-size:12px;font-family:monospace;margin:2px 0 0 14px">'+ok+' '+m.name+': '
         +'<b>'+m.measured+'</b> '+(m.unit||'')+' (exp '+m.expected+')</div>';
    }});
    if(st.error)h+='<div style="font-size:12px;color:#e55;margin-left:14px">'+st.error+'</div>';
    h+='</div>';
  }});
  if(s.running){{
    h+='<div style="margin-top:10px"><button onclick="livePause()" style="background:'+(livePaused?'#3a7':'#a73')+';color:#fff;border:none;border-radius:4px;padding:5px 12px;cursor:pointer">'
      +(livePaused?'&#9654; Resume':'&#9208; Pause')+'</button> '
      +'<button onclick="liveStop()" style="background:#555;color:#fff;border:none;border-radius:4px;padding:5px 12px;cursor:pointer;margin-left:6px">&#9632; Stop</button></div>';
  }} else {{
    if(liveTimer){{clearInterval(liveTimer);liveTimer=null;}}
    document.getElementById('livebtn').disabled=false;
    var np=s.steps.filter(x=>x.passed).length;
    h+='<div style="margin-top:10px;font-weight:bold;color:#5cf">Done — '+np+'/'+s.n+' rulers passed (numbers). You watched each on AE.</div>';
    loadReports();
  }}
  box.innerHTML=h;
}}).catch(()=>{{}});}}

// ---- HF noise bench ----
// [name, label, knob-or-null, knob-label, min, max, step]
var NOISE_CH=[
 ['voice','Voice (WAV)',null,'',0,0,0],
 ['cw','CW tone','hz','Hz',300,1200,10],
 ['white','White / AWGN',null,'',0,0,0],
 ['pink','Pink / band hiss',null,'',0,0,0],
 ['qrn','QRN impulse',',rate','imp/s',1,60,1],
 ['powerline','Power-line buzz','freq','Hz',50,60,10],
 ['crashes','Static crashes','rate','cr/s',0.1,3,0.1],
 ['birdie','Birdie carrier','hz','Hz',300,3000,10],
 ['hash','SMPS hash','prf','Hz',30,400,10],
 ['woodpecker','Woodpecker','prf','pps',2,50,1]];
var NOISE_PRESETS=['quiet-20m','night-40m','storm','noisy-qth','birdie-hell','voice-in-noise','cw-in-noise'];
function nset(k,v){{fetch('/set?noise_'+k+'='+v).then(()=>setTimeout(pollNoise,80));}}
// Build via DOM APIs + event LISTENERS (no inline on*="..." handlers) so there
// are no nested single-quotes to survive the Python f-string -- that mangling
// (\' -> '') was silently killing this whole function and leaving the grid empty.
function nEl(tag,attrs){{var e=document.createElement(tag);for(var k in attrs)e.setAttribute(k,attrs[k]);return e;}}
function buildNoise(){{
  var g=document.getElementById('noisebench');g.innerHTML='';
  NOISE_CH.forEach(function(c){{
    var name=c[0],lbl=c[1],knob=c[2];
    // column 1: toggle + label
    var lab=nEl('label',{{style:'margin:0;font-size:13px'}});
    var cb=nEl('input',{{type:'checkbox',id:'nc_'+name}});
    cb.addEventListener('change',function(){{nset(name,this.checked?1:0);}});
    lab.appendChild(cb);lab.appendChild(document.createTextNode(' '+lbl));
    // column 2: level slider + readout
    var d2=nEl('div',{{style:'display:flex;align-items:center;gap:6px'}});
    var ls=nEl('input',{{type:'range',id:'nl_'+name,min:-60,max:0,step:1,style:'width:100%'}});
    var lv=nEl('span',{{id:'nlv_'+name,style:'font-family:monospace;color:#5cf;min-width:34px;display:inline-block'}});
    lv.textContent='-';
    ls.addEventListener('input',function(){{lv.textContent=this.value;nset(name+'_level',this.value);}});
    d2.appendChild(ls);d2.appendChild(lv);d2.appendChild(document.createTextNode(' dB'));
    // column 3: optional knob
    var d3=nEl('div',{{style:'white-space:nowrap'}});
    if(knob){{var kk=knob.replace(',',''),mn=c[4],mx=c[5],st=c[6];
      var kl=nEl('span',{{style:'font-size:12px;color:#888'}});kl.textContent=c[3]+' ';
      var ks=nEl('input',{{type:'range',id:'nk_'+name,min:mn,max:mx,step:st,style:'width:70px'}});
      var kv=nEl('span',{{id:'nkv_'+name,style:'font-family:monospace;color:#8cf'}});kv.textContent='-';
      ks.addEventListener('input',function(){{kv.textContent=this.value;nset(name+'_'+kk,this.value);}});
      d3.appendChild(kl);d3.appendChild(ks);d3.appendChild(document.createTextNode(' '));d3.appendChild(kv);
    }}
    g.appendChild(lab);g.appendChild(d2);g.appendChild(d3);
  }});
  var p=document.getElementById('presets');p.innerHTML='';
  NOISE_PRESETS.forEach(function(nm){{
    var b=nEl('button',{{style:'background:#234;color:#bdf;border:1px solid #456;border-radius:4px;padding:4px 8px;margin:2px;cursor:pointer;font-size:12px'}});
    b.textContent=nm;
    b.addEventListener('click',function(){{fetch('/set?noise_preset='+nm).then(function(){{setTimeout(pollNoise,120);}});}});
    p.appendChild(b);
  }});
  pollNoise();
}}
function pollNoise(){{fetch('/state').then(r=>r.json()).then(function(arr){{
  // single-radio panel: /state is a list; this radio is index 0 here
  var st=Array.isArray(arr)?arr[0]:arr; if(!st||!st.noise)return;
  NOISE_CH.forEach(function(c){{var name=c[0],ch=st.noise[name];if(!ch)return;
    var cb=document.getElementById('nc_'+name);if(cb)cb.checked=!!ch.enabled;
    var l=document.getElementById('nl_'+name);if(l){{l.value=ch.level_db;
      document.getElementById('nlv_'+name).textContent=ch.level_db;}}
    var kk=c[2]?c[2].replace(',',''):null;
    if(kk&&ch[kk]!==undefined){{var k=document.getElementById('nk_'+name);
      if(k){{k.value=ch[kk];document.getElementById('nkv_'+name).textContent=ch[kk];}}}}
  }});
  var vw=document.getElementById('noise_voicewav');
  if(vw)vw.innerHTML=st.noise.voice&&st.noise.voice.enabled?
    ('Voice channel uses WAV: <code>'+(st.noise_wav||'(none set — use the WAV path box above)')+'</code>'):'';
}}).catch(()=>{{}});}}

loadFixtures();loadRulers();loadReports();buildNoise();
pollStatus();setInterval(pollStatus,1500);setInterval(pollNoise,2000);
upd();
</script></body></html>"""


# Pattern groups for the dropdown (presentation only — the HINTS map explains each).
# Any pattern not listed here falls into "Other" so new patterns still show up.
PATTERN_GROUPS = [
    ("Realistic (looks like on-air)", ["ssb"]),
    ("Calibration rulers", ["carrier", "two_tone", "noise_cal", "cal_tones", "comb", "test_card"]),
    ("Diagnostic / stress", ["ramp", "swept_carrier", "step", "impulse", "staircase", "tx_blank", "noise"]),
    ("Keying", ["cw"]),
]


def _grouped_pattern_options(current):
    def opt(p):
        return f'<option value="{p}"{" selected" if p == current else ""}>{p}</option>'
    grouped, seen = [], set()
    for label, pats in PATTERN_GROUPS:
        avail = [p for p in pats if p in PATTERNS]
        if not avail:
            continue
        seen.update(avail)
        grouped.append(f'<optgroup label="{label}">' + "".join(opt(p) for p in avail) + "</optgroup>")
    other = [p for p in sorted(PATTERNS) if p not in seen]
    if other:
        grouped.append('<optgroup label="Other">' + "".join(opt(p) for p in other) + "</optgroup>")
    return "".join(grouped)


def start_control_server(radio, port):
    opts = _grouped_pattern_options(radio.pattern)
    page = CONTROL_HTML.format(options=opts, tt_khz=f"{TWO_TONE_SPACING_HZ / 1000.0:g}").encode()

    class H(http.server.BaseHTTPRequestHandler):
        def log_message(self, *a):
            pass

        def _json(self, obj):
            body = json.dumps(obj).encode()
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Cache-Control", "no-store")
            self.end_headers(); self.wfile.write(body)

        def do_GET(self):
            u = urllib.parse.urlparse(self.path)
            # ---- read-only QA regression-suite results viewer (optional module) ----
            if u.path == "/qa" or u.path == "/qa/view":
                try:
                    import qa_viewer
                    return qa_viewer.handle(self, u.path)
                except Exception:
                    self.send_response(404); self.end_headers()
                    self.wfile.write(b"qa viewer not available"); return
            # ---- live status (panel polls this for connection + stream state) ----
            if u.path == "/status":
                return self._json({
                    "connected": radio.conn is not None,
                    "peer": radio.ae_peer_ip,
                    "streaming": radio.streaming,
                    "paused": radio.paused,
                    "pattern": radio.pattern,
                    "tx": radio.tx_on,
                    "meter_dbm": round(radio.last_vfo_dbm, 1),
                })
            if u.path == "/txchain":
                # The #4510 chain verdict — radio-side stages + first missing.
                return self._json(radio.txchain())
            if u.path == "/txchain/contend":
                # multiFLEX contention (#4640): ?on=1 makes ANOTHER client hold
                # the dax_tx TX bit, so our create no longer grants it and the
                # chain stops at stage 5 with the foreign owner named. ?on=0
                # returns the radio to single-client, where create grants TX
                # implicitly. Optional ?handle=<hex> sets the rival's handle.
                q = urllib.parse.parse_qs(u.query)
                on = q.get("on", ["1"])[0] == "1"
                try:
                    h = int(q.get("handle", ["0xDEADBEEF"])[0], 16)
                except ValueError:
                    h = 0xDEADBEEF
                radio.daxtx_foreign_owner = h if on else 0
                if radio.daxtx_id is not None:
                    # Ownership can move mid-session — that is the point: the
                    # same radio that worked a minute ago now produces no RF.
                    radio.daxtx_tx = not on
                return self._json(radio.txchain())
            if u.path == "/state":
                # List form (one radio) so the noise-bench JS is identical to the
                # rack panel's /state — pollNoise() reads arr[0].
                return self._json([radio.state()])
            # ---- test-bench routes (JSON / report files) ----
            if u.path == "/fixtures":
                return self._json(list_fixtures())
            if u.path == "/reports":
                return self._json(list_reports())
            if u.path == "/run-tests":
                q = urllib.parse.parse_qs(u.query)
                only = [s for s in q.get("only", [""])[0].split(",") if s in RULER_NAMES] or None
                return self._json({"id": start_test_run(only)})
            if u.path == "/run-status":
                q = urllib.parse.parse_qs(u.query)
                return self._json(test_run_status(q.get("id", [""])[0]))
            # ---- LIVE test (drive the real AE) ----
            if u.path == "/live-test/start":
                q = urllib.parse.parse_qs(u.query)
                try: dwell = float(q.get("dwell", ["5"])[0])
                except ValueError: dwell = 5.0
                return self._json(live_test_start(radio, dwell))
            if u.path == "/live-test/status":
                return self._json(live_test_status())
            if u.path == "/live-test/pause":
                q = urllib.parse.parse_qs(u.query)
                return self._json(live_test_set(paused=q.get("paused", ["1"])[0] == "1"))
            if u.path == "/live-test/stop":
                return self._json(live_test_set(abort=True))
            if u.path.startswith("/reports/"):
                # Serve a report file — basename only, must match the report pattern (no traversal).
                name = os.path.basename(u.path[len("/reports/"):])
                if not name.startswith("ruler-report_") or not name.endswith((".html", ".json")):
                    self.send_response(404); self.end_headers(); return
                fp = os.path.join(REPORTS_DIR, name)
                if not os.path.isfile(fp):
                    self.send_response(404); self.end_headers(); return
                self.send_response(200)
                self.send_header("Content-Type", "text/html" if name.endswith(".html") else "application/json")
                self.end_headers()
                with open(fp, "rb") as f:
                    self.wfile.write(f.read())
                return
            if u.path == "/set":
                q = urllib.parse.parse_qs(u.query)
                try:
                    if "floor" in q:   radio.noise_floor_dbm = float(q["floor"][0])
                    if "level" in q:   radio.sig_level_dbm = float(q["level"][0])
                    if "width" in q:   radio.sig_width_khz = float(q["width"][0])
                    if "tilt" in q:    radio.noise_color = float(q["tilt"][0])
                    if "txpw" in q:    radio.tx_power_w = float(q["txpw"][0])
                    if "swr" in q:     radio.tx_swr = float(q["swr"][0])
                    if "qsk" in q:     radio.qsk = q["qsk"][0] == "1"
                    if "paused" in q:
                        radio.paused = q["paused"][0] == "1"
                        log(f"[ctl] output {'PAUSED' if radio.paused else 'resumed'}")
                    if "tx" in q and radio.pattern not in AUTO_TX_PATTERNS and not radio.cwx_active:
                        newtx = q["tx"][0] == "1"
                        if newtx != radio.tx_mox:
                            radio.tx_mox = newtx
                            radio.emit_transmit_status()
                            log(f"[ctl] TX {'ON' if newtx else 'off'} "
                                f"({radio.tx_power_w:.0f} W, SWR {radio.tx_swr:.1f})")
                    if "pattern" in q and q["pattern"][0] in PATTERNS and q["pattern"][0] != radio.pattern:
                        leaving_auto = radio.pattern in AUTO_TX_PATTERNS
                        radio.pattern = q["pattern"][0]
                        if leaving_auto and radio.tx_mox:            # don't leave TX stuck on after a gap/over
                            radio.tx_mox = False
                            radio.emit_transmit_status()
                        log(f"[ctl] pattern -> {radio.pattern}  (floor={radio.noise_floor_dbm:.0f}dBm "
                            f"level={radio.sig_level_dbm:.0f}dBm width={radio.sig_width_khz:.0f} tilt={radio.noise_color:.0f})")
                    if "audio_src" in q:
                        src = q["audio_src"][0]
                        if src in (AUDIO_SRC_SILENCE, AUDIO_SRC_TONE, AUDIO_SRC_NOISE, AUDIO_SRC_WAV):
                            radio.audio_source = src
                            log(f"[ctl] audio_source -> {src}")
                    if "wav_path" in q:
                        radio.audio_wav_path = q["wav_path"][0] or None
                        log(f"[ctl] audio_wav_path -> {radio.audio_wav_path}")
                    apply_noise_params(radio, q)
                except ValueError:
                    pass
                self.send_response(200); self.end_headers(); self.wfile.write(b"ok")
            else:
                self.send_response(200)
                self.send_header("Content-Type", "text/html")
                self.send_header("Cache-Control", "no-store, no-cache, must-revalidate")
                self.send_header("Pragma", "no-cache")
                self.end_headers(); self.wfile.write(page)

    srv = http.server.ThreadingHTTPServer(("0.0.0.0", port), H)
    threading.Thread(target=srv.serve_forever, daemon=True).start()


# ---- the rack: one process hosting N radios + a combined "1U strip" panel ----
class Rack:
    def __init__(self, ip, ae_ip, n, base_port, pattern, bins, fps, width_khz, models=None,
                 serial_prefix=SERIAL_PREFIX):
        prefix, last = ip.rsplit(".", 1)                  # each radio gets its own IP (like real rigs):
        base_last = int(last)                             # base, base+1, ... on the same port (default 4992)
        models = models or []
        self.radios = []
        for i in range(n):
            rip = f"{prefix}.{base_last + i}"
            mdl = models[i % len(models)] if models else MODEL
            self.radios.append(Radio(rip, ae_ip, pattern, bins, fps, width_khz,
                                     port=base_port, radio_id=i, model=mdl,
                                     serial_prefix=serial_prefix))

    def start(self):
        for r in self.radios:
            threading.Thread(target=r.discovery_loop, daemon=True).start()
            threading.Thread(target=r.prime_loop, daemon=True).start()
            threading.Thread(target=r.serve, daemon=True).start()


RACK_HTML = """<!DOCTYPE html><html><head><meta charset=utf-8><title>flex-sim rack</title>
<style>
body{{font-family:sans-serif;background:#0d0f13;color:#dfe7ef;margin:0;padding:18px}}
h2{{color:#5cf;margin:0 0 2px}} .sub{{color:#6b7480;font-size:13px;margin-bottom:14px}}
.rack{{background:#14171c;border:1px solid #2b313b;border-radius:12px;padding:10px;display:flex;flex-direction:column;gap:8px}}
.strip{{display:flex;align-items:center;gap:8px;background:#1b2129;border:1px solid #2f3742;border-radius:8px;padding:8px 6px}}
.strip.off{{background:#15181d;border-color:#23272e;opacity:.55}}
.pwr{{width:26px;height:26px;border-radius:50%;border:1px solid #2c7a44;background:#16331f;color:#34d058;cursor:pointer;font-size:13px;line-height:1}}
.pwr.off{{border-color:#3a3f47;background:#23272e;color:#5a616b}}
.rid{{width:74px;line-height:1.15}} .rid b{{font-size:13px}} .rid span{{display:block;font-size:10px;color:#6b7480}}
.lcd{{width:112px;font-family:monospace;font-size:14px;color:#f0b34a;background:#0c0e12;border-radius:4px;padding:4px 6px;text-align:right}}
.mode{{width:46px;font-size:11px;color:#9fb3c8;background:#232a33;border-radius:4px;padding:4px 0;text-align:center}}
.pill{{width:44px;font-size:11px;border-radius:4px;padding:4px 0;text-align:center;font-weight:bold}}
.pill.rx{{color:#4ade80;background:#123a1e;border:1px solid #1f6b35}}
.pill.tx{{color:#f87171;background:#3a1212;border:1px solid #6b1f1f}}
.pill.off{{color:#5a616b;background:#1e232a}}
select{{width:116px;font-size:12px;color:#cfe0f0;background:#232a33;border:1px solid #38414d;border-radius:5px;padding:4px 6px}}
.mdl{{width:104px;font-size:11px}}
.met{{flex:1;display:flex;align-items:center;gap:6px}}
.bar{{flex:1;height:10px;background:#0c0e12;border-radius:3px;overflow:hidden}}
.fill{{height:100%;width:0;background:#2faf5a}}
.mlbl{{font-size:11px;color:#9fb3c8;width:36px}}
</style></head><body>
<h2>flex-sim rack</h2><div class=sub>{n} virtual radios &mdash; driving AetherSDR</div>
<div class=rack>{strips}</div>
<script>
function fmtFreq(m){{var hz=Math.round(m*1e6),s=String(hz).padStart(7,'0');
return (s.slice(0,-6)||'0')+'.'+s.slice(-6,-3)+'.'+s.slice(-3);}}
function setp(i,p){{fetch('/set?radio='+i+'&pattern='+p);}}
function setm(i,m){{fetch('/set?radio='+i+'&model='+m);}}
function pwr(i,on){{fetch('/set?radio='+i+'&power='+(on?1:0));}}
function S(d){{return d>=-73?('S9+'+Math.round(d+73)):('S'+Math.max(0,Math.round(9+(d+73)/6)));}}
async function poll(){{
 try{{var st=await(await fetch('/state')).json();st.forEach(function(s){{
  var g=function(p){{return document.getElementById(p+'-'+s.id);}};
  g('strip').className='strip'+(s.on?'':' off');
  g('pwr').className='pwr'+(s.on?'':' off');
  var live=s.on&&s.connected;
  g('lcd').textContent=live?fmtFreq(s.freq):'—.———.———';
  g('mode').textContent=live?s.mode:'—';
  var md=g('mdl'); if(md&&md.value!==s.model)md.value=s.model;
  var pl=g('state');
  if(!s.on){{pl.textContent='off';pl.className='pill off';}}
  else if(!s.connected){{pl.textContent='idle';pl.className='pill off';}}
  else if(s.tx){{pl.textContent='TX';pl.className='pill tx';}}
  else{{pl.textContent='RX';pl.className='pill rx';}}
  var f=g('fill'),l=g('mlbl');
  if(!live){{f.style.width='0';l.textContent='—';}}
  else if(s.tx){{f.style.width=Math.min(100,s.power_w)+'%';f.style.background='#e0892a';l.textContent=s.power_w+'W';}}
  else{{f.style.width=Math.max(0,Math.min(100,(s.meter_dbm+130)/1.1))+'%';f.style.background='#2faf5a';l.textContent=S(s.meter_dbm);}}
 }});}}catch(e){{}}
 setTimeout(poll,750);
}}
poll();
</script></body></html>"""


def start_rack_control_server(rack, port):
    def strip(r):
        opts = "".join(f'<option{" selected" if p == r.pattern else ""}>{p}</option>'
                       for p in sorted(PATTERNS))
        mopts = "".join(f'<option{" selected" if m == r.model else ""}>{m}</option>' for m in MODELS)
        i = r.radio_id
        return (f'<div class=strip id=strip-{i}>'
                f'<button class=pwr id=pwr-{i} title=power '
                f'onclick="pwr({i},this.classList.contains(\'off\'))">&#9211;</button>'
                f'<span class=rid title="{r.serial}"><b>R{i + 1}</b><span>{r.ip}</span></span>'
                f'<select class=mdl id=mdl-{i} onchange="setm({i},this.value)">{mopts}</select>'
                f'<span class=lcd id=lcd-{i}>&mdash;.&mdash;&mdash;&mdash;.&mdash;&mdash;&mdash;</span>'
                f'<span class=mode id=mode-{i}>&mdash;</span>'
                f'<span class="pill off" id=state-{i}>off</span>'
                f'<select id=pat-{i} onchange="setp({i},this.value)">{opts}</select>'
                f'<span class=met><span class=bar><span class=fill id=fill-{i}></span></span>'
                f'<span class=mlbl id=mlbl-{i}>&mdash;</span></span></div>')
    page = RACK_HTML.format(n=len(rack.radios), strips="".join(strip(r) for r in rack.radios)).encode()

    class H(http.server.BaseHTTPRequestHandler):
        def log_message(self, *a): pass

        def do_GET(self):
            u = urllib.parse.urlparse(self.path)
            if u.path == "/qa" or u.path == "/qa/view":
                # Read-only QA regression-suite results viewer. Optional: only
                # active if qa_viewer.py sits alongside flex_sim.py. Never fails
                # the panel if it's absent.
                try:
                    import qa_viewer
                    return qa_viewer.handle(self, u.path)
                except Exception:
                    self.send_response(404); self.end_headers()
                    self.wfile.write(b"qa viewer not available"); return
            if u.path == "/state":
                body = json.dumps([r.state() for r in rack.radios]).encode()
                self.send_response(200); self.send_header("Content-Type", "application/json")
                self.send_header("Cache-Control", "no-store"); self.end_headers()
                self.wfile.write(body); return
            if u.path == "/set":
                q = urllib.parse.parse_qs(u.query)
                try:
                    r = rack.radios[int(q.get("radio", ["0"])[0])]
                    if "power" in q:
                        r.set_power(q["power"][0] == "1")
                        log(f"[rack] R{r.radio_id + 1} power {'on' if r.enabled else 'off'}")
                    if "pattern" in q and q["pattern"][0] in PATTERNS and q["pattern"][0] != r.pattern:
                        leaving_auto = r.pattern in AUTO_TX_PATTERNS
                        r.pattern = q["pattern"][0]
                        if leaving_auto and r.tx_mox:
                            r.tx_mox = False; r.emit_transmit_status()
                        log(f"[rack] R{r.radio_id + 1} pattern -> {r.pattern}")
                    if "model" in q and q["model"][0] in MODELS and q["model"][0] != r.model:
                        r.model = q["model"][0]
                        r.max_slices = MODELS[r.model]["slices"]
                        log(f"[rack] R{r.radio_id + 1} model -> {r.model} (re-discovering)")
                        c = r.conn                         # drop -> AE re-discovers it as the new model
                        if c:
                            try: c.shutdown(socket.SHUT_RDWR); c.close()
                            except OSError: pass
                except (ValueError, IndexError):
                    pass
                self.send_response(200); self.end_headers(); self.wfile.write(b"ok"); return
            self.send_response(200)
            self.send_header("Content-Type", "text/html")
            self.send_header("Cache-Control", "no-store, no-cache, must-revalidate")
            self.send_header("Pragma", "no-cache")
            self.end_headers(); self.wfile.write(page)

    srv = http.server.ThreadingHTTPServer(("0.0.0.0", port), H)
    threading.Thread(target=srv.serve_forever, daemon=True).start()


def main():
    ap = argparse.ArgumentParser(prog="flex-sim",
                                 description="Synthetic FlexRadio-6000 emulator / test bench for AetherSDR.")
    ap.add_argument("--version", action="version", version=f"flex-sim {FLEX_SIM_VERSION}")
    ap.add_argument("--ip", default=None)
    ap.add_argument("--ae", default=None)
    ap.add_argument("--pattern", default="ramp", choices=sorted(PATTERNS))
    ap.add_argument("--bins", type=int, default=BINS)
    ap.add_argument("--fps", type=int, default=FPS)
    ap.add_argument("--width-khz", type=float, default=SIGNAL_WIDTH_KHZ)
    ap.add_argument("--port", type=int, default=DEFAULT_PORT,
                    help="control/data port we bind + advertise (default 4992; set e.g. 5992 to "
                         "coexist with AE on one host). Discovery is always sent to AE's :4992.")
    ap.add_argument("--radios", type=int, default=1,
                    help="number of virtual radios in one process (>1 = rack mode)")
    ap.add_argument("--base-port", type=int, default=None,
                    help="rack: port for the radios (each gets its own IP, so they share a port; default = --port)")
    ap.add_argument("--serial", default=SERIAL_PREFIX, metavar="PREFIX",
                    help=f"serial-number prefix (serial = PREFIX + 2-digit radio index). Defaults "
                         f"to this machine's hostname ({SERIAL_PREFIX}), so two sims on one "
                         f"network can never collide: AE's 'connect local serial <s>' matches on "
                         f"SERIAL, and identical prefixes make it silently take whichever radio it "
                         f"discovered first. Pass --serial FLEXSIM for the pre-2026-07-31 default")
    ap.add_argument("--models", default=None,
                    help="model(s): comma-list, one per radio in rack mode, e.g. "
                         "FLEX-6300,FLEX-6600,FLEX-6700. With a single radio the first entry "
                         "sets its model (FLEX-6600/6700 advertise 2-SCU, so AE offers DIV)")
    ap.add_argument("--ctl-port", type=int, default=8731, help="web control-panel port")
    ap.add_argument("--txlog", action="store_true",
                    help="log UDP arriving FROM AetherSDR: one line per new stream "
                         "(id/PCC/size) plus a periodic rate report. Use it to find out "
                         "which stream a TX mode really uses before building against it "
                         "(see TX_AUDIO_DESIGN.md). Off by default — it is a discovery tool.")
    ap.add_argument("--txlog-interval", type=float, default=5.0,
                    help="seconds between --txlog rate reports (default 5)")
    args = ap.parse_args()
    ip = args.ip or local_ip()
    if args.radios > 1:                                    # ---- rack mode: N radios + strip panel ----
        base = args.base_port or args.port
        models = [m.strip() for m in args.models.split(",")] if args.models else None
        rack = Rack(ip, args.ae, args.radios, base, args.pattern, args.bins, args.fps, args.width_khz,
                    models=models, serial_prefix=args.serial)
        rack.start()
        start_rack_control_server(rack, args.ctl_port)
        log(f"flex-sim {FLEX_SIM_VERSION} - RACK of {args.radios} radios on :{base}")
        for r in rack.radios:
            log(f"   R{r.radio_id + 1}  {r.serial}  {r.model}  {r.ip}:{r.port}")
        log(f"** rack panel: http://{ip}:{args.ctl_port}/  (open in your host browser) **")
        try:
            while True:
                time.sleep(1)
        except KeyboardInterrupt:
            log("bye")
        return
    # --models also selects the model for a SINGLE radio (first entry wins); an
    # unknown name falls back to MODEL inside Radio.__init__.
    single_model = ([m.strip() for m in args.models.split(",")] or [MODEL])[0] if args.models else MODEL
    radio = Radio(ip, args.ae, args.pattern, args.bins, args.fps, args.width_khz,
                  port=args.port, model=single_model, serial_prefix=args.serial)
    radio.txlog = args.txlog
    radio.txlog_interval = args.txlog_interval
    if args.txlog:
        log(f"[txlog] enabled — logging UDP from AE (report every {args.txlog_interval:g}s)")
    threading.Thread(target=radio.discovery_loop, daemon=True).start()
    threading.Thread(target=radio.prime_loop, daemon=True).start()
    start_control_server(radio, args.ctl_port)
    # Report the RADIO's model, not the module default — they differ whenever
    # --models is used, and logging the global made a wrong model look correct.
    log(f"flex-sim {FLEX_SIM_VERSION} - model={radio.model} serial={radio.serial} ip={ip} pattern={args.pattern}")
    log(f"discovery -> AE's UDP :{DISCOVERY_PORT}; control/data on :{args.port}"
        + ("  (same-host mode)" if args.port != DISCOVERY_PORT else ""))
    log(f"** control panel: http://{ip}:{args.ctl_port}/  (open in your host browser) **")
    try:
        radio.serve()
    except KeyboardInterrupt:
        radio.run = False
        log("bye")


if __name__ == "__main__":
    main()