Lab065: validate OTA gain improvement at TX VGA 20 dB

tests/test_lab065_rf_level_linearity.py

@@ -0,0 +1,151 @@
+"""Lab065: неизменный waveform, TX VGA 20 дБ и границы классификации."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+import subprocess
+import sys
+
+import pytest
+
+from experiments import lab065_rf_level_linearity as lab
+from tools.hackrf_process_check import ProcessCheck, ProcessState
+
+
+def test_only_operating_parameter_change_is_tx_vga():
+    old = lab.lab064.parameters()
+    new = lab.parameters()
+    assert new.pop("geometry_unchanged_from_lab064") is True
+    assert new.pop("waveform_source") == "Lab064 exact CS8 copy"
+    assert new["tx_vga_db"] == 20
+    new["tx_vga_db"] = 10
+    assert new == old
+
+
+def test_tx_command_uses_20_db_and_rx_remains_24_24():
+    transmit = lab.transfer_command("tx.cs8", 100, transmit=True)
+    receive = lab.transfer_command("rx.cs8", 100)
+    assert transmit[transmit.index("-x") + 1] == "20"
+    assert receive[receive.index("-l") + 1] == "24"
+    assert receive[receive.index("-g") + 1] == "24"
+    assert transmit[-5:] == receive[-5:] == ["-a", "0", "-p", "0", "-B"]
+
+
+def test_exact_lab064_waveform_is_required():
+    metadata = lab.validate_lab064_waveform()
+    assert metadata["tx_sha256"] == lab.EXPECTED_TX_SHA256
+    assert metadata["tx_samples"] == 6_000_000
+
+
+def _result(matched_gain, residual_gain, improvements, errors=(0.0, 0.0, 0.0)):
+    old = {row["label"]: row for row in json.loads(
+        lab.LAB064_RESULT.read_text(encoding="utf-8")
+    )["rows"]}
+    rows = []
+    for index, label in enumerate(("L0", "L1", "L2", "L3")):
+        baseline = old[label]
+        rows.append({
+            "label": label,
+            "matched_signal_power": baseline["matched_signal_power"] * 10 ** (matched_gain[index] / 10),
+            "residual_power": baseline["residual_power"] * 10 ** (residual_gain[index] / 10),
+            "matched_to_residual_db": baseline["matched_to_residual_db"] + improvements[index],
+            "gain_power_abs_h_squared": baseline["gain_power_abs_h_squared"] * 10 ** (matched_gain[index] / 10),
+        })
+    return {
+        "capture_valid": True,
+        "rows": rows,
+        "comparisons": [{"linearity_error_db": error} for error in errors],
+    }
+
+
+@pytest.mark.parametrize(
+    ("result", "expected"),
+    [
+        (_result([10] * 4, [1] * 4, [9] * 4), "LINEAR_GAIN_IMPROVEMENT"),
+        (_result([10] * 4, [10] * 4, [0] * 4), "RESIDUAL_RISES_WITH_SIGNAL"),
+        (_result([1] * 4, [0] * 4, [1] * 4), "NO_MEANINGFUL_GAIN"),
+        (_result([10] * 4, [1] * 4, [9] * 4, errors=(0, 0, -1.6)), "COMPRESSION_AT_HIGHER_TX_GAIN"),
+    ],
+)
+def test_classification_cases(result, expected):
+    comparison = lab.compare_sessions(
+        json.loads(lab.LAB064_RESULT.read_text(encoding="utf-8")), result["rows"]
+    )
+    assert lab.classify_result(result, comparison)["case"] == expected
+
+
+@pytest.mark.parametrize("failure", [None, "clock", "overrun", "clipping"])
+def test_preflight_never_transmits_and_fails_closed(tmp_path, monkeypatch, failure):
+    raw, output = tmp_path / "raw", tmp_path / "output"
+    raw.mkdir()
+    output.mkdir()
+    monkeypatch.setattr(lab, "RAW", raw)
+    monkeypatch.setattr(lab, "OUTPUT", output)
+    (raw / "tx_level_staircase.cs8").write_bytes(b"test")
+    metadata = {"tx_sha256": "TEST"}
+    (output / "software_gate.json").write_text('{"passed": true, "tx_sha256": "TEST"}')
+    monkeypatch.setattr(lab, "validate_prepared", lambda: metadata)
+    monkeypatch.setattr(lab.lab054.phy.lab045, "_verify_device_inventory", lambda runner: {"passed": True})
+    monkeypatch.setattr(lab.lab054, "disable_clock_outputs", lambda runner: [])
+    monkeypatch.setattr(lab, "parameters", lambda: {})
+
+    def clock(point, runner, guard, **kwargs):
+        return {"passed": failure != "clock", "master_reg29": 0x26,
+                "slave_reg29": 0x26, "slave_clkin_status": "DETECTED"}
+
+    def runner(command):
+        assert "-t" not in command
+        if "-r" in command:
+            value = 127 if failure == "clipping" else 1
+            Path(command[command.index("-r") + 1]).write_bytes(bytes([value, value]) * 800_000)
+        log = "1 RX overruns" if failure == "overrun" else "0 RX overruns"
+        return subprocess.CompletedProcess(command, 0, log, "")
+
+    options = {
+        "runner": runner,
+        "clock_checker": clock,
+        "process_checker": lambda: ProcessCheck(ProcessState.CLEAR, ()),
+        "sleep_function": lambda _: None,
+    }
+    if failure:
+        with pytest.raises(RuntimeError):
+            lab.preflight(**options)
+    else:
+        assert lab.preflight(**options)["passed"]
+    assert json.loads((raw / "preflight_result.json").read_text())["rf_tx_count"] == 0
+
+
+def test_capture_is_unavailable_without_new_authorization():
+    with pytest.raises(PermissionError):
+        lab.capture_once(authorization=False)
+
+
+def test_existing_attempt_blocks_capture_before_hardware(tmp_path, monkeypatch):
+    monkeypatch.setattr(lab, "RAW", tmp_path)
+    monkeypatch.setattr(lab, "validate_prepared", lambda: {"tx_sha256": "TEST"})
+    monkeypatch.setattr(lab, "require_preflight", lambda metadata: {})
+    (tmp_path / "hardware_attempt.json").write_text("{}", encoding="utf-8")
+    with pytest.raises(FileExistsError):
+        lab.capture_once(
+            authorization=True,
+            runner=lambda command: pytest.fail("hardware must not run"),
+            process_starter=lambda *args: pytest.fail("hardware must not run"),
+        )
+
+
+def test_cli_labels_current_rf_state_after_completed_capture(monkeypatch, capsys):
+    monkeypatch.setattr(
+        lab,
+        "capture_once",
+        lambda authorization=False: {
+            "stage": "SINGLE_RF_LEVEL_CAPTURE", "rf_tx_count": 1, "passed": True,
+        },
+    )
+    monkeypatch.setattr(
+        sys, "argv", ["lab065", "--stage", "capture", "--authorize-one-rf"]
+    )
+    lab.main()
+    output = capsys.readouterr().out
+    assert '"rf_tx_count": 1' in output
+    assert output.endswith("RF TX NOW = 0\n")