"""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()