199 lines
8.0 KiB
Python
199 lines
8.0 KiB
Python
"""Lab064: цифровая лестница, измерительные метрики и запрет RF без разрешения."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from experiments import lab064_rf_level_linearity as lab
|
|
from tools.hackrf_process_check import ProcessCheck, ProcessState
|
|
|
|
|
|
def test_fixed_operating_point_and_measurement_scope():
|
|
parameters = lab.parameters()
|
|
assert parameters["rf_frequency_hz"] == 1_400_000_000
|
|
assert parameters["sample_rate_hz"] == 8_000_000
|
|
assert parameters["symbol_rate"] == 20_000
|
|
assert parameters["samples_per_symbol"] == 400
|
|
assert parameters["digital_carrier_hz"] == 50_000
|
|
assert parameters["levels_db_relative_nominal"] == [-18.0, -12.0, -6.0, 0.0]
|
|
assert parameters["full_phy_packet"] is False
|
|
assert parameters["tx_vga_db"] == 10
|
|
assert parameters["rx_lna_db"] == parameters["rx_vga_db"] == 24
|
|
assert lab.POINT.distance_m == 1.2
|
|
assert lab.POINT.visibility == "LOS indoor"
|
|
|
|
|
|
def test_waveform_has_exact_guards_segments_gaps_and_levels():
|
|
rows, blocks = lab.prepared_segments()
|
|
assert lab.TX_SAMPLES == 6_000_000
|
|
assert lab.TX_SAMPLES / lab.SAMPLE_RATE == 0.75
|
|
assert all(row["samples"] == 800_000 for row in rows)
|
|
assert [row["start_sample"] for row in rows] == [800_000, 2_000_000, 3_200_000, 4_400_000]
|
|
assert all(len(blocks[row["label"]]) == 2 * lab.SEGMENT_SAMPLES for row in rows)
|
|
assert all(not row["metrics"]["digital_clipping"] for row in rows)
|
|
steps = [row["tx_input_difference_from_previous_db"] for row in rows[1:]]
|
|
assert np.allclose(steps, 6.0, atol=0.1)
|
|
|
|
|
|
def test_background_windows_cover_both_guards_and_all_gaps():
|
|
windows = lab.background_windows()
|
|
assert [row["label"] for row in windows] == [
|
|
"INITIAL_GUARD", "GAP_L0_L1", "GAP_L1_L2", "GAP_L2_L3", "FINAL_GUARD"
|
|
]
|
|
assert all(row["samples"] == 160_000 for row in windows)
|
|
zero_ranges = [
|
|
(0, 800_000),
|
|
(1_600_000, 2_000_000),
|
|
(2_800_000, 3_200_000),
|
|
(4_000_000, 4_400_000),
|
|
(5_200_000, 6_000_000),
|
|
]
|
|
for row, (low, high) in zip(windows, zero_ranges):
|
|
assert low <= row["tx_relative_start_sample"]
|
|
assert row["tx_relative_start_sample"] + row["samples"] <= high
|
|
|
|
|
|
def test_background_stability_uses_ac_power_range():
|
|
stable = lab.summarize_background([
|
|
{"metrics": {"ac_power_code2": 1.0, "ac_rms_code": 1.0}},
|
|
{"metrics": {"ac_power_code2": 1.5, "ac_rms_code": np.sqrt(1.5)}},
|
|
])
|
|
changed = lab.summarize_background([
|
|
{"metrics": {"ac_power_code2": 1.0, "ac_rms_code": 1.0}},
|
|
{"metrics": {"ac_power_code2": 4.0, "ac_rms_code": 2.0}},
|
|
])
|
|
assert stable["stable"] is True
|
|
assert changed["stable"] is False
|
|
assert changed["maximum_to_minimum_ac_power_db"] == pytest.approx(6.020599913)
|
|
|
|
|
|
def test_synthetic_linear_channel_recovers_digital_steps():
|
|
result = lab.synthetic_linearity_gate()
|
|
assert result["passed"]
|
|
assert max(abs(value) for value in result["linearity_errors_db"]) < 0.02
|
|
gains = [row["gain_abs"] for row in result["rows"]]
|
|
assert np.allclose(gains, 0.37, atol=0.001)
|
|
|
|
|
|
def test_counter_parser_keeps_every_intermediate_and_final_observation():
|
|
log = "0 TX underruns\n1 TX underrun\n3 TX underruns\n3 TX underruns\n"
|
|
summary = lab.counter_summary(log, "underrun")
|
|
assert summary["selected_counts"] == [0, 1, 3, 3]
|
|
assert summary["final_count"] == 3
|
|
assert summary["maximum_count"] == 3
|
|
assert summary["clean"] is False
|
|
assert [row["character_offset"] for row in summary["observations"]] == sorted(
|
|
row["character_offset"] for row in summary["observations"]
|
|
)
|
|
assert lab.counter_summary("0 RX overruns", "overrun")["clean"] is True
|
|
|
|
|
|
def test_intermediate_loss_cannot_be_hidden_by_final_zero():
|
|
tx = lab.counter_summary("0 TX underruns\n1 TX underrun\n0 TX underruns", "underrun")
|
|
rx = lab.counter_summary("0 RX overruns", "overrun")
|
|
validity = lab.capture_validity({
|
|
"passed": False,
|
|
"rx_clipping": False,
|
|
"tx_counter_summary": tx,
|
|
"rx_counter_summary": rx,
|
|
})
|
|
assert tx["final_count"] == 0
|
|
assert tx["maximum_count"] == 1
|
|
assert validity["valid"] is False
|
|
assert "TX_MAXIMUM_COUNTER_NOT_ZERO" in validity["invalid_reasons"]
|
|
assert "CAPTURE_RUNNER_NOT_PASSED" in validity["invalid_reasons"]
|
|
|
|
|
|
def test_commands_fix_gains_and_keep_rx_separate_from_tx():
|
|
receive = lab.transfer_command("rx.cs8", 800_000)
|
|
transmit = lab.transfer_command("tx.cs8", 100, transmit=True)
|
|
assert "-r" in receive and "-t" not in receive
|
|
assert "-t" in transmit and "-r" not in transmit
|
|
assert receive[receive.index("-f") + 1] == "1400000000"
|
|
assert receive[receive.index("-s") + 1] == "8000000"
|
|
assert receive[receive.index("-l") + 1] == receive[receive.index("-g") + 1] == "24"
|
|
assert transmit[transmit.index("-x") + 1] == "10"
|
|
assert receive[-5:] == ["-a", "0", "-p", "0", "-B"]
|
|
assert transmit[-5:] == ["-a", "0", "-p", "0", "-B"]
|
|
|
|
|
|
@pytest.mark.parametrize("failure", [None, "clock", "clipping", "strong", "overrun"])
|
|
def test_preflight_never_transmits_and_fails_closed(tmp_path, monkeypatch, failure):
|
|
raw = tmp_path / "raw"
|
|
output = tmp_path / "processed"
|
|
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"}', encoding="utf-8"
|
|
)
|
|
monkeypatch.setattr(lab, "validate_prepared", lambda: metadata)
|
|
monkeypatch.setattr(
|
|
lab.lab054.phy.lab045,
|
|
"_verify_device_inventory",
|
|
lambda runner: {"passed": True},
|
|
)
|
|
|
|
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 (110 if failure == "strong" 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, "")
|
|
|
|
monkeypatch.setattr(lab.lab054, "disable_clock_outputs", lambda runner: [])
|
|
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"]
|
|
result = json.loads((raw / "preflight_result.json").read_text(encoding="utf-8"))
|
|
assert result["rf_tx_count"] == 0
|
|
assert result["stop_before_tx"] is True
|
|
|
|
|
|
def test_second_preflight_is_refused_before_hardware(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(lab, "RAW", tmp_path)
|
|
monkeypatch.setattr(lab, "OUTPUT", tmp_path)
|
|
(tmp_path / "preflight_attempt.json").write_text("{}", encoding="utf-8")
|
|
(tmp_path / "software_gate.json").write_text(
|
|
'{"passed": true, "tx_sha256": "TEST"}', encoding="utf-8"
|
|
)
|
|
monkeypatch.setattr(lab, "validate_prepared", lambda: {"tx_sha256": "TEST"})
|
|
with pytest.raises(FileExistsError):
|
|
lab.preflight(runner=lambda command: pytest.fail("hardware must not run"))
|
|
|
|
|
|
def test_capture_refuses_missing_authorization_before_files_or_hardware(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(lab, "RAW", tmp_path)
|
|
monkeypatch.setattr(lab, "OUTPUT", tmp_path)
|
|
with pytest.raises(PermissionError):
|
|
lab.capture_once(
|
|
authorization=False,
|
|
runner=lambda command: pytest.fail("hardware must not run"),
|
|
)
|
|
assert not (tmp_path / "hardware_attempt.json").exists()
|