diff --git a/docs/adaptive-link.md b/docs/adaptive-link.md index 7a8fe464..06a65587 100644 --- a/docs/adaptive-link.md +++ b/docs/adaptive-link.md @@ -107,6 +107,28 @@ with only local safety overrides (thermal back-off, a max-range failsafe when feedback is lost). This is the same "ground decides" stance the OpenIPC world uses; the difference is purely the cost function it optimizes. +**The model proposes; measurement disposes.** A single SNR reading plus a link +model is a *counterfactual* about every rate not currently flying, and on real +hardware the counterfactual can be wrong — receiver quirks, multipath, +interference, amplifier distortion and calibration error are all rate-specific. +So a small, deterministic share of frames (a few percent) flies the rates +adjacent to the selected one with everything else held fixed, and both ends +derive which frames those were from the sequence number alone — no signalling. +The receiver then holds *measured* per-rate delivery with honest small-sample +confidence bounds — for the adjacent candidates from the probes, and for the +selected rate itself from the main stream, whose evidence is free and arrives +at full frame rate. The evidence corrects the model in the safe direction +only: a faster rate is entered only once its measured lower bound clears what +the FEC needs; a candidate whose upper bound cannot is barred outright; and a +selected rate whose own measured delivery stays condemned is abandoned and +barred even while the model still believes in it — the escape hatch for the +failures no SNR reading shows (interference, a miscalibrated table), and what +makes the deliberately model-trusted first pick safe to keep fast. Emergencies +keep trusting the model, because escaping a failing rate must never wait for +samples. The trade is deliberate: a mis-modelled rate can no longer trap the +controller, at the cost of slower, evidence-paced promotions and the probes' +small airtime, both priced in the same energy ledger as everything else. + **Re-finding each other.** When the command uplink drops, the drone falls to a robust failsafe and then to a low-duty listen on a known channel; the mains-powered ground station beacons for it quickly. The duty cycle is diff --git a/tests/devourer_events.py b/tests/devourer_events.py index 780fb2e0..18fbb9b2 100644 --- a/tests/devourer_events.py +++ b/tests/devourer_events.py @@ -53,3 +53,20 @@ def iter_events(lines, ev=None): obj = parse_event(line, ev) if obj is not None: yield obj + + +def desc_rate_to_mcs(code): + """rx.frame's `rate` (the chip's DESC_RATE index) -> ("ht"|"vht", mcs). + + Returns None for legacy/CCK and for multi-SS VHT: the consumers (the + adaptive link's rate-verified probe attribution) fly 1SS, and an + unclassifiable rate must read as "no rate evidence", never as a match. + HT keeps the full 0..31 index range so a multi-SS HT frame maps outside + a 1SS mcs_set and is ignored naturally.""" + if code is None: + return None + if 12 <= code <= 43: # DESC_RATEMCS0..MCS31 + return ("ht", code - 12) + if 44 <= code <= 53: # DESC_RATEVHTSS1MCS0..9 + return ("vht", code - 44) + return None diff --git a/tests/mcs_probe_onair.sh b/tests/mcs_probe_onair.sh new file mode 100755 index 00000000..42ce2bce --- /dev/null +++ b/tests/mcs_probe_onair.sh @@ -0,0 +1,136 @@ +#!/usr/bin/env bash +# On-air A/B for the adaptive link's adjacent-MCS probes: VTX (8812) <-> VRX +# (8821), both running duplex driven by adaptive_link.py, same rig as +# tests/adaptive_onair.sh. Four paired cells on one channel: +# +# A probe=off unbiased — baseline; discovers the settled MCS M +# B probe=off bias (M+1) — the model-only controller believes M+1 works +# at 12 dB less SNR than it does and promotes +# into it (the miscalibration trap, on-air) +# C probe=on bias (M+1) — probe evidence gates/blocks the same lie +# D probe=on unbiased — probe cost vs A (duty, operating point) +# +# The mismatch needs the upper rates to GENUINELY fail on this channel, which +# near-field bench SNR does not give — so the B210 interferer runs by default +# (USE_INTERFERER=0 disables; without it the harness still validates the +# plumbing end-to-end: probes fly, stats accrue, no baseline regression, but +# the B-vs-C delivery contrast may not appear). The interferer's effective +# bite sits on a ~2 dB cliff and drifts between cells/runs, so judge within +# one run and lean on the SAME-BIAS pair: B vs C isolates the probes as the +# only difference. frames/win is the health signal; deliv= is seq-based and +# lies under crc floods. +# +# sudo bash tests/mcs_probe_onair.sh +# USE_INTERFERER=0 sudo bash tests/mcs_probe_onair.sh # plumbing only +set -u +ROOT="$(cd "$(dirname "$0")/.." && pwd)" +PREC="$ROOT/tools/precoder" + +VTX_PID=${VTX_PID:-0x8812}; VTX_VID=${VTX_VID:-0x0bda}; VTX=${VTX_SYSFS:-9-2} +VRX_PID=${VRX_PID:-0x0120}; VRX_VID=${VRX_VID:-0x2357}; VRX=${VRX_SYSFS:-9-1.4} +CH=${CH:-6}; SECS=${SECS:-40}; VTX_ID=${VTX_ID:-0xABCD} +USE_INTERFERER=${USE_INTERFERER:-1}; IGAIN=${IGAIN:-75}; IMODE=${IMODE:-noise} +# Keep the host otherwise idle: B210 TX underruns (host-load-starved) turn the +# interferer into chaotic bursts and the cells stop being comparable. +IRATE=${IRATE:-20e6} +# Mild optimism: a few dB is the realistic miscalibration AND keeps the biased +# row out of the (deliberately ungated) cold pick — the trap the probes exist +# for is the PROMOTION into the lie, not a wrong first guess. +BIAS_DB=${BIAS_DB:--4} +LOGDIR=${LOGDIR:-/tmp/mcs-probe-onair} +VIDEO=$LOGDIR/video.bin + +KILL(){ sudo pkill -9 -f adaptive_link 2>/dev/null; sudo pkill -9 duplex 2>/dev/null + sudo pkill -9 -f sdr_interferer 2>/dev/null; } +trap KILL EXIT + +mkdir -p "$LOGDIR" +head -c 4000000 /dev/urandom > "$VIDEO" + +# free both Wi-Fi adapters (B210 is uhd-accessed) +for D in "$VTX" "$VRX"; do + for i in /sys/bus/usb/devices/$D/$D:*; do + ifc=$(basename "$i"); drv=$(readlink -f "$i/driver" 2>/dev/null) + [ -n "$drv" ] && echo "$ifc" | sudo tee "$drv/unbind" >/dev/null 2>&1 + done +done; sleep 1 + +# ---- one cell: [interferer] + VRX + VTX for SECS, logs under $LOGDIR/ +run_cell(){ # name probe_flag bias_spec + local name=$1 probe=$2 bias=$3 + local vrx_log=$LOGDIR/${name}_vrx.log vtx_log=$LOGDIR/${name}_vtx.log + local vrx_args=(--role vrx --pid $VRX_PID --channel $CH --vtx-id $VTX_ID + --duplex "$ROOT/build/duplex") + local vtx_args=(--role vtx --pid $VTX_PID --channel $CH --vtx-id $VTX_ID + --video "$VIDEO" --duplex "$ROOT/build/duplex") + [ "$probe" = 1 ] && { vrx_args+=(--mcs-probe); vtx_args+=(--mcs-probe); } + [ -n "$bias" ] && vrx_args+=(--mcs-bias "$bias") + + KILL; sleep 1 + if [ "$USE_INTERFERER" = 1 ]; then + sudo python3 "$ROOT/tests/sdr_interferer.py" --channel $CH --tx-gain "$IGAIN" \ + --rate "$IRATE" --mode "$IMODE" --secs $((SECS + 30)) \ + >"$LOGDIR/${name}_intf.log" 2>&1 & + sleep 11 + fi + echo "=== cell $name: probe=$probe bias='${bias:-none}' ===" + sudo env DEVOURER_VID=$VRX_VID DEVOURER_PID=$VRX_PID PYTHONPATH="$PREC" \ + python3 "$PREC/adaptive_link.py" "${vrx_args[@]}" >"$vrx_log" 2>&1 & + sleep 6 + sudo env DEVOURER_VID=$VTX_VID DEVOURER_PID=$VTX_PID PYTHONPATH="$PREC" \ + python3 "$PREC/adaptive_link.py" "${vtx_args[@]}" >"$vtx_log" 2>&1 & + sleep "$SECS" + KILL; sleep 1 + grep '' "$vrx_log" | tail -1 +} + +# deliv= is only meaningful on populated windows: near-empty windows read +# 1.000 (no gaps to see) and crc-flooded ones read ~0 (corrupt bodies carry +# garbage seqs) — so gate on frames>=20 and report the window fill too. +cell_deliv(){ grep -oP '.*deliv=\K[0-9.]+(?=.*frames=(2[0-9]|[3-9][0-9]|[0-9]{3,}))' \ + "$LOGDIR/${1}_vrx.log" | tail -20 | sort -n \ + | awk '{a[NR]=$1} END{print (NR ? a[int(NR/2)+1] : "dead")}'; } +cell_frames(){ grep -oP '.*frames=\K\d+' "$LOGDIR/${1}_vrx.log" \ + | tail -20 | sort -n | awk '{a[NR]=$1} END{print (NR ? a[int(NR/2)+1] : 0)}'; } +cell_mcs(){ grep -oP '.*-> MCS\K\d+' "$LOGDIR/${1}_vrx.log" | tail -1; } +flew_mcs(){ grep -c "ladder=CRIT=MCS${2}/" "$LOGDIR/${1}_vtx.log" 2>/dev/null || true; } + +run_cell A 0 "" +M=$(cell_mcs A) +if [ -z "$M" ]; then echo "FATAL: cell A produced no controller trajectory"; exit 1; fi +T=$((M + 1)) +if [ "$T" -gt 7 ]; then + echo "WARN: settled at MCS$M (top of the set) — no row above to bias;" + echo " raise IGAIN to pull the operating point down and re-run." + T=7 +fi +BIAS_SPEC="$T:$BIAS_DB" +echo "settled MCS$M -> biasing MCS$T by ${BIAS_DB} dB (model-optimistic)" + +run_cell B 0 "$BIAS_SPEC" +run_cell C 1 "$BIAS_SPEC" +run_cell D 1 "" + +echo +echo "=== RESULT (settled MCS$M, biased candidate MCS$T) ===" +printf "%-4s %-8s %-10s %-14s %-10s %-14s %s\n" cell probe bias "median-deliv" "frames/win" "flew-MCS$T(s)" "probe-evidence" +for c in A B C D; do + case $c in A) p=off b=none;; B) p=off b=yes;; C) p=on b=yes;; D) p=on b=none;; esac + pe=$(grep -c 'probe\[MCS' "$LOGDIR/${c}_vrx.log" 2>/dev/null || true) + printf "%-4s %-8s %-10s %-14s %-10s %-14s %s\n" \ + "$c" "$p" "$b" "$(cell_deliv $c)" "$(cell_frames $c)" "$(flew_mcs $c $T)" "${pe} lines" +done +echo +echo "verdict guide:" +echo " A vs B — the trap on-air: B flies MCS$T and its frames/win collapse." +echo " C — the measured ESCAPE: on this static bench a bias attractive" +echo " enough to matter is usually captured by the (model-trusted) cold" +echo " pick, so C may fly MCS$T briefly — but its measured active-row" +echo " delivery must condemn it within seconds (flew-MCS$T well below" +echo " B's), delivery must recover toward A/D, and probe[..] evidence" +echo " lines must be present (zero probe[MCS lines = dead rate feed)." +echo " D — the promotion gate with real evidence: D must not fly a" +echo " candidate whose measured lcb is low." +echo " (The promotion gate + escape are also pinned deterministically by" +echo " mcs_probe_ab_sim.py, incl. the no-acquisition 'ambush' scenario.)" +echo "logs: $LOGDIR" diff --git a/tools/precoder/README.md b/tools/precoder/README.md index 9c089cdb..dec911cd 100644 --- a/tools/precoder/README.md +++ b/tools/precoder/README.md @@ -90,6 +90,49 @@ OpenIPC's `alink` and other adaptive systems — is documented in `rc_proto.py`, `score.py`, `rendezvous.py`, `adaptive_link.py`) and the offline energy headline (`tests/sim_loop.py`) all live alongside this suite. +### Live adjacent-MCS probes (opt-in) + +`--mcs-probe` (both roles) turns on measured candidate PER: ~3% of video seqs +fly the rates adjacent to the commanded MCS (`rc_proto.probe_mcs`, slots 4/20 +of each 64-seq cycle — disjoint from the bandwidth-probe slots, so every probe +changes exactly one variable). Schedule + `mcs_set` + the commanded profile are +a VTX↔VRX **protocol invariant** (versioned: `rc_proto.MCS_PROBE_SCHED_VER`), +like `bw_set`. The VRX attributes received/gap-lost probe seqs per candidate +(`score.McsProbeWindow` — successes are rate-verified against the frame's +PHY-decoded rate, losses are epoch-gated across profile transitions, crc_err +bodies are never trusted for a seq) and the controller uses Wilson confidence +bounds on the evidence (`Controller.report_mcs_delivery`): while the current +point still works, a row above the current MCS is admissible only when its +lower bound clears the raw frame delivery its FEC needs +(`mcs_probe_min_samples` fresh samples within `mcs_probe_max_age_ms`), and a +candidate whose upper bound cannot reach even the heaviest overhead's +requirement is blocked for `mcs_probe_block_hold_ms`. The ACTIVE row is +measured too — every non-probe seq attributes to it (bandwidth-probe slots +excluded), so its evidence arrives at full frame rate and can condemn an +operating point the model still believes in (interference and miscalibration +are SNR-blind): after `mcs_probe_active_trip_ticks` consecutive condemned +reports (the persistence guard that lets bursts ride through) the row is +blocked, `cur_ok` fails despite the model, and the fast-down escapes — which +is what makes the deliberately model-trusted cold pick safe: a cold pick that +lands on a lie is measured out within seconds and the surviving block keeps +it from being re-picked. Total video silence takes the symmetric stance: the +VRX resets to MAX_RANGE and forgets the committed point (measured blocks +survive), so an operating point it cannot even hear is never commanded +indefinitely. Two consequences to know: the raw-delivery +requirement rises steeply as overhead falls, so promotions land at mid/heavy +FEC first and the (ungated) same-MCS overhead trim follows; and probes fly at +the incumbent's TXAGC, so a promotion needs the faster rate to already work at +the current power — with probing on, upward moves are rarer and evidence-paced. +Up-probes never ride protected frames (`probe_mcs_for_seq(..., protected=True)`) +nor failsafe/listen/discovery states, and `suspend_upward_probes()` is the +thermal/congestion back-off hook. A/B evidence: `mcs_probe_ab_sim.py` +(model-vs-truth miscalibration across static/triangle/fade/burst — the +model-only arm promotes into an optimistically-biased row and stays, the probe +arm holds the SLA) and `tests/mcs_probe_onair.sh` (paired on-air cells; the +`--mcs-bias` flag fabricates the model optimism on the VRX). Off by default: +`mcs_probe_enabled=False` is behaviourally identical to the pre-probe +controller. + ## End-to-end recipe ```sh diff --git a/tools/precoder/adaptive_link.py b/tools/precoder/adaptive_link.py index 7cbfe462..20c1c576 100644 --- a/tools/precoder/adaptive_link.py +++ b/tools/precoder/adaptive_link.py @@ -25,7 +25,7 @@ import rc_proto as rp import rendezvous as rz from controller import Controller, ControllerConfig -from score import ScoreWindow, ScoreConfig, RungWindow +from score import ScoreWindow, ScoreConfig, RungWindow, McsProbeWindow import op_table @@ -49,6 +49,14 @@ def op_to_ladder(op) -> str: return ladder_spec(getattr(op, "mode", "ht"), op.mcs, op.bw) +def rate_spec(mode: str, mcs: int, bw: int) -> str: + """One rate as a SET_RATE / DEVOURER_TX_RATE spec string. The /bw suffix is + always explicit so a probe frame and its restore pin the commanded + bandwidth rather than inheriting whatever the session last flew.""" + name = "VHT1SS_MCS" if mode == "vht" else "MCS" + return f"{name}{mcs}/{bw}" + + def overhead_to_16ths(ov: float) -> int: return max(1, min(16, round(ov * 16))) @@ -72,25 +80,51 @@ def __init__(self, link, calib, vtx_id: int, # embedding app turns into a coordinated channel move. bw_set = self.ctrl.cfg.bw_set self.rungs = RungWindow(bw_set) if bw_set else None + # Adjacent-MCS probe evidence (opt-in): the selected rate follows the + # controller's committed op, so the window resets itself on every + # commanded (mode, mcs, bw) transition — see _track_selected(). + cfg = self.ctrl.cfg + self.mcs_win = McsProbeWindow( + cfg.mcs_set, mode=cfg.mode, + samples_per_mcs=cfg.mcs_probe_window_samples, + max_age_ms=cfg.mcs_probe_max_age_ms, + confidence=cfg.mcs_probe_confidence, + bw_set=cfg.bw_set) if cfg.mcs_probe_enabled else None + self._sel_key: tuple | None = None self.rz = rz.VrxRendezvous(rz.VrxConfig(vtx_id=vtx_id, op_channel=op_channel)) self.feedback_period_ms = feedback_period_ms self._last_fb_ms = -1 << 60 self._seq = 0 self.cur_op = op_table.MAX_RANGE self._cur_txagc = 32 + self._video_lost = False @property def primary_dirty(self) -> bool: return self.ctrl.primary_dirty def on_video(self, rssi: float, snr: float, crc_err: bool, seq: int, - now_ms: float, residual_loss: float | None = None) -> None: + now_ms: float, residual_loss: float | None = None, + rate: tuple | None = None) -> None: + """`rate` is the frame's PHY-decoded (mode, mcs) when the event stream + carries it (devourer_events.desc_rate_to_mcs) — the probe attribution + is rate-verified, so without it the MCS window stays fail-inert.""" self.win.add_frame(rssi, snr, crc_err, seq, now_ms / 1000.0) if self.rungs is not None: self.rungs.add_seq(seq) + if self.mcs_win is not None: + self.mcs_win.add_seq(seq, now_ms, rate=rate, crc_err=crc_err) self.rz.feed_video(now_ms) self._residual = residual_loss + def _track_selected(self, now_ms: float) -> None: + op = self.cur_op + key = (getattr(op, "mode", "ht"), op.mcs, op.bw) + if key != self._sel_key: + self._sel_key = key + if self.mcs_win is not None: + self.mcs_win.set_selected(key[0], key[1], now_ms) + def on_disc_ack(self, buf: bytes, now_ms: float) -> None: ack = rp.parse_disc_ack(buf) if ack: @@ -100,20 +134,36 @@ def step(self, now_ms: float) -> bytes | None: """Return a frame to TX back to the VTX (RCF or DISC), or None.""" act = self.rz.tick(now_ms) if act == rz.A_BEACON: + # Video gone entirely: an operating point we cannot hear is not + # worth keeping, and with zero frames the estimator can never + # condemn it — the frozen-command deadlock. Take the max-range + # stance (the dual of the VTX's RC-loss failsafe; RCFs stopping is + # what lets the VTX failsafe fire) and re-pick fresh on + # reacquisition; measured MCS blocks survive the reset so the new + # cold pick avoids the rows the evidence just condemned. + if not self._video_lost: + self._video_lost = True + self.cur_op = op_table.MAX_RANGE + self.ctrl.reset_operating_point() + self._track_selected(now_ms) return rp.pack_disc(self.rz.beacon()) if act != rz.A_TX_FEEDBACK: return None + self._video_lost = False if now_ms - self._last_fb_ms < self.feedback_period_ms: return None self._last_fb_ms = now_ms if self.rungs is not None: self.ctrl.report_rung_delivery(self.rungs.stats(), now_ms) + if self.mcs_win is not None: + self.ctrl.report_mcs_delivery(self.mcs_win.stats(now_ms), now_ms) snr = self.win.snr_estimate() if snr is not None: op = self.ctrl.update(snr, self._cur_txagc, now_ms) if op is not None: self.cur_op = op self._cur_txagc = op.txagc + self._track_selected(now_ms) self._seq = (self._seq + 1) & 0xFFFF # profile carries the GS-chosen operating point — mode/MCS/bw packed by # the shared v2 encoding; the VTX builds its SVC ladder from it @@ -137,13 +187,20 @@ class TxState: overhead: float = 0.25 ladder: str = "CRIT=MCS1/20;T0=MCS2/20;T1=MCS4/20;T2=MCS5/20" failsafe: bool = False + # The commanded base rate as fields (kept in lockstep with `ladder`): the + # live SET_RATE plumbing and the MCS-probe schedule derive from these + # rather than re-parsing the ladder string. + mode: str = "ht" + mcs: int = 1 + bw: int = 20 class AdaptiveVtx: def __init__(self, vtx_id: int, vtx_cfg: rz.VtxConfig | None = None, failsafe_txagc: int = 63, failsafe_overhead: float = 1.0, failsafe_ladder: str | None = None, - bw_set: tuple | None = None): + bw_set: tuple | None = None, + mcs_probe: bool = False, mcs_set: tuple = tuple(range(8))): self.vtx_id = vtx_id self.rz = rz.VtxRendezvous(vtx_cfg or rz.VtxConfig(vtx_id=vtx_id)) self.state = TxState() @@ -154,6 +211,13 @@ def __init__(self, vtx_id: int, vtx_cfg: rz.VtxConfig | None = None, # Bandwidth-probe rungs — must match the VRX controller's bw_set (a # config invariant, like the profile-table version). None = no probing. self.bw_set = tuple(sorted(bw_set)) if bw_set else None + # Adjacent-MCS probing — mcs_set is the same config invariant as + # bw_set; the schedule itself (rc_proto.probe_mcs) is versioned. + self.mcs_probe = mcs_probe + self.mcs_set = tuple(sorted(mcs_set)) + self.probe_counters: dict[int, int] = {} # candidate mcs -> attempts + self._last_act: str | None = None + self._up_suspended = False def probe_bw_for_seq(self, seq: int) -> int | None: """Rung this video seq must fly at as a bandwidth probe (the shared @@ -164,6 +228,33 @@ def probe_bw_for_seq(self, seq: int) -> int | None: return None return rp.probe_bw(seq, self.bw_set) + def suspend_upward_probes(self, suspended: bool) -> None: + """Hook for the embedding app's back-off signals (thermal, congestion, + delivery collapse): an up-probe is speculative airtime the link may not + afford, a down-probe stays useful (it tells 'drop MCS' apart from 'all + rates failing').""" + self._up_suspended = bool(suspended) + + def probe_mcs_for_seq(self, seq: int, protected: bool = False) -> int | None: + """Candidate MCS this video seq must fly at as a rate probe (shared + schedule, commanded bandwidth/power/FEC untouched), else None -> the + commanded rate. `protected` marks a frame whose loss is not disposable + (control, base-layer critical, IDR): it never carries an UP-probe — + losing it to a speculative faster rate would damage exactly the traffic + the SLA protects. Down-probes are more robust than the operating point + and stay allowed. Probing pauses entirely outside the steady TX-video + state (failsafe/listen/discovery): those regimes fly reach-first.""" + if not self.mcs_probe or self.state.failsafe \ + or self._last_act != rz.A_TX_VIDEO: + return None + cand = rp.probe_mcs(seq, self.state.mcs, self.mcs_set) + if cand is None: + return None + if cand > self.state.mcs and (protected or self._up_suspended): + return None + self.probe_counters[cand] = self.probe_counters.get(cand, 0) + 1 + return cand + def on_rc_frame(self, buf: bytes, now_ms: float) -> bytes | None: """Process an inbound control frame. Applies an RCF (updates TxState) or answers a DISC with a DISC_ACK to TX. Returns a frame to TX, or None.""" @@ -181,6 +272,7 @@ def on_rc_frame(self, buf: bytes, now_ms: float) -> bytes | None: # (mode/MCS/bw, shared v2 encoding + shared ladder builder). mode, m, bw = rp.decode_profile(r.profile) self.state.ladder = ladder_spec(mode, m, bw) + self.state.mode, self.state.mcs, self.state.bw = mode, m, bw return None if t == rp.T_DISC: d = rp.parse_disc(buf) @@ -194,10 +286,12 @@ def step(self, now_ms: float) -> str: """Advance the failsafe/discovery SM; clamp TxState to failsafe when lost. Returns the rendezvous action (tx_video / failsafe / listen / idle).""" act = self.rz.tick(now_ms) + self._last_act = act if act == rz.A_FAILSAFE: self.state = TxState(txagc=self.failsafe_txagc, overhead=self.failsafe_overhead, - ladder=self.failsafe_ladder, failsafe=True) + ladder=self.failsafe_ladder, failsafe=True, + mode="ht", mcs=0, bw=20) # match failsafe_ladder return act def apply_ladder_from_op(self, op) -> None: @@ -256,7 +350,7 @@ def selftest(verbose: bool = False): from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "tests")) -from devourer_events import parse_event # noqa: E402 +from devourer_events import parse_event, desc_rate_to_mcs # noqa: E402 def ctl_frame(op: int, payload: bytes = b"") -> bytes: @@ -272,16 +366,30 @@ def psdu_frame(body: bytes) -> bytes: SET_PWR, SET_RATE, SET_CHAN = 1, 2, 3 +def _parse_mcs_bias(spec: str) -> dict[int, float]: + """'4:-6,5:-8' -> {4: -6.0, 5: -8.0} (per-MCS model snr_req shift, dB).""" + out = {} + for part in spec.split(","): + m, db = part.split(":") + out[int(m)] = float(db) + return out + + def _now_ms() -> float: import time return time.monotonic() * 1000.0 -def run_vrx(proc, link, calib, vtx_id, channel, feedback_period_ms=100): +def run_vrx(proc, link, calib, vtx_id, channel, feedback_period_ms=100, + mcs_probe=False, mcs_bias=None): """Read the VTX's video frames, score, run the controller, TX feedback.""" vrx = AdaptiveVrx(link, calib, vtx_id, - ctrl_cfg=ControllerConfig(target=0.99, allow_shed=False), + ctrl_cfg=ControllerConfig(target=0.99, allow_shed=False, + mcs_probe_enabled=mcs_probe), op_channel=channel, feedback_period_ms=feedback_period_ms) + if mcs_bias: + from controller import apply_model_bias + apply_model_bias(vrx.ctrl, mcs_bias) import threading def reader(): @@ -300,7 +408,8 @@ def reader(): # frame incl. retries, so it over-counts loss — see score.seq_gap_loss.) seq = (int.from_bytes(body[:2], "little") & 0xFFF if len(body) >= 2 else ev["seq"]) - vrx.on_video(rssi, snr, ev["crc"] != 0, seq, _now_ms()) + vrx.on_video(rssi, snr, ev["crc"] != 0, seq, _now_ms(), + rate=desc_rate_to_mcs(ev.get("rate"))) threading.Thread(target=reader, daemon=True).start() import time last_log = 0.0 @@ -313,19 +422,42 @@ def reader(): if now - last_log > 1000: # 1 Hz trajectory log (to stderr) last_log = now op = vrx.cur_op + # per-candidate probe evidence: attempts@lcb — a silent line here + # with probing on means the rate feed is dead, not the probes good + probes = "" + if vrx.mcs_win is not None: + st = vrx.mcs_win.stats(now) + probes = " probe[" + ",".join( + f"MCS{m}:{s.attempts}@{s.lcb:.2f}" + for m, s in sorted(st.items())) + "]" sys.stderr.write( f"state={vrx.rz.state} snr={vrx.win.snr_estimate()} " f"score={vrx.win.score()} deliv={1.0 - vrx.win.seq_gap_loss():.3f} " f"-> MCS{op.mcs} ov{op.overhead} " - f"txagc{op.txagc} frames={vrx.win.n()}\n") + f"txagc{op.txagc} frames={vrx.win.n()}{probes}\n") sys.stderr.flush() time.sleep(0.01) -def run_vtx(proc, vtx_id, video_path, channel): - """Stream video; apply inbound RCF/DISC to the live knobs via control ops.""" - vtx = AdaptiveVtx(vtx_id) +def run_vtx(proc, vtx_id, video_path, channel, mcs_probe=False): + """Stream video; apply inbound RCF/DISC to the live knobs via control ops. + + All stdin writers share one lock, and each logical action goes down the + pipe as ONE composed write: duplex's TX thread consumes stdin strictly in + order, so a probe's SET_RATE / PSDU / SET_RATE-restore triple applies + exactly to its frame — and the reader thread can't interleave a fresher + commanded rate into the middle of it.""" + vtx = AdaptiveVtx(vtx_id, mcs_probe=mcs_probe) import threading + wlock = threading.Lock() + + def _write(buf: bytes) -> None: + with wlock: + proc.stdin.write(buf) + proc.stdin.flush() + + def _cmd_rate() -> bytes: + return rate_spec(vtx.state.mode, vtx.state.mcs, vtx.state.bw).encode() def reader(): for line in proc.stdout: @@ -336,14 +468,16 @@ def reader(): t = rp.frame_type(body) if t not in (rp.T_RCF, rp.T_DISC): continue - ack = vtx.on_rc_frame(body, _now_ms()) - if ack is not None: # DISC_ACK to TX - proc.stdin.write(psdu_frame(ack)); proc.stdin.flush() - if t == rp.T_RCF: # apply the new operating point - proc.stdin.write(ctl_frame(SET_PWR, bytes([vtx.state.txagc]))) - base = f"MCS{vtx.state.ladder.split('CRIT=MCS')[1].split('/')[0]}" - proc.stdin.write(ctl_frame(SET_RATE, base.encode())) - proc.stdin.flush() + with wlock: # state + pipe change together + ack = vtx.on_rc_frame(body, _now_ms()) + out = b"" + if ack is not None: # DISC_ACK to TX + out += psdu_frame(ack) + if t == rp.T_RCF: # apply the new operating point + out += ctl_frame(SET_PWR, bytes([vtx.state.txagc])) + out += ctl_frame(SET_RATE, _cmd_rate()) + if out: + proc.stdin.write(out); proc.stdin.flush() threading.Thread(target=reader, daemon=True).start() import sys @@ -362,25 +496,36 @@ def reader(): want = (vtx.rz.cfg.discovery_channel if vtx.rz.state == rz.DISCOVERY else vtx.rz.op_channel if vtx.rz.state == rz.RC_OK else cur_chan) if want != cur_chan: - proc.stdin.write(ctl_frame(SET_CHAN, bytes([want, 0, 0]))) # 20 MHz - proc.stdin.flush() + _write(ctl_frame(SET_CHAN, bytes([want, 0, 0]))) # 20 MHz cur_chan = want prev_state = vtx.rz.state if act == rz.A_FAILSAFE: - proc.stdin.write(ctl_frame(SET_PWR, bytes([vtx.state.txagc]))) - proc.stdin.flush() + _write(ctl_frame(SET_PWR, bytes([vtx.state.txagc]))) if vid is not None and act in (rz.A_TX_VIDEO, rz.A_FAILSAFE): chunk = vid.read(1022) # room for the 2-byte app-seq header if not chunk: vid.seek(0); chunk = vid.read(1022) - proc.stdin.write(psdu_frame(app_seq.to_bytes(2, "little") + chunk)) - proc.stdin.flush() + psdu = psdu_frame(app_seq.to_bytes(2, "little") + chunk) + with wlock: + # The synthetic stream has no protected frame classes; an + # embedding app passes protected=True for control/IDR bodies. + cand = vtx.probe_mcs_for_seq(app_seq) + if cand is not None: + probe = rate_spec(vtx.state.mode, cand, vtx.state.bw) + buf = (ctl_frame(SET_RATE, probe.encode()) + psdu + + ctl_frame(SET_RATE, _cmd_rate())) + else: + buf = psdu + proc.stdin.write(buf) + proc.stdin.flush() app_seq = (app_seq + 1) & 0xFFF if now - last_log > 1000: last_log = now + probes = (f" probes={dict(sorted(vtx.probe_counters.items()))}" + if vtx.mcs_probe else "") sys.stderr.write(f"state={vtx.rz.state} txagc={vtx.state.txagc} " f"ov={vtx.state.overhead} ladder={vtx.state.ladder} " - f"failsafe={vtx.state.failsafe}\n") + f"failsafe={vtx.state.failsafe}{probes}\n") sys.stderr.flush() time.sleep(0.002) @@ -404,6 +549,17 @@ def main(): ap.add_argument("--feedback-ms", type=int, default=100, help="VRX RCF feedback period (ms). Higher = fewer half-duplex " "RX-blind windows on the ground (diagnostic knob)") + ap.add_argument("--mcs-probe", action="store_true", + help="adjacent-MCS probing: the VTX flies the scheduled " + "probe seqs on the neighbour rates, the VRX gates " + "promotions on the measured candidate delivery. Must " + "be set on BOTH roles (schedule = protocol invariant)") + ap.add_argument("--mcs-bias", type=_parse_mcs_bias, default=None, + metavar="M:DB,...", + help="VRX-only: shift the model's per-MCS snr_req (e.g. " + "'4:-6,5:-8'; negative = model optimistic) — the " + "deliberate miscalibration the probe A/B measures " + "against") a = ap.parse_args() if a.role == "selftest": @@ -419,9 +575,10 @@ def main(): calib = em.load_calibration(a.energy_calib) if a.role == "vrx": run_vrx(proc, link, calib, a.vtx_id, a.channel, - feedback_period_ms=a.feedback_ms) + feedback_period_ms=a.feedback_ms, + mcs_probe=a.mcs_probe, mcs_bias=a.mcs_bias) else: - run_vtx(proc, a.vtx_id, a.video, a.channel) + run_vtx(proc, a.vtx_id, a.video, a.channel, mcs_probe=a.mcs_probe) if __name__ == "__main__": diff --git a/tools/precoder/controller.py b/tools/precoder/controller.py index c632142e..b2fa68c4 100644 --- a/tools/precoder/controller.py +++ b/tools/precoder/controller.py @@ -19,6 +19,7 @@ from __future__ import annotations +import math from dataclasses import dataclass, field import energy_model as em @@ -74,6 +75,39 @@ class ControllerConfig: rung_block_delta: float = 0.15 rung_block_hold_ms: int = 5000 rung_min_samples: int = 8 + # Live adjacent-MCS probes (OPT-IN, default OFF -> behaviour unchanged): + # report_mcs_delivery() feeds measured candidate delivery (score. + # McsProbeWindow.stats()) and the evidence corrects the MODEL, it does not + # replace it. While the current point still works, a row with mcs ABOVE the + # current one is admissible only when its probe LOWER confidence bound + # clears the raw frame-delivery its FEC needs; a candidate whose UPPER + # bound cannot reach even the heaviest overhead's requirement is blocked + # outright for a hold. Emergency downgrades keep trusting the model — never + # wait for probes to escape a failing point. The raw-delivery requirement + # rises steeply as overhead falls (ov=0.10 needs ~0.99 per-frame), and a + # Wilson lower bound caps at n/(n+z^2), so the only promote path is + # mcs+1 at mid/heavy overhead first, then the ungated same-MCS overhead + # trim — measured caution, by construction. Sample-rate dependence: one + # up-probe per 64 frames, so min_samples=12 inside max_age needs sustained + # >=~100 fps injection traffic; thresholds are sized for the streaming + # (~250+ fps) use, not the selftest tick rate. + mcs_probe_enabled: bool = False + mcs_probe_min_samples: int = 12 + mcs_probe_window_samples: int = 64 + mcs_probe_max_age_ms: int = 8000 + mcs_probe_block_hold_ms: int = 5000 + mcs_probe_confidence: float = 0.90 + mcs_probe_margin: float = 0.0 + # The ACTIVE row's measured delivery (its samples are the main stream, so + # they arrive at full frame rate) may condemn the operating point the + # model still believes in — the SNR-blind failure modes (interference, a + # miscalibrated row the cold pick landed on) show up only here. Tripping + # is deliberately slower than a candidate block: the active evidence must + # stay condemned for this many CONSECUTIVE feedback reports, so a burst + # that briefly floods the fast-filling window rides through while a + # persistent lie cannot. Candidates need no such guard — their samples + # accrue at probe duty, so min_samples already spans seconds. + mcs_probe_active_trip_ticks: int = 3 class Controller: @@ -95,6 +129,10 @@ def __init__(self, link, calib, cfg: ControllerConfig | None = None): self._rung_block: dict[int, float] = {} # bw -> blocked-until (ms) self._now_ms: float = -1 << 60 # last update() time, for _best self.primary_dirty = False # all rungs bad incl. narrowest + self._mcs_block: dict[int, float] = {} # mcs -> blocked-until (ms) + self._mcs_stats: dict = {} # mcs -> score.ProbeStat + self._frame_req_cache: dict[float, float] = {} + self._active_trip = 0 # consecutive condemned reports # --- estimation -------------------------------------------------------- # def _path_loss(self, reported_snr: float, reported_txagc: int) -> float: @@ -126,10 +164,97 @@ def report_rung_delivery(self, stats: dict[int, tuple[float, int]], self.cfg.rung_block_delta for d in usable.values())) - def _best(self, path_loss: float, margin: float) -> OpPoint | None: + # --- adjacent-MCS probe evidence ---------------------------------------- # + def _mcs_blocked(self, mcs: int) -> bool: + until = self._mcs_block.get(mcs) + return until is not None and self._now_ms < until + + def _required_frame_delivery(self, overhead: float) -> float: + """Raw FCS-clean frame delivery this overhead's FEC needs to hit the + block-delivery target — the exact inverse of the plain (no-SBI) branch + of fec_ab_sim.sim_interframe, on the same generation geometry the link + rows were built from. Conservative on purpose: SBI salvage only makes + the real requirement laxer.""" + key = round(overhead, 3) + if key in self._frame_req_cache: + return self._frame_req_cache[key] + import link_model as _lm + frames = getattr(self.link, "frames_per_gen", 12) + n_sub = getattr(self.link, "n_sub", _lm.N_SUB) + total = frames * n_sub + src = max(1, min(total, round(total / (1.0 + overhead)))) + if src >= total: # no room for parity: unattainable + self._frame_req_cache[key] = req = 1.01 + return req + k_frames = math.ceil(src / n_sub) + + def block_ok(p: float) -> float: + return sum(math.comb(frames, i) * p ** i * (1 - p) ** (frames - i) + for i in range(k_frames, frames + 1)) + + lo, hi = 0.0, 1.0 + for _ in range(40): # bisect the smallest clearing p + mid = (lo + hi) / 2 + if block_ok(mid) >= self.cfg.target: + hi = mid + else: + lo = mid + self._frame_req_cache[key] = hi + return hi + + def report_mcs_delivery(self, stats: dict, now_ms: float) -> None: + """Feed per-MCS measured delivery (mcs -> score.ProbeStat: the adjacent + candidates from the probe schedule plus the ACTIVE row from the main + stream). An MCS whose UPPER confidence bound falls short of even the + heaviest overhead's raw frame-delivery requirement cannot meet the SLA + at any FEC — block it for a hold. The lowest rate is never blocked: it + is the fallback. Candidates block on first sufficient evidence; the + active row must stay condemned for `mcs_probe_active_trip_ticks` + consecutive reports (its full-rate window fills in fractions of a + second, so a single interference burst would otherwise read as a dead + rate). A tripped active row makes cur_ok fail on the next decide (the + measured fast-down the model cannot veto) and the block keeps _best + from re-picking the row the model still believes in.""" + self._mcs_stats = dict(stats) + req = self._required_frame_delivery(max(self.cfg.overhead_set)) + lowest = min(self.cfg.mcs_set) + active = self.cur.mcs if self.cur is not None else None + active_bad = False + for mcs, st in stats.items(): + if mcs == lowest: + continue + if st.attempts >= self.cfg.mcs_probe_min_samples and st.ucb < req: + if mcs == active: + active_bad = True + if self._active_trip + 1 < self.cfg.mcs_probe_active_trip_ticks: + continue + self._mcs_block[mcs] = now_ms + self.cfg.mcs_probe_block_hold_ms + self._active_trip = self._active_trip + 1 if active_bad else 0 + + def _mcs_gate_ok(self, r) -> bool: + """Probe gate for one row (only rows ABOVE the current MCS): admissible + when fresh probe evidence's LOWER bound clears the row's raw + frame-delivery requirement. Probes fly at the commanded bandwidth, so + evidence does not transfer to another bw — those rows wait for the + stepwise path (bw move at same MCS, then MCS with fresh evidence).""" + if not self.cfg.mcs_probe_enabled or self.cur is None \ + or r.mcs <= self.cur.mcs: + return True + if r.bw != self.cur.bw: + return False + st = self._mcs_stats.get(r.mcs) + if st is None or st.attempts < self.cfg.mcs_probe_min_samples: + return False + return st.lcb >= (self._required_frame_delivery(r.overhead) + + self.cfg.mcs_probe_margin) + + def _best(self, path_loss: float, margin: float, + gate_up: bool = False) -> OpPoint | None: best = None for r in self.rows: - if self._rung_blocked(r.bw): + if self._rung_blocked(r.bw) or self._mcs_blocked(r.mcs): + continue + if gate_up and not self._mcs_gate_ok(r): continue op = resolve(r, path_loss, self.calib, self.link, self.cfg.payload_bytes, self.cfg.src_bitrate_bps, margin) @@ -176,11 +301,18 @@ def _decide(self, path_loss: float, now_ms: float) -> OpPoint | None: self.cur.snr_req), path_loss, self.calib, self.link, self.cfg.payload_bytes, self.cfg.src_bitrate_bps, 0.0) cur_ok = (cur_now is not None and cur_now.p_deliver >= self.cfg.target - and not self._rung_blocked(self.cur.bw)) + and not self._rung_blocked(self.cur.bw) + and not self._mcs_blocked(self.cur.mcs)) if cur_ok: self.cur = cur_now # refresh txagc/e_bit at new path loss - cand = self._best(path_loss, self.cfg.margin_db) # candidate with hysteresis + # Candidate with hysteresis. The probe gate rides only while the current + # point still works — an emergency downgrade must take the model's word + # (waiting for probe samples to escape a failing point would be + # backwards), and the cold start (cur None) is model-trusted: a bad + # first pick self-corrects through the existing fast-down, after which + # the probes prevent re-entry. + cand = self._best(path_loss, self.cfg.margin_db, gate_up=cur_ok) if cand is None: # nothing clears even with margin: keep current if it still works, @@ -224,10 +356,29 @@ def on_tick(self, now_ms: float) -> OpPoint | None: self.cur = self._failsafe_point() return self.cur + def reset_operating_point(self) -> None: + """Forget the committed point — the next update makes a fresh, + model-trusted pick — while KEEPING the measured MCS blocks: a + reacquisition cold pick must not walk straight back into a row whose + measured delivery just condemned it. The VRX calls this when it loses + the video entirely (an operating point it cannot hear is not worth + keeping, and its own estimator is starved of the evidence to fix it).""" + self.cur = None + self.shed = False + self._active_trip = 0 + def _failsafe_point(self) -> OpPoint: return MAX_RANGE def _commit(self, op: OpPoint, now_ms: float) -> OpPoint: + # Probe evidence is bound to the (mode, bw) it was measured in — a + # committed move to another context invalidates blocks and stats (the + # VRX-side window resets itself on the same transition). + if self.cur is None or (op.mode, op.bw) != (self.cur.mode, self.cur.bw): + self._mcs_block.clear() + self._mcs_stats = {} + if self.cur is None or op.mcs != self.cur.mcs: + self._active_trip = 0 self.cur = op self._last_change_ms = now_ms return op @@ -257,6 +408,41 @@ def _fade_headroom(self, op: OpPoint | None, path_loss: float) -> OpPoint | None op.snr_req, eb, pdel) +class BiasedLink: + """A link model whose per-MCS belief is shifted by `bias_db` dB (NEGATIVE = + the model believes the rate works at that much less SNR than it does). + The bias runs through BOTH faces of the model — `snr_required` (row + thresholds) and `p_deliver` (the SLA check in resolve/cur_ok) — because a + real miscalibration is a wrong belief, not just a wrong table entry: with + only the rows shifted, resolve()'s own p_deliver veto would silently + un-bias the controller.""" + + def __init__(self, base, bias_db: dict[int, float]): + self.base = base + self.bias = {int(m): float(db) for m, db in bias_db.items()} + self.frames_per_gen = getattr(base, "frames_per_gen", 12) + + def p_deliver(self, snr_db, mcs, overhead, sbi=True): + return self.base.p_deliver(snr_db - self.bias.get(mcs, 0.0), mcs, + overhead, sbi) + + def snr_required(self, mcs, overhead, target, sbi=True, **kw): + return (self.base.snr_required(mcs, overhead, target, sbi=sbi, **kw) + + self.bias.get(mcs, 0.0)) + + +def apply_model_bias(ctrl: Controller, bias_db: dict[int, float]) -> None: + """Deliberately miscalibrate a controller's model per MCS (dB; negative = + optimistic) and rebuild its rows from the biased belief. The lever the + probe A/B harnesses measure against — the probes' whole job is to catch + exactly this.""" + ctrl.link = BiasedLink(ctrl.link, bias_db) + ctrl.rows = build_link_rows(ctrl.link, ctrl.cfg.target, ctrl.cfg.mcs_set, + ctrl.cfg.overhead_set, ctrl.cfg.bw, + sbi=ctrl.cfg.sbi, bw_set=ctrl.cfg.bw_set, + mode=ctrl.cfg.mode) + + # --------------------------------------------------------------------------- # # SVC per-layer UEP: a bank of controllers, one per temporal layer. Base/IDR # layers hold a high delivery target and never shed; enhancement layers carry a diff --git a/tools/precoder/mcs_probe_ab_sim.py b/tools/precoder/mcs_probe_ab_sim.py new file mode 100644 index 00000000..c683514a --- /dev/null +++ b/tools/precoder/mcs_probe_ab_sim.py @@ -0,0 +1,271 @@ +"""A/B sim: model-only vs model+adjacent-MCS-probes under a miscalibrated model. + +The controller's whole failure mode is a WRONG BELIEF: its link model says a +rate clears the SLA when the real channel disagrees. This sim splits the two — +the controller runs on a deliberately biased model (`controller.apply_model_ +bias`, the same lever the on-air harness uses) while frame delivery comes from +the untouched nominal channel — and drives the full VTX<->VRX loop per frame +(~4 ms, so the probe-window arithmetic is realistic, unlike selftest's 100 ms +tick). Both arms see identical channel randomness (paired seeds). + +What the arms show: + * model-only: promotes into the optimistically-biased row and STAYS there — + its cur_ok check believes the same wrong model, so nothing pulls it back + (post-FEC delivery collapses while the model reads healthy); + * probes on: the candidate's measured delivery never clears the promote + gate's lower confidence bound, so the biased row is never entered (and its + upper bound eventually blocks it outright). + +Probe COST is priced with no special-casing: probes are real frames flying a +different rate, so their airtime, PA energy, and losses land in the same +energy / delivered-source-bit ledger as everything else. The honest caveat the +clean scenario quantifies: probes measure candidates AT THE INCUMBENT'S TXAGC, +so with probing on a promotion needs the faster rate to already work at the +current power — steady-state upward moves get rarer (delivery holds, energy +may idle slightly above the model-only optimum). Run `python mcs_probe_ab_sim +.py` for the report; test_mcs_probe_ab_sim.py pins the headline. +""" + +from __future__ import annotations + +import math +import random +from dataclasses import dataclass + +import energy_model as em +import link_model as lm +import rc_proto as rp +import rendezvous as rz +from adaptive_link import AdaptiveVrx, AdaptiveVtx +from controller import ControllerConfig, apply_model_bias + +VTX_ID = 0xABCD +PAYLOAD = 1024 +DT_MS = 4.0 # ~250 fps injection traffic +FEEDBACK_MS = 100 +GEN_FRAMES = 12 # FEC generation geometry (link_model default) +N_SUB = 10 + + +# --------------------------------------------------------------------------- # +# Scenarios: path loss (free SNR at txagc 0) per frame index. Every scenario +# opens with a 4 s deep-range ACQUISITION segment — a real session starts at +# MAX_RANGE/discovery distance, and the controller's cold pick is deliberately +# model-trusted (ungated), so opening at close range would walk BOTH arms +# straight into an optimistically-biased row before a single probe has flown. +# --------------------------------------------------------------------------- # +ACQ_FRAMES = 1000 +ACQ_PL = -16.0 + + +def sc_static(i: int) -> float: + return 2.0 + + +def sc_triangle(i: int, n: int = 14000) -> float: + """Fly out and back: near -> far -> near (post-acquisition).""" + half = n // 2 + return 8.0 - 22.0 * (i if i < half else n - i) / half + + +def make_sc_fade(seed: int, base: float = 4.0, rho: float = 0.995, + sigma: float = 0.3): + """AR(1) time-correlated fading (~0.8 s correlation, ~3 dB std).""" + rng = random.Random(seed ^ 0xFADE) + state = {"x": 0.0, "i": -1} + + def f(i: int) -> float: + while state["i"] < i: # advance once per frame + state["x"] = rho * state["x"] + rng.gauss(0.0, sigma) + state["i"] += 1 + return base + state["x"] + return f + + +def sc_burst(i: int) -> float: + """Steady link with a 160 ms deep interference slam every 5 s (past the + post-downgrade hold, so recovery climbs are part of the picture).""" + return 4.0 - (20.0 if (i % 1250) < 40 else 0.0) + + +SCENARIOS = { + "static": sc_static, + "triangle": sc_triangle, + "fade": None, # built per seed + "burst": sc_burst, +} + + +def with_acquisition(pl_fn): + def f(i: int) -> float: + return ACQ_PL if i < ACQ_FRAMES else pl_fn(i - ACQ_FRAMES) + return f + + +@dataclass +class ArmResult: + delivery: float # post-FEC generation delivery (warm period) + e_bit_nj: float # realized nJ per delivered source bit + commits: int # (mode, mcs, bw, overhead) changes (churn) + biased_row_ms: float | None # first time the biased MCS was flown as the op + biased_frames: int # TX frames spent commanded to the biased MCS + first_block_ms: float | None # first controller block on the biased MCS + probe_frames: int + probe_lost: int + probe_airtime_frac: float + settle_mcs: int + + +def run_arm(pl_fn, n_frames: int, probe: bool, bias: dict[int, float] | None, + seed: int, trials: int = 300) -> ArmResult: + truth = lm.LinkModel(trials=trials) # the channel: nominal + model = lm.LinkModel(trials=trials) # the controller's belief + calib = em.load_calibration() + # sbi=False: the frame-level Bernoulli channel below has no sub-block + # salvage, so the controller's belief must price FEC the same way — a + # mixed-semantics pair would misread as miscalibration in BOTH arms. + vrx = AdaptiveVrx(model, calib, VTX_ID, + ControllerConfig(target=0.99, allow_shed=False, sbi=False, + mcs_probe_enabled=probe)) + if bias: + apply_model_bias(vrx.ctrl, bias) + vtx = AdaptiveVtx(VTX_ID, mcs_probe=probe) + rng = random.Random(seed) + biased_mcs = max(bias) if bias else None + + energy_j = 0.0 + delivered_bits = 0.0 + airtime_s = probe_air_s = 0.0 + probe_frames = probe_lost = 0 + commits = 0 + prev_key = None + biased_row_ms = first_block_ms = None + biased_frames = 0 + settle_mcs = 0 + gen_ok: list[bool] = [] + gen_ov = 0.25 + gens_total = gens_ok = 0 + warm = n_frames // 10 + seq = 0 + + for i in range(n_frames): + now = i * DT_MS + act = vtx.step(now) + energy_j += calib.p_baseline_w * DT_MS / 1000.0 # floor: always on + if act in (rz.A_TX_VIDEO, rz.A_FAILSAFE): + cand = vtx.probe_mcs_for_seq(seq) + sel, mode = vtx.state.mcs, vtx.state.mode + flown = sel if cand is None else cand + txagc = vtx.state.txagc + recv = pl_fn(i) + calib.gain_db(txagc) + ok = rng.random() < 1.0 - truth.channel(flown, recv)["corrupt_rate"] + + t_on = calib.t_pre_us * 1e-6 + 8.0 * PAYLOAD / ( + em.phy_rate_mbps(mode, flown, vtx.state.bw) * 1e6) + energy_j += calib.pa_w(txagc) * t_on + airtime_s += t_on + if cand is not None: + probe_frames += 1 + probe_air_s += t_on + probe_lost += 0 if ok else 1 + + if ok: + vrx.on_video(rssi=-52.0, snr=recv, crc_err=False, seq=seq, + now_ms=now, rate=(mode, flown)) + # FEC generation accounting at the overhead the generation started + if not gen_ok: + gen_ov = vtx.state.overhead + gen_ok.append(ok) + if len(gen_ok) == GEN_FRAMES: + total = GEN_FRAMES * N_SUB + src = max(1, min(total, round(total / (1.0 + gen_ov)))) + need = math.ceil(src / N_SUB) + if i >= warm: + gens_total += 1 + if sum(gen_ok) >= need: + gens_ok += 1 + delivered_bits += 8.0 * PAYLOAD * GEN_FRAMES / (1.0 + gen_ov) + gen_ok = [] + seq = (seq + 1) & 0xFFF + + fb = vrx.step(now) + if fb is not None: + out = vtx.on_rc_frame(fb, now) + if out is not None: + vrx.on_disc_ack(out, now) + + key = (vtx.state.mode, vtx.state.mcs, vtx.state.bw, vtx.state.overhead) + if key != prev_key: + if prev_key is not None and i >= warm: + commits += 1 + prev_key = key + if i == warm: + settle_mcs = vtx.state.mcs + if biased_mcs is not None: + if vtx.state.mcs == biased_mcs and not vtx.state.failsafe: + biased_frames += 1 + if biased_row_ms is None: + biased_row_ms = now + if first_block_ms is None and vrx.ctrl._mcs_blocked(biased_mcs): + first_block_ms = now + + return ArmResult( + delivery=gens_ok / gens_total if gens_total else 0.0, + e_bit_nj=(energy_j / delivered_bits * 1e9) if delivered_bits else + float("inf"), + commits=commits, + biased_row_ms=biased_row_ms, + biased_frames=biased_frames, + first_block_ms=first_block_ms, + probe_frames=probe_frames, + probe_lost=probe_lost, + probe_airtime_frac=probe_air_s / airtime_s if airtime_s else 0.0, + settle_mcs=settle_mcs) + + +def run_scenario(name: str, n_frames: int = 15000, bias: dict | None = None, + seed: int = 1, trials: int = 300): + """`ambush` = the static scenario WITHOUT the acquisition segment: the + session opens at close range and the deliberately-ungated cold pick lands + straight on the biased row — the case the measured active-row escape + exists for.""" + arms = {} + for probe in (False, True): + if name == "fade": # paired fading realization + f = make_sc_fade(seed) + elif name == "triangle": + f = lambda i: sc_triangle(i, n_frames - ACQ_FRAMES) # noqa: E731 + else: + f = SCENARIOS["static" if name == "ambush" else name] + arms["probe" if probe else "model"] = run_arm( + f if name == "ambush" else with_acquisition(f), + n_frames, probe, bias, seed, trials) + return arms + + +def main(): + bias = {5: -8.0} # the model believes MCS5 needs 8 dB less than it does + print(f"MCS-probe A/B — {DT_MS:g} ms frames, model bias {bias} " + f"(clean scenario runs unbiased)\n") + hdr = (f"{'scenario':<10} {'arm':<6} {'deliv':>6} {'nJ/bit':>8} " + f"{'settle':>6} {'commits':>7} {'flew-bad':>9} {'bad-frm':>7} " + f"{'blocked':>8} {'probe%':>7} {'p-lost':>6}") + print(hdr) + print("-" * len(hdr)) + for name in ("static", "ambush", "triangle", "fade", "burst"): + for label, b in (("", bias), ("-clean", None)): + if label == "-clean" and name != "static": + continue # one unbiased control is enough + arms = run_scenario(name, bias=b) + for arm, r in arms.items(): + fb = "-" if r.biased_row_ms is None else f"{r.biased_row_ms/1000:.1f}s" + bl = "-" if r.first_block_ms is None else f"{r.first_block_ms/1000:.1f}s" + print(f"{name + label:<10} {arm:<6} {r.delivery:>6.3f} " + f"{r.e_bit_nj:>8.2f} {'MCS' + str(r.settle_mcs):>6} " + f"{r.commits:>7d} {fb:>9} {r.biased_frames:>7d} " + f"{bl:>8} {r.probe_airtime_frac:>6.1%} {r.probe_lost:>6d}") + print() + + +if __name__ == "__main__": + main() diff --git a/tools/precoder/pyproject.toml b/tools/precoder/pyproject.toml index 35ace93d..58632270 100644 --- a/tools/precoder/pyproject.toml +++ b/tools/precoder/pyproject.toml @@ -44,5 +44,6 @@ testpaths = ["test_pipeline.py", "test_stream.py", "test_stream_fec.py", "test_controller.py", "test_rc_proto.py", "test_score.py", "test_rendezvous.py", "test_adaptive_link.py", + "test_mcs_probe_ab_sim.py", "test_svc_pipeline.py", "test_hop_diversity_sim.py", "test_hop_rx_combine.py"] diff --git a/tools/precoder/rc_proto.py b/tools/precoder/rc_proto.py index 03f13371..392501bd 100644 --- a/tools/precoder/rc_proto.py +++ b/tools/precoder/rc_proto.py @@ -88,6 +88,47 @@ def probe_bw(seq: int, bw_set) -> int | None: return None +# --------------------------------------------------------------------------- # +# MCS probe schedule — same protocol-invariant shape as probe_bw: both ends +# derive each seq's role from the seq alone, so measuring the delivery of the +# ADJACENT rates needs no wire fields. Slot 4 of each 64-seq cycle flies the +# next rate ABOVE the selected MCS, slot 20 the next BELOW (adjacent within the +# config-agreed mcs_set); everything else rides the commanded rate. 2/64 ≈ 3% +# duty. The slots are disjoint from the bandwidth-probe slots ({0,8,16} mod 32) +# so every probe observation has exactly one changed variable, and the period +# divides 4096 so the 12-bit app-seq wrap is phase-safe. +# +# PROTOCOL INVARIANT (versioned): period, slots, mcs_set and the selected-MCS +# derivation (the commanded profile) must be identical on VTX and VRX — like +# bw_set and PROFILE_TABLE_VERSION. The VRX additionally rate-verifies received +# probes and epoch-gates gap attribution (score.McsProbeWindow), so a one-sided +# enablement degrades to inert stats rather than mis-evidence. +# --------------------------------------------------------------------------- # +MCS_PROBE_PERIOD = 64 +_MCS_PROBE_SLOTS = (4, 20) # (up, down) +MCS_PROBE_SCHED_VER = 1 + + +def probe_mcs(seq: int, selected_mcs: int, mcs_set, + period: int = MCS_PROBE_PERIOD, + slots=_MCS_PROBE_SLOTS) -> int | None: + """Candidate MCS this video seq must fly at as a rate probe, else None. + + Returns the adjacent-in-`mcs_set` rate above (slot `slots[0]`) or below + (slot `slots[1]`) `selected_mcs`; None off-slot, at the set boundaries, or + when `selected_mcs` is not in the set (nothing adjacent is well-defined).""" + rates = sorted(mcs_set) + if selected_mcs not in rates: + return None + i = rates.index(selected_mcs) + slot = seq % period + if slot == slots[0]: + return rates[i + 1] if i + 1 < len(rates) else None + if slot == slots[1]: + return rates[i - 1] if i > 0 else None + return None + + def _crc(buf: bytes) -> int: return fec_subblock.crc16_ccitt(buf) diff --git a/tools/precoder/score.py b/tools/precoder/score.py index 36cde206..3c65f7a8 100644 --- a/tools/precoder/score.py +++ b/tools/precoder/score.py @@ -10,6 +10,8 @@ from __future__ import annotations +import math +import threading from collections import deque from dataclasses import dataclass, field @@ -147,3 +149,191 @@ def stats(self) -> dict[int, tuple[float, int]]: if h: out[bw] = (sum(h) / len(h), len(h)) return out + + +# One-sided normal quantiles for the Wilson bounds (each bound is used as its +# own one-sided test: promote on the LOWER bound, block on the UPPER). +_Z_ONE_SIDED = {0.90: 1.2816, 0.95: 1.6449, 0.99: 2.3263} + + +def wilson_bounds(successes: int, n: int, z: float) -> tuple[float, float]: + """Wilson-score interval for a binomial proportion — (lcb, ucb). + + Unlike the bare mean, the bounds stay honest at small n: a perfect record + caps at lcb = n/(n+z²), so "0 failures in 12" cannot read as proof of 99% + delivery. (0, n) -> (0, 1) when there is no evidence at all.""" + if n <= 0: + return 0.0, 1.0 + ph = successes / n + z2 = z * z + denom = 1.0 + z2 / n + center = ph + z2 / (2 * n) + half = z * math.sqrt(ph * (1.0 - ph) / n + z2 / (4.0 * n * n)) + return (max(0.0, (center - half) / denom), + min(1.0, (center + half) / denom)) + + +@dataclass(frozen=True) +class ProbeStat: + """Per-candidate-MCS probe evidence: raw FCS-clean frame delivery with + Wilson confidence bounds and the staleness of the newest sample.""" + delivery: float + successes: int + attempts: int + lcb: float + ucb: float + age_ms: float + + +class McsProbeWindow: + """Per-MCS raw delivery from the scheduled rate probes AND the active row. + + The rc_proto.probe_mcs schedule names which seqs fly the adjacent rates, so + like RungWindow no wire fields are needed — but rate evidence is regime- + dependent (the candidates move with the commanded profile), so attribution + is guarded three ways where RungWindow needs none: + + * successes are RATE-VERIFIED: a received probe-slot seq counts only when + the PHY-decoded (mode, mcs) matches the expected candidate — a frame + that demonstrably flew elsewhere (command lag, suppressed probe, spec + parse fallback) is ignored, never mis-credited; + * gap losses are EPOCH-GATED: attributed only while a non-probe frame has + confirmed the VTX is flying the commanded rate (and a mismatch + un-confirms), so losses around a profile transition or a failsafe + overlap are dropped rather than blamed on the wrong candidate; + * crc_err frames never advance the gap walk and never attribute BY SEQ — + their body is untrusted bits — but their PHY-decoded rate is from the + descriptor (pre-FCS), so a corrupt frame at the selected or an + adjacent-candidate rate IS a rate-verified failure of that rate and is + pushed seq-free. This is what keeps the evidence flowing in the + crc-flooded regimes where escape speed matters most; the same lost + frame may additionally be counted by the next clean frame's gap walk, + an accepted over-weighting of visible corruption in the conservative + direction. + + The SELECTED row is attributed by the same rules from every non-probe seq + (bandwidth-probe slots excluded when `bw_set` is given — those fly the + selected rate at another bandwidth, so their loss is bandwidth evidence, + not rate evidence). That makes the active row's measured delivery free — + the main stream is its probe — and it is what lets the controller escape + an operating point its model wrongly believes in: SNR-blind failure + (interference, a miscalibrated row) shows up here when no model does. + + set_selected() (any commanded (mode, mcs, bw) change) resets the window — + evidence from another operating context must not authorize a promotion. + Thread-safe: add_seq runs on the RX reader thread while stats()/ + set_selected() run on the control loop. + """ + + def __init__(self, mcs_set, mode: str = "ht", samples_per_mcs: int = 64, + max_age_ms: float = 8000.0, confidence: float = 0.90, + period: int | None = None, slots=None, bw_set=None): + import rc_proto as _rp + self._rp = _rp + self.mcs_set = tuple(sorted(mcs_set)) + self._period = period if period is not None else _rp.MCS_PROBE_PERIOD + self._slots = tuple(slots) if slots is not None else _rp._MCS_PROBE_SLOTS + self._bw_set = tuple(sorted(bw_set)) if bw_set else None + self._maxlen = samples_per_mcs + self.max_age_ms = max_age_ms + self._z = _Z_ONE_SIDED.get(round(confidence, 2), 1.2816) + self._hist: dict[int, deque] = {} # mcs -> deque[(t_ms, ok)] + self._mode = mode + self._selected: int | None = None + self._epoch_confirmed = False + self._last_seq: int | None = None + self._lock = threading.Lock() + + def _probe(self, seq: int) -> int | None: + return self._rp.probe_mcs(seq, self._selected, self.mcs_set, + self._period, self._slots) + + def _slot_mcs(self, seq: int) -> int | None: + """The MCS this seq is scheduled to fly, or None for a seq that is no + rate evidence at all (a bandwidth-probe slot).""" + cand = self._probe(seq) + if cand is not None: + return cand + if self._bw_set is not None and \ + self._rp.probe_bw(seq, self._bw_set) is not None: + return None + return self._selected + + def _tracked(self) -> tuple: + """The rates evidence is currently collected for: the selected row and + its adjacent-in-set candidates.""" + rates = self.mcs_set + if self._selected not in rates: + return (self._selected,) + i = rates.index(self._selected) + out = [self._selected] + if i + 1 < len(rates): + out.append(rates[i + 1]) + if i > 0: + out.append(rates[i - 1]) + return tuple(out) + + def set_selected(self, mode: str, mcs: int, now_ms: float) -> None: + """Commanded operating point changed: blank the window (new epoch).""" + with self._lock: + self._mode = mode + self._selected = mcs + self._hist.clear() + self._epoch_confirmed = False + self._last_seq = None + + def _push(self, mcs: int, now_ms: float, ok: bool) -> None: + h = self._hist.get(mcs) + if h is None: + h = self._hist[mcs] = deque(maxlen=self._maxlen) + h.append((now_ms, ok)) + + def add_seq(self, seq: int, now_ms: float, rate: tuple | None = None, + crc_err: bool = False) -> None: + """One received video seq. `rate` is the PHY-decoded (mode, mcs) of the + frame (devourer_events.desc_rate_to_mcs), or None when unavailable — + rate-less frames are dropped (fail-inert: no admission, no blocking).""" + with self._lock: + if self._selected is None: + return + if crc_err: + # the descriptor rate survives body corruption: a corrupt + # frame at a rate we track is that rate's measured failure + if rate is not None and rate[0] == self._mode and \ + rate[1] in self._tracked(): + self._push(rate[1], now_ms, False) + return + if self._last_seq is not None and self._epoch_confirmed: + gap = (seq - self._last_seq) % 4096 + # walk the missing seqs (bounded like RungWindow: a monster gap + # is a link outage, not per-candidate information) + for d in range(1, min(gap, 128)): + m = self._slot_mcs((self._last_seq + d) % 4096) + if m is not None: + self._push(m, now_ms, False) + cand = self._probe(seq) + if cand is None: + if rate is not None: + self._epoch_confirmed = (rate == (self._mode, self._selected)) + if self._epoch_confirmed and self._slot_mcs(seq) is not None: + self._push(self._selected, now_ms, True) + elif rate is not None and rate == (self._mode, cand): + self._push(cand, now_ms, True) + self._last_seq = seq + + def stats(self, now_ms: float) -> dict[int, ProbeStat]: + """mcs -> ProbeStat for candidates with any fresh samples.""" + out = {} + with self._lock: + for mcs, h in self._hist.items(): + while h and h[0][0] < now_ms - self.max_age_ms: + h.popleft() + if not h: + continue + n = len(h) + s = sum(1 for _, ok in h if ok) + lcb, ucb = wilson_bounds(s, n, self._z) + out[mcs] = ProbeStat(delivery=s / n, successes=s, attempts=n, + lcb=lcb, ucb=ucb, + age_ms=now_ms - h[-1][0]) + return out diff --git a/tools/precoder/test_adaptive_link.py b/tools/precoder/test_adaptive_link.py index c289e828..40ef7de1 100644 --- a/tools/precoder/test_adaptive_link.py +++ b/tools/precoder/test_adaptive_link.py @@ -93,6 +93,130 @@ def test_vtx_decodes_v2_profile_to_vht_ladder_and_probes(): assert "CRIT=MCS3/20" in vtx.state.ladder +def test_rate_spec_ht_and_vht(): + assert al.rate_spec("ht", 4, 20) == "MCS4/20" + assert al.rate_spec("vht", 3, 80) == "VHT1SS_MCS3/80" + + +def test_vtx_tracks_commanded_rate_fields(): + """TxState carries the commanded (mode, mcs, bw) as fields — the SET_RATE + plumbing and probe schedule read these, never re-parse the ladder string + (which broke on VHT ladders).""" + vtx = al.AdaptiveVtx(0xABCD) + vtx.on_rc_frame(rp.pack_rcf(rp.Rcf(vtx_id=0xABCD, + profile=rp.encode_profile("vht", 2, 80))), 0) + assert (vtx.state.mode, vtx.state.mcs, vtx.state.bw) == ("vht", 2, 80) + assert al.rate_spec(vtx.state.mode, vtx.state.mcs, vtx.state.bw) == \ + "VHT1SS_MCS2/80" + + +def test_vtx_mcs_probe_schedule_and_suppression(): + """probe_mcs_for_seq flies the shared schedule only in the steady TX-video + state; up-probes never ride protected frames, an explicit suspend, or + failsafe; counters expose the attempted probes per candidate.""" + vtx = al.AdaptiveVtx(0xABCD, rz.VtxConfig(vtx_id=0xABCD, fallback_ms=100), + mcs_probe=True) + # before any feedback/step: not in TX-video -> no probes + assert vtx.probe_mcs_for_seq(4) is None + vtx.on_rc_frame(rp.pack_rcf(rp.Rcf(vtx_id=0xABCD, + profile=rp.encode_profile("ht", 4, 20))), 0) + assert vtx.step(10) == rz.A_TX_VIDEO + assert vtx.probe_mcs_for_seq(4) == 5 # up slot + assert vtx.probe_mcs_for_seq(20) == 3 # down slot + assert vtx.probe_mcs_for_seq(5) is None # off-slot + # protected frames never carry the speculative up-probe; down stays allowed + assert vtx.probe_mcs_for_seq(4 + 64, protected=True) is None + assert vtx.probe_mcs_for_seq(20 + 64, protected=True) == 3 + # explicit back-off hook (thermal/congestion): same asymmetry + vtx.suspend_upward_probes(True) + assert vtx.probe_mcs_for_seq(4 + 128) is None + assert vtx.probe_mcs_for_seq(20 + 128) == 3 + vtx.suspend_upward_probes(False) + assert vtx.probe_counters == {5: 1, 3: 3} + # failsafe clamps probing entirely (and the failsafe rate fields match the + # failsafe ladder, not the last commanded profile) + assert vtx.step(500) == rz.A_FAILSAFE + assert (vtx.state.mode, vtx.state.mcs, vtx.state.bw) == ("ht", 0, 20) + assert vtx.probe_mcs_for_seq(4 + 192) is None + assert vtx.probe_mcs_for_seq(20 + 192) is None + + +def test_vrx_mcs_probe_stats_feed_controller_and_block(): + """Closed VRX-side loop with probing on: up-probe seqs go missing (their + rate is too fragile on the real channel) -> the estimator attributes the + losses to the candidate, and the controller blocks it; a profile change + then blanks the evidence (context-bound).""" + import energy_model as em + import link_model as lm + from controller import ControllerConfig + link, calib = lm.LinkModel(trials=300), em.load_calibration() + vrx = al.AdaptiveVrx(link, calib, 0xABCD, + ControllerConfig(target=0.99, allow_shed=False, + mcs_probe_enabled=True)) + state = {"seq": 0} + + def run(n_frames, t0, pl, lose_up=False): + t = t0 + for _ in range(n_frames): + seq = state["seq"] + state["seq"] = (seq + 1) & 0xFFF # continuous across phases + t += 4.0 + sel = vrx.cur_op.mcs + # honest feedback: the reported SNR follows the applied power + snr = pl + calib.gain_db(vrx._cur_txagc) + cand = rp.probe_mcs(seq, sel, vrx.ctrl.cfg.mcs_set) + flown = sel if cand is None else cand + if lose_up and cand is not None and cand > sel: + pass # lost on air -> a gap + else: + vrx.on_video(rssi=-52.0, snr=snr, crc_err=False, seq=seq, + now_ms=t, rate=("ht", flown)) + vrx.step(t) + return t + + t = run(400, t0=0, pl=0.0) # settle the operating point + sel = vrx.cur_op.mcs + assert 0 < sel < 7 + t = run(1400, t0=t, pl=0.0, lose_up=True) + assert vrx.cur_op.mcs == sel # blocked/ungated: no promotion + st = vrx.mcs_win.stats(t) + # a couple of pre-loss successes may linger inside max_age; the verdict + # lives in the confidence bound, not the point estimate + assert st[sel + 1].attempts >= 12 and st[sel + 1].ucb < 0.5 + assert st[sel - 1].delivery == 1.0 # down-probes keep arriving + assert vrx.ctrl._mcs_blocked(sel + 1) # empirically bad: blocked + # a commanded-rate change resets the evidence window + vrx.mcs_win.set_selected("ht", sel - 1, t) + assert vrx.mcs_win.stats(t) == {} + + +def test_vrx_video_loss_resets_to_max_range(): + """Total video silence -> the VRX drops its command to MAX_RANGE and + forgets the committed point (an operating point it cannot hear starves the + very estimator that could condemn it — the frozen-command deadlock), while + the measured MCS blocks survive so reacquisition cannot re-pick a + condemned row. RCFs stopping is also what lets the VTX-side failsafe fire.""" + import energy_model as em + import link_model as lm + import op_table + from controller import ControllerConfig + link, calib = lm.LinkModel(trials=300), em.load_calibration() + vrx = al.AdaptiveVrx(link, calib, 0xABCD, + ControllerConfig(target=0.99, allow_shed=False, + mcs_probe_enabled=True)) + for i in range(30): + vrx.on_video(rssi=-50.0, snr=25.0, crc_err=False, seq=i, now_ms=i * 4.0, + rate=("ht", vrx.cur_op.mcs)) + vrx.step(i * 4.0) + assert vrx.cur_op.mcs > 0 # settled on a real rate + vrx.ctrl._mcs_block[7] = 1 << 40 # a held measured block + fb = vrx.step(5000.0) # > link_lost_ms of silence + assert fb is not None and rp.frame_type(fb) == rp.T_DISC + assert vrx.cur_op is op_table.MAX_RANGE + assert vrx.ctrl.cur is None + assert 7 in vrx.ctrl._mcs_block # measured evidence survives + + def test_vrx_senses_rungs_and_flags_primary_dirty(): """Closed VRX-side loop: video seqs arrive with the 80-rung probes missing -> the controller vacates 80; then ALL probe rungs go missing while video diff --git a/tools/precoder/test_controller.py b/tools/precoder/test_controller.py index f8fd4a8d..b07b3ed1 100644 --- a/tools/precoder/test_controller.py +++ b/tools/precoder/test_controller.py @@ -217,3 +217,164 @@ def test_primary_dirty_fires_only_when_narrowest_rung_bad_too(): # too few samples -> no verdict change c.report_rung_delivery({20: (0.0, 2)}, now_ms=200) assert not c.primary_dirty + + +# --------------------------------------------------------------------------- # +# Adjacent-MCS probe evidence: the gate/block on the model +# --------------------------------------------------------------------------- # +def _stat(lcb, ucb, attempts, delivery=None): + from score import ProbeStat + return ProbeStat(delivery=lcb if delivery is None else delivery, + successes=int(attempts * (delivery or lcb)), + attempts=attempts, lcb=lcb, ucb=ucb, age_ms=50.0) + + +_SHARED_LINK = lm.LinkModel(trials=400) # shared p_deliver cache across cases + + +def _fast_ctrl(**kw): + return Controller(_SHARED_LINK, em.load_calibration(), ControllerConfig(**kw)) + + +def _settle_pl(c, calib, pl, t0, ticks=8, txagc=32, dt=200): + """Drive at a constant path loss with honest feedback (reported SNR follows + the applied TXAGC), keeping the time base continuous across phases.""" + op = None + for i in range(ticks): + op = c.update(pl + calib.gain_db(txagc), txagc, now_ms=t0 + i * dt) + if op is not None: + txagc = op.txagc + return op, t0 + ticks * dt, txagc + + +def test_required_frame_delivery_mirrors_sim_interframe(): + """The gate's raw-frame requirement is the exact inverse of the plain + (no-SBI) FEC math the link rows were built from: at the requirement the + binomial block delivery crosses the target.""" + import math as m + import fec_ab_sim + c = _fast_ctrl(target=0.99, allow_shed=False) + # generation geometry 12x10: heavy FEC needs 6/12 clean frames, light 11/12 + req_heavy = c._required_frame_delivery(1.00) + req_light = c._required_frame_delivery(0.10) + assert 0.70 < req_heavy < 0.85 + assert 0.97 < req_light < 1.0 + assert req_heavy < c._required_frame_delivery(0.50) < \ + c._required_frame_delivery(0.25) < req_light + + def tail(p, k): + return sum(m.comb(12, i) * p ** i * (1 - p) ** (12 - i) + for i in range(k, 13)) + assert tail(req_heavy, 6) >= 0.99 > tail(req_heavy - 0.02, 6) + assert tail(req_light, 11) >= 0.99 > tail(req_light - 0.005, 11) + # cross-check against sim_interframe itself (whole-frame erasures) + rng = random.Random(7) + ch = {"corrupt_rate": 1.0 - (req_heavy + 0.04), "n_sub": 10, + "survivor_hist": {0: 1}} + d_hi, _ = fec_ab_sim.sim_interframe(ch, 1.00, 12, 3000, rng, sbi=False) + ch["corrupt_rate"] = 1.0 - (req_heavy - 0.06) + d_lo, _ = fec_ab_sim.sim_interframe(ch, 1.00, 12, 3000, rng, sbi=False) + assert d_hi > 0.98 and d_lo < 0.98 + + +def test_mcs_probe_disabled_is_behaviourally_unchanged(): + """Default config (mcs_probe_enabled=False): identical trajectory to a + plain controller — the opt-in gate never changes shipped behaviour.""" + calib = em.load_calibration() + seq = [10 + i * 1.2 for i in range(25)] # climb through promotions + a = _drive_pl(_ctrl(target=0.99, allow_shed=False), seq, calib) + b = _drive_pl(_ctrl(target=0.99, allow_shed=False, mcs_probe_enabled=False), + seq, calib) + assert (a.mcs, a.overhead, a.txagc) == (b.mcs, b.overhead, b.txagc) + + +def test_mcs_block_vacates_candidate_and_recovers(): + """A candidate whose probe UPPER bound cannot reach even the heaviest + overhead's requirement is blocked from candidacy for the hold, then + re-admitted; the lowest rate is never blocked (it is the fallback).""" + calib = em.load_calibration() + kw = dict(target=0.99, allow_shed=False, improve_frac=0.0, + mcs_probe_block_hold_ms=5000) + ref, _, _ = _settle_pl(_fast_ctrl(**kw), calib, pl=0.0, t0=0) + assert ref.mcs > 0 # the unblocked optimum + c = _fast_ctrl(**kw) + c.report_mcs_delivery({ref.mcs: _stat(0.0, 0.2, 12), 0: _stat(0.0, 0.2, 12)}, + now_ms=0) + assert 0 not in c._mcs_block # floor never blocked + early, _, _ = _settle_pl(c, calib, pl=0.0, t0=100, ticks=4) + assert early.mcs != ref.mcs # vacated while held + # after the hold expires the block no longer bars the pick (a cold pick — + # displacing an INCUMBENT additionally rides the normal e_bit hysteresis) + c3 = _fast_ctrl(**kw) + c3.report_mcs_delivery({ref.mcs: _stat(0.0, 0.2, 12)}, now_ms=0) + late, _, _ = _settle_pl(c3, calib, pl=0.0, t0=6000) + assert late.mcs == ref.mcs + # too few samples -> no block + c2 = _fast_ctrl(**kw) + c2.report_mcs_delivery({ref.mcs: _stat(0.0, 0.2, 4)}, now_ms=0) + ok, _, _ = _settle_pl(c2, calib, pl=0.0, t0=100) + assert ok.mcs == ref.mcs + + +def test_up_gate_requires_probe_lcb_then_trims_overhead(): + """With probing on, a faster row is entered only on probe evidence — and + since the raw requirement rises steeply as overhead falls, the promote + path is two-step: mcs+1 at the admissible (heavier) overhead first, then + the ungated same-MCS overhead trim.""" + calib = em.load_calibration() + kw = dict(target=0.99, allow_shed=False, mcs_probe_enabled=True, + improve_frac=0.0, mcs_set=(3, 4), overhead_set=(0.10, 0.25, 0.50)) + c = _fast_ctrl(**kw) + # weak: MCS4 unreachable even at max power -> settles on the floor rate + op0, t, txagc = _settle_pl(c, calib, pl=-3.5, t0=0) + assert op0.mcs == 3 + # strong link, NO probe evidence: the model alone may not promote + opA, t, txagc = _settle_pl(c, calib, pl=25.0, t0=t, ticks=12, txagc=txagc) + assert opA.mcs == 3 + # evidence for MCS4 whose LCB clears ov=0.50 (req~0.88) but not ov=0.25 + # (req~0.96): only the heavier-FEC row is admissible on the entry tick + c.report_mcs_delivery({4: _stat(0.93, 1.0, 40)}, now_ms=t) + opB, t, txagc = _settle_pl(c, calib, pl=25.0, t0=t, ticks=1, txagc=txagc) + assert opB.mcs == 4 and opB.overhead == 0.50 + # the same-MCS overhead trim is ungated: it follows on the next ticks + opC, t, _ = _settle_pl(c, calib, pl=25.0, t0=t, ticks=4, txagc=txagc) + assert opC.mcs == 4 and opC.overhead < 0.50 + + +def test_measured_active_collapse_escapes_model_belief(): + """A row the model wrongly believes in (here: cold-picked on an optimistic + bias) is escaped by MEASURED active-row delivery: after the persistence + guard the row is blocked, cur_ok fails despite the model, and the + fast-down commits a row the model alone would never have left for. A + non-persistent condemnation (one burst-flooded report) does not trip.""" + from controller import apply_model_bias + calib = em.load_calibration() + c = _fast_ctrl(target=0.99, allow_shed=False, mcs_probe_enabled=True, + improve_frac=0.0) + apply_model_bias(c, {5: -10.0}) + op, t, txagc = _settle_pl(c, calib, pl=0.0, t0=0) + assert op.mcs == 5 # the cold pick lands on the lie + bad = {5: _stat(0.0, 0.3, 40)} # measured: the active row is dead + # burst guard: an interrupted condemnation never trips + c.report_mcs_delivery(bad, now_ms=t) + c.report_mcs_delivery({}, now_ms=t + 100) + c.report_mcs_delivery(bad, now_ms=t + 200) + op2 = c.update(0.0 + calib.gain_db(txagc), txagc, now_ms=t + 300) + assert op2.mcs == 5 + # persistent condemnation: trips, blocks, escapes + for i in range(3): + c.report_mcs_delivery(bad, now_ms=t + 400 + i * 100) + esc, _, _ = _settle_pl(c, calib, pl=0.0, t0=t + 800, ticks=2, txagc=txagc) + assert esc.mcs != 5 + assert c._mcs_blocked(5) # the lie cannot be re-picked + + +def test_fast_down_is_ungated(): + """Escaping a failing point never waits for probe samples: on !cur_ok the + model's candidate commits with zero evidence even with probing enabled.""" + calib = em.load_calibration() + c = _fast_ctrl(target=0.99, allow_shed=False, mcs_probe_enabled=True) + op0, t, txagc = _settle_pl(c, calib, pl=20.0, t0=0) + assert op0.mcs > 0 + drop, _, _ = _settle_pl(c, calib, pl=-15.0, t0=t, ticks=6, txagc=txagc) + assert drop is not None and drop.mcs < op0.mcs # downgraded immediately diff --git a/tools/precoder/test_mcs_probe_ab_sim.py b/tools/precoder/test_mcs_probe_ab_sim.py new file mode 100644 index 00000000..98ba1079 --- /dev/null +++ b/tools/precoder/test_mcs_probe_ab_sim.py @@ -0,0 +1,70 @@ +"""Tests for the MCS-probe A/B sim (`mcs_probe_ab_sim.py`). + +Pins the headline the feature exists for: a model that is optimistically wrong +about one rate traps the model-only controller (it promotes in and its own +belief keeps it there), while probe evidence keeps the probing controller out — +SLA held, energy per delivered bit orders of magnitude apart. The clean-channel +control quantifies the honest cost: bounded probe airtime and a bounded energy +premium from the evidence-conservative operating point. +""" + +from __future__ import annotations + +import mcs_probe_ab_sim as ab + + +def _static(bias): + return ab.run_scenario("static", n_frames=9000, bias=bias, seed=1, + trials=250) + + +def test_biased_model_traps_model_arm_probes_hold_sla(): + arms = _static(bias={5: -8.0}) + model, probe = arms["model"], arms["probe"] + # the model arm enters the lie and stays (its cur_ok shares the wrong + # belief); the probe arm may touch it transiently (cold picks are + # model-trusted, and a video-loss reset re-picks cold), but the measured + # active-row delivery condemns and blocks it within seconds + assert model.biased_row_ms is not None + assert model.biased_frames > 4000 + assert probe.biased_frames < 500 + assert probe.delivery >= 0.95 + assert model.delivery < 0.5 + # delivered-bit energy: the collapse makes the model arm ruinous + assert probe.e_bit_nj < model.e_bit_nj / 10 + # probing did not add churn + assert probe.commits <= model.commits + 3 + + +def test_ambush_cold_capture_self_heals_with_probes(): + """No acquisition segment: the ungated cold pick lands BOTH arms on the + biased row. The probe arm's measured active-row delivery condemns it + within seconds (trip -> block -> fast-down; the block outlives the escape + so it is not re-picked), while the model-only arm — whose belief is the + thing that is wrong — stays for the whole run.""" + arms = ab.run_scenario("ambush", n_frames=9000, bias={5: -8.0}, seed=1, + trials=250) + model, probe = arms["model"], arms["probe"] + assert model.biased_row_ms is not None + assert probe.biased_row_ms is not None # cold-captured too... + assert probe.biased_frames < 1500 # ...but escapes in seconds + assert model.biased_frames > 4000 # stuck for most of the run + assert probe.first_block_ms is not None + assert probe.delivery >= 0.9 + assert model.delivery < 0.5 + assert probe.e_bit_nj < model.e_bit_nj / 5 + + +def test_clean_channel_probe_cost_is_bounded(): + arms = _static(bias=None) + model, probe = arms["model"], arms["probe"] + # a well-calibrated model needs no probes: both arms hold the SLA... + assert model.delivery >= 0.97 and probe.delivery >= 0.97 + # ...and the probe arm's cost is bounded: ~3% scheduled duty on airtime + # and a bounded energy premium from its evidence-conservative operating + # point (promotions wait for proof at the incumbent's power) + assert probe.probe_airtime_frac < 0.06 + assert probe.e_bit_nj < model.e_bit_nj * 1.6 + # probe bookkeeping is alive (attempted and priced) + assert probe.probe_frames > 50 + assert model.probe_frames == 0 diff --git a/tools/precoder/test_rc_proto.py b/tools/precoder/test_rc_proto.py index eca35627..ebd37a97 100644 --- a/tools/precoder/test_rc_proto.py +++ b/tools/precoder/test_rc_proto.py @@ -103,3 +103,38 @@ def test_probe_schedule_is_shared_and_low_duty(): assert probes == 30 # two-rung set only uses two slots assert rp.probe_bw(16, (20, 40)) is None + + +def test_mcs_probe_schedule_adjacent_and_boundaries(): + mcs_set = tuple(range(8)) + # slot 4 = the next rate above, slot 20 = the next below + assert rp.probe_mcs(4, 4, mcs_set) == 5 + assert rp.probe_mcs(20, 4, mcs_set) == 3 + assert rp.probe_mcs(5, 4, mcs_set) is None # off-slot + assert rp.probe_mcs(4 + 64, 4, mcs_set) == 5 # periodic + # boundaries of the set: nothing adjacent -> None + assert rp.probe_mcs(4, 7, mcs_set) is None # no rate above top + assert rp.probe_mcs(20, 0, mcs_set) is None # no rate below bottom + # adjacency is within the SET, not selected±1 (sparse sets probe neighbours) + sparse = (0, 2, 4, 6, 7) + assert rp.probe_mcs(4, 4, sparse) == 6 + assert rp.probe_mcs(20, 4, sparse) == 2 + # selected not in the set: no well-defined neighbour -> never probe + assert rp.probe_mcs(4, 3, sparse) is None + assert rp.probe_mcs(20, 3, sparse) is None + + +def test_mcs_probe_schedule_disjoint_duty_and_wrap(): + bw_set = (20, 40, 80) + mcs_set = tuple(range(8)) + # disjoint from the bandwidth-probe slots over the whole 12-bit seq space: + # every probe observation has exactly one changed variable + for s in range(4096): + assert not (rp.probe_bw(s, bw_set) is not None + and rp.probe_mcs(s, 4, mcs_set) is not None) + # duty: 2 probes (one up, one down) per 64 seqs + probes = sum(1 for s in range(640) if rp.probe_mcs(s, 4, mcs_set) is not None) + assert probes == 20 + # the period divides 4096, so the wrap is phase-safe: seq 4 and 4+4096k agree + assert 4096 % rp.MCS_PROBE_PERIOD == 0 + assert rp.probe_mcs(4, 4, mcs_set) == rp.probe_mcs((4 + 4096) % 4096, 4, mcs_set) diff --git a/tools/precoder/test_score.py b/tools/precoder/test_score.py index 15d62f37..363811f6 100644 --- a/tools/precoder/test_score.py +++ b/tools/precoder/test_score.py @@ -79,3 +79,163 @@ def test_rung_window_attributes_losses_by_probe_schedule(): st = rw.stats() assert st[20][0] == 1.0 and st[40][0] == 1.0 assert st[80][0] == 0.0 and st[80][1] >= 8 + + +# --------------------------------------------------------------------------- # +# Wilson bounds + McsProbeWindow +# --------------------------------------------------------------------------- # +def test_wilson_bounds_known_values(): + from score import wilson_bounds + z = 1.2816 + assert wilson_bounds(0, 0, z) == (0.0, 1.0) # no evidence at all + lcb, ucb = wilson_bounds(12, 12, z) + # a perfect record caps at n/(n+z^2): "0 failures in 12" is NOT proof of 99% + assert abs(lcb - 12 / (12 + z * z)) < 1e-9 + assert ucb == 1.0 and lcb < 0.89 + lcb, ucb = wilson_bounds(5, 10, z) + assert lcb < 0.5 < ucb + l31, _ = wilson_bounds(31, 31, z) + assert l31 > lcb # more evidence -> tighter + + +MCS_SET = tuple(range(8)) + + +def _mcs_stream(win, seqs, sel=4, mode="ht", lost=(), t0=0.0, dt_ms=4.0, + rate_of=None): + """Feed a seq stream: probe seqs fly their candidate, others the selected + rate; seqs in `lost` are skipped (a gap at the VRX).""" + import rc_proto as rp + t = t0 + for s in seqs: + t += dt_ms + if s in lost: + continue + flown = rp.probe_mcs(s, sel, MCS_SET) + flown = sel if flown is None else flown + rate = (mode, flown) if rate_of is None else rate_of(s) + win.add_seq(s, t, rate=rate, crc_err=False) + return t + + +def test_mcs_probe_window_attributes_by_schedule(): + """Kill exactly the up-probe seqs: the up candidate's delivery collapses + while the down candidate stays perfect — sensing from gaps + verified + receptions alone.""" + import rc_proto as rp + from score import McsProbeWindow + win = McsProbeWindow(MCS_SET) + win.set_selected("ht", 4, 0.0) + lost = {s for s in range(64 * 16) if rp.probe_mcs(s, 4, MCS_SET) == 5} + t = _mcs_stream(win, range(64 * 16), lost=lost) + st = win.stats(t) + assert st[5].delivery == 0.0 and st[5].attempts >= 12 + assert st[3].delivery == 1.0 and st[3].attempts >= 12 + assert st[5].ucb < 0.3 and st[3].lcb > 0.8 + + +def test_mcs_probe_window_epoch_gating(): + """Gap losses are attributed only while a non-probe frame has confirmed the + VTX flies the commanded rate; a mismatch un-confirms.""" + import rc_proto as rp + from score import McsProbeWindow + lost = {s for s in range(256) if rp.probe_mcs(s, 4, MCS_SET) is not None} + # never confirmed: every non-probe frame decodes at the WRONG rate + win = McsProbeWindow(MCS_SET) + win.set_selected("ht", 4, 0.0) + t = _mcs_stream(win, range(256), lost=lost, rate_of=lambda s: ("ht", 0)) + assert win.stats(t) == {} + # confirmed: same losses now count as candidate failures + win2 = McsProbeWindow(MCS_SET) + win2.set_selected("ht", 4, 0.0) + t = _mcs_stream(win2, range(256), lost=lost) + st = win2.stats(t) + assert st[5].attempts >= 3 and st[5].delivery == 0.0 + # un-confirm: after a mismatching non-probe frame, attribution stops + win3 = McsProbeWindow(MCS_SET) + win3.set_selected("ht", 4, 0.0) + win3.add_seq(0, 1.0, rate=("ht", 4)) # confirm + win3.add_seq(1, 2.0, rate=("ht", 0)) # VTX visibly elsewhere -> unconfirm + win3.add_seq(6, 3.0, rate=("ht", 4)) # gap walks the lost up-probe seq 4 + assert 5 not in win3.stats(4.0) + + +def test_mcs_probe_window_rate_verify_and_crc(): + from score import McsProbeWindow + # a received probe-slot frame with a MISMATCHED rate is ignored, not failed: + # the candidate never verifiably flew (lag, suppressed probe, parse fallback) + win = McsProbeWindow(MCS_SET) + win.set_selected("ht", 4, 0.0) + win.add_seq(3, 1.0, rate=("ht", 4)) # confirm epoch + win.add_seq(4, 2.0, rate=("ht", 4)) # up-probe slot flew at selected + assert 5 not in win.stats(3.0) + # rate-less frames are dropped entirely (fail-inert without telemetry) + win.add_seq(68, 3.0, rate=None) # up-probe slot, no rate info + assert 5 not in win.stats(4.0) + # a crc_err frame never advances the gap walk (its body seq is untrusted + # bits) but its PHY-decoded rate is descriptor truth: a corrupt frame at a + # tracked rate is that rate's measured failure, pushed seq-free — the + # evidence that keeps flowing when the stream is crc-flooded + win2 = McsProbeWindow(MCS_SET) + win2.set_selected("ht", 4, 0.0) + win2.add_seq(3, 1.0, rate=("ht", 4)) # confirm; last_seq = 3 + win2.add_seq(999, 2.0, rate=("ht", 4), crc_err=True) # active-rate failure + win2.add_seq(998, 2.1, rate=("ht", 5), crc_err=True) # candidate failure + win2.add_seq(997, 2.2, rate=("ht", 0), crc_err=True) # untracked: ignored + win2.add_seq(996, 2.3, rate=None, crc_err=True) # rate-less: ignored + win2.add_seq(5, 3.0, rate=("ht", 4)) # gap 3->5 walks the lost seq 4 + st = win2.stats(4.0) + assert st[5].attempts == 2 and st[5].delivery == 0.0 # gap walk + crc + assert (4 in st) and not st[4].delivery == 1.0 # active crc counted + assert 0 not in st + + +def test_mcs_probe_window_tracks_active_row(): + """The selected row's delivery is measured from the main stream (free + evidence — this is what lets the controller escape a rate its model + wrongly believes in); bandwidth-probe slots are bandwidth evidence and + never touch it.""" + import rc_proto as rp + from score import McsProbeWindow + bw_set = (20, 40, 80) + win = McsProbeWindow(MCS_SET, bw_set=bw_set) + win.set_selected("ht", 4, 0.0) + t = _mcs_stream(win, range(256)) + st = win.stats(t) + assert st[4].delivery == 1.0 + assert st[4].attempts > st[5].attempts # main stream >> probe duty + # every bw-probe seq killed: the ACTIVE row stays clean (their loss is + # about the rung's spectrum, not the rate) + lost = {s for s in range(256, 512) if rp.probe_bw(s, bw_set) is not None} + t = _mcs_stream(win, range(256, 512), lost=lost, t0=t) + assert win.stats(t)[4].delivery == 1.0 + # the active (non-probe) seqs killed: the selected row's delivery + # collapses while the probe slots keep verifying + lost = {s for s in range(512, 768) + if rp.probe_mcs(s, 4, MCS_SET) is None + and rp.probe_bw(s, bw_set) is None} + t = _mcs_stream(win, range(512, 768), lost=lost, t0=t) + st = win.stats(t) + assert st[4].delivery < 0.2 and st[4].ucb < 0.5 + assert st[5].delivery == 1.0 + + +def test_mcs_probe_window_wrap_age_and_reset(): + import rc_proto as rp + from score import McsProbeWindow + win = McsProbeWindow(MCS_SET, max_age_ms=8000.0) + win.set_selected("ht", 4, 0.0) + # contiguous stream straddling the 12-bit wrap: no spurious attribution + t = _mcs_stream(win, [4094, 4095, 0, 1, 2], t0=0.0) + assert all(s.delivery == 1.0 for s in win.stats(t).values()) + # probes keep landing on the right candidates after the wrap + t = _mcs_stream(win, range(3, 130), t0=t) + st = win.stats(t) + assert st[5].attempts >= 2 and st[3].attempts >= 2 + # age expiry: far-future stats() prunes everything + assert win.stats(t + 9000.0) == {} + # a profile change blanks the window (evidence is context-bound) + t = _mcs_stream(win, range(200, 400), t0=0.0) + assert win.stats(t) != {} + win.set_selected("ht", 5, t) + assert win.stats(t) == {}