import subprocess import sys from pathlib import Path from types import SimpleNamespace import numpy as np import pytest from experiments import lab043_hackrf_counter as counter from experiments import lab043_hackrf_hardware as hackrf from experiments import lab043_pluto_to_rtlsdr as lab043 from experiments import lab043_prbs11_hardware as pluto_hardware class _FakeHackrfReceiver: def __init__(self) -> None: self.gain_calls: list[tuple[int, int]] = [] def configure_rx(self) -> dict: return { "set_center_frequency_hz": 435_000_000, "set_sample_rate_hz": 2_400_000.0, "rf_amp_enabled": False, "lna_gain_db": 8, "vga_gain_db": 8, "antenna_bias_enabled": False, } def set_rx_gains(self, lna_gain_db: int, vga_gain_db: int) -> dict: self.gain_calls.append((lna_gain_db, vga_gain_db)) return { "rf_amp_enabled": False, "antenna_bias_enabled": False, "lna_gain_db": lna_gain_db, "vga_gain_db": vga_gain_db, "gain_readback_available": False, "set_calls_succeeded": True, } class _FakeHackrfModule: def __init__(self) -> None: self.preflight_calls: list[tuple[int, int]] = [] def preflight_hackrf(self, lna_gain_db: int, vga_gain_db: int) -> dict: self.preflight_calls.append((lna_gain_db, vga_gain_db)) return _FakeHackrfReceiver().configure_rx() class _FakePlutoModule: def __init__(self) -> None: self.preflight_uris: list[str] = [] @staticmethod def discover_pluto_usb_uri() -> str: return "usb:1.2.3" def preflight_pluto(self, uri: str) -> dict: self.preflight_uris.append(uri) return {"uri": uri, "context_opened": True} def test_counter_uses_dedicated_output_directory() -> None: assert counter.PROCESSED_DIRECTORY == Path( "data/processed/lab043_hackrf_counter" ) @pytest.mark.parametrize( "script_name", ["lab043_hackrf_hardware.py", "lab043_hackrf_counter.py"], ) def test_hackrf_scripts_support_direct_help_invocation(script_name: str) -> None: project_root = Path(__file__).resolve().parents[1] completed = subprocess.run( [sys.executable, str(project_root / "experiments" / script_name), "--help"], cwd=project_root, check=False, capture_output=True, text=True, encoding="utf-8", errors="replace", ) assert completed.returncode == 0, completed.stdout + completed.stderr def test_counter_receiver_is_forced_to_working_point_8_8() -> None: receiver = _FakeHackrfReceiver() parameters = counter.configure_counter_receiver(receiver) assert receiver.gain_calls == [(8, 8)] assert parameters["lna_gain_db"] == 8 assert parameters["vga_gain_db"] == 8 assert parameters["rf_amp_enabled"] is False assert parameters["antenna_bias_enabled"] is False def test_counter_and_hackrf_module_share_working_point_8_8() -> None: assert (counter.HACKRF_LNA_GAIN_DB, counter.HACKRF_VGA_GAIN_DB) == (8, 8) assert (hackrf.HACKRF_LNA_GAIN_DB, hackrf.HACKRF_VGA_GAIN_DB) == (8, 8) def test_counter_preflight_explicitly_requests_working_point_8_8() -> None: module = _FakeHackrfModule() parameters = counter.preflight_counter_receiver(module) assert module.preflight_calls == [(8, 8)] assert parameters["lna_gain_db"] == 8 assert parameters["vga_gain_db"] == 8 def test_counter_hardware_preflight_uses_discovered_usb_pluto() -> None: hackrf_module = _FakeHackrfModule() pluto_module = _FakePlutoModule() result = counter.preflight_counter_hardware(hackrf_module, pluto_module) assert pluto_module.preflight_uris == ["usb:1.2.3"] assert result["pluto"]["uri"] == "usb:1.2.3" assert hackrf_module.preflight_calls == [(8, 8)] def test_pluto_working_point_is_checked_before_tx() -> None: counter.validate_pluto_working_point( { "actual_center_frequency_hz": 435_000_000, "actual_sample_rate_hz": 2_400_000, "actual_gain_db": -30.0, } ) counter.validate_pluto_working_point( { "actual_center_frequency_hz": 435_000_000, "actual_sample_rate_hz": 2_399_998, "actual_gain_db": -30.0, } ) with pytest.raises(RuntimeError, match=r"Рабочая точка Pluto\+ не подтверждена"): counter.validate_pluto_working_point( { "actual_center_frequency_hz": 435_000_000, "actual_sample_rate_hz": 2_400_000, "actual_gain_db": -20.0, } ) with pytest.raises(RuntimeError, match=r"Рабочая точка Pluto\+ не подтверждена"): counter.validate_pluto_working_point( { "actual_center_frequency_hz": 435_000_000, "actual_sample_rate_hz": 2_399_000, "actual_gain_db": -30.0, } ) @pytest.mark.parametrize( ("field", "bad_value"), [ ("set_center_frequency_hz", 434_999_999), ("set_sample_rate_hz", 2_399_999.0), ("lna_gain_db", 0), ("vga_gain_db", 0), ("rf_amp_enabled", True), ("antenna_bias_enabled", True), ], ) def test_counter_rejects_any_other_hackrf_working_point( field: str, bad_value: object, ) -> None: parameters = _FakeHackrfReceiver().configure_rx() parameters[field] = bad_value with pytest.raises(RuntimeError, match="Рабочая точка HackRF не подтверждена"): counter.validate_counter_working_point(parameters) def test_event_rows_use_null_and_explicit_status_for_unknown_iq_count() -> None: events = [ { "received_symbol_index": 0, "type": "пропуск", } ] boundaries = ( SimpleNamespace(callback_index=1, accepted_end_sample=100), ) rows, _ = counter._event_hardware_rows( events, np.asarray([100.0]), boundaries, np.zeros(200, dtype=np.complex64), ) assert rows[0]["exact_lost_or_repeated_iq_samples"] is None assert rows[0]["exact_iq_sample_count_status"] == "not_computed" assert rows[0]["exact_iq_sample_count_reason"] def test_hardware_orchestration_saves_raw_before_analysis_and_transmits_once( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: events: list[str] = [] transmitted: list[np.ndarray] = [] configured_gains: list[tuple[int, int]] = [] class FakeLibrary: @staticmethod def hackrf_is_streaming(_device: object) -> int: return 1 class FakeDevice: library = FakeLibrary() device = object() def __enter__(self) -> "FakeDevice": return self def __exit__(self, *_arguments: object) -> None: return None def configure_rx(self) -> dict: events.append("configure_hackrf") return _FakeHackrfReceiver().configure_rx() def set_rx_gains(self, lna_gain_db: int, vga_gain_db: int) -> dict: configured_gains.append((lna_gain_db, vga_gain_db)) return { "rf_amp_enabled": False, "antenna_bias_enabled": False, "lna_gain_db": lna_gain_db, "vga_gain_db": vga_gain_db, "gain_readback_available": False, "set_calls_succeeded": True, } def capture(self, target_sample_count: int) -> SimpleNamespace: events.append("capture") assert target_sample_count == 4 return SimpleNamespace( samples=np.zeros(target_sample_count, dtype=np.complex64), boundaries=(), diagnostics={"callback_count_total": 1}, ) monkeypatch.setattr(counter, "RX_LEADING_SECONDS", 0.0) monkeypatch.setattr(counter, "RX_TRAILING_SECONDS", 0.0) monkeypatch.setattr(counter, "FIXED_TX_UPLOAD_MARGIN_SECONDS", 0.0) monkeypatch.setattr(hackrf, "HACKRF_EXPECTED_CALLBACK_BYTES", 0) monkeypatch.setattr(hackrf, "HackrfRxDevice", FakeDevice) monkeypatch.setattr( hackrf, "preflight_hackrf", lambda lna, vga: { **_FakeHackrfReceiver().configure_rx(), "lna_gain_db": lna, "vga_gain_db": vga, "serial_number": "test-serial", "firmware_version": "test-firmware", "libhackrf_version": "test-library", }, ) monkeypatch.setattr( pluto_hardware, "discover_pluto_usb_uri", lambda: "usb:1.2.3", ) monkeypatch.setattr( pluto_hardware, "preflight_pluto", lambda uri: {"context_opened": True, "uri": uri}, ) monkeypatch.setattr( pluto_hardware, "configure_pluto", lambda uri: ( object(), { "uri": uri, "actual_center_frequency_hz": 435_000_000, "actual_sample_rate_hz": 2_400_000, "actual_gain_db": -30.0, }, ), ) def fake_transmit( _device: object, samples: np.ndarray, _sample_rate_hz: float, ) -> float: events.append("transmit") transmitted.append(np.asarray(samples).copy()) return 0.0 monkeypatch.setattr( pluto_hardware, "transmit_noncyclic_buffer_once", fake_transmit, ) monkeypatch.setattr( counter, "build_counter_waveform", lambda _sample_rate_hz: ( np.asarray([0.25 + 0.25j] * 4, dtype=np.complex64), {"total_duration_seconds": 4 / counter.NOMINAL_SAMPLE_RATE}, ), ) real_save_raw = lab043.save_raw_iq_before_processing def save_raw(*arguments: object, **keyword_arguments: object): events.append("save_raw") return real_save_raw(*arguments, **keyword_arguments) monkeypatch.setattr(lab043, "save_raw_iq_before_processing", save_raw) def fake_analyze(*_arguments: object) -> tuple[dict, dict]: events.append("analyze") return {"classification": "A"}, {} monkeypatch.setattr(counter, "analyze_hardware_counter", fake_analyze) monkeypatch.setattr( counter, "save_counter_artifacts", lambda *_arguments, **_keyword_arguments: events.append("save_processed"), ) result = counter.run_one_hardware_counter( raw_directory=tmp_path / "raw", processed_directory=tmp_path / "processed", ) assert result["processing_status"] == "success" assert configured_gains == [(8, 8)] assert len(transmitted) == 1 assert events.count("transmit") == 1 assert events.index("save_raw") < events.index("analyze") assert result["waveform"]["purpose"] == ( "цифровой счётчик времени, не измерение BER" ) def _reference(count: int = 20_000, start: int = 12_345) -> np.ndarray: return counter.counter_symbol_ids(start, count) def test_prbs20_period_and_unique_state_map() -> None: bits = counter.prbs20_period_bits() positions = counter.bit_state_to_position() assert len(bits) == (1 << 20) - 1 assert positions[0] == -1 assert np.count_nonzero(positions >= 0) == (1 << 20) - 1 assert len(np.unique(positions[positions >= 0])) == (1 << 20) - 1 def test_qpsk_alphabet_has_four_separated_unit_states() -> None: states = counter.QPSK_STATES assert len(states) == 4 assert np.allclose(np.abs(states), 1.0) distances = np.abs(states[:, None] - states[None, :]) assert np.min(distances[distances > 0]) == pytest.approx(np.sqrt(2.0)) def test_hardware_waveform_parameters_are_fixed_before_tx() -> None: waveform, parameters = counter.build_counter_waveform(2_399_998.0) assert counter.SYMBOL_RATE == 20_000.0 assert parameters["samples_per_symbol"] == 120 assert parameters["useful_symbol_count"] == 100_000 assert parameters["transmitted_symbol_count"] == 110_000 assert parameters["margin_symbol_count_each_side"] == 5_000 assert len(waveform) == 13_200_000 assert parameters["useful_duration_seconds"] == pytest.approx(5.00000416667) def test_zero_cfo_exclusion_is_reproduced_and_diagnostic_override_recovers() -> None: waveform, parameters = counter.build_counter_waveform(2_399_998.0) default = counter.recover_counter_from_iq(waveform, 2_399_998.0) diagnostic = counter.recover_counter_from_iq( waveform, 2_399_998.0, minimum_fourth_power_hz=0.0, ) assert default["carrier"]["carrier_offset_hz"] == pytest.approx(44.99, abs=0.1) assert ( default["counter_analysis"]["recovered_symbol_count"] < parameters["transmitted_symbol_count"] ) assert diagnostic["carrier"]["carrier_offset_hz"] == pytest.approx(0.0, abs=0.1) assert ( diagnostic["counter_analysis"]["recovered_symbol_count"] == parameters["transmitted_symbol_count"] ) assert diagnostic["counter_analysis"]["stable_run_count"] == 1 def test_ideal_counter_is_continuous() -> None: result = counter.analyze_counter_symbols(_reference()) assert result["classification"] == "A" assert result["events"] == [] assert result["ordinary_symbol_error_count"] == 0 @pytest.mark.parametrize("deleted", [1, 2, 10]) def test_counter_detects_deleted_symbols(deleted: int) -> None: reference = _reference() received = np.delete(reference, np.s_[8_000 : 8_000 + deleted]) result = counter.analyze_counter_symbols(received) assert result["classification"] == "B" assert any( event["type"] == "пропуск" and event["minimum_confirmed_gap_symbols"] == deleted for event in result["events"] ) @pytest.mark.parametrize("repeated", [1, 4]) def test_counter_detects_repeated_symbols(repeated: int) -> None: reference = _reference() at = 8_000 received = np.insert(reference, at, reference[at - repeated : at]) result = counter.analyze_counter_symbols(received) assert result["classification"] == "B" assert any( event["type"] == "повтор" and event["minimum_confirmed_gap_symbols"] == repeated for event in result["events"] ) def test_counter_detects_foreign_insertion() -> None: reference = _reference() received = np.insert(reference, 8_000, np.asarray([0, 0, 0], dtype=np.uint8)) result = counter.analyze_counter_symbols(received) assert result["classification"] == "B" assert any(event["type"] == "вставка" for event in result["events"]) def test_counter_detects_reordered_neighboring_blocks() -> None: reference = _reference() first = reference[7_000:7_200] second = reference[7_200:7_450] received = np.concatenate((reference[:7_000], second, first, reference[7_450:])) result = counter.analyze_counter_symbols(received) assert result["classification"] == "B" assert len(result["events"]) >= 2 def test_partial_capture_from_middle_locates_position_without_false_jump() -> None: received = counter.counter_symbol_ids(456_789, 12_000) result = counter.analyze_counter_symbols(received) assert result["classification"] == "A" assert result["runs"][0]["first_sequence_position"] == 456_789 def test_capture_ending_early_is_not_a_discontinuity() -> None: reference = _reference() result = counter.analyze_counter_symbols(reference[:9_321]) assert result["classification"] == "A" assert result["events"] == [] def test_single_symbol_error_is_not_a_large_position_jump() -> None: received = _reference() received[8_000] = (received[8_000] + 1) % 4 result = counter.analyze_counter_symbols(received) assert result["classification"] == "A" assert result["events"] == [] assert result["ordinary_symbol_error_count"] == 1 @pytest.mark.parametrize( ("carrier_offset_hz", "clock_error_ppm", "noise"), [ (1_000.0, 0.0, 0.0), (1_000.0, 35.0, 0.0), (1_000.0, 0.0, 0.12), (-750.0, -30.0, 0.10), ], ) def test_iq_recovery_handles_frequency_clock_error_and_noise( carrier_offset_hz: float, clock_error_ppm: float, noise: float, ) -> None: sample_rate = 120_000.0 samples_per_symbol = 12 received = counter.synthesize_counter_iq( start_position=345_678, symbol_count=10_000, samples_per_symbol=samples_per_symbol, sample_rate_hz=sample_rate, carrier_offset_hz=carrier_offset_hz, sample_clock_error_ppm=clock_error_ppm, noise_standard_deviation=noise, ) recovered = counter.recover_counter_from_iq( received, sample_rate, nominal_samples_per_symbol=samples_per_symbol, fft_samples=65_536, minimum_fourth_power_hz=200.0, maximum_fourth_power_hz=10_000.0, ) assert recovered["counter_analysis"]["classification"] == "A" assert recovered["counter_analysis"]["anchor_count"] > 8_000 assert recovered["counter_analysis"]["events"] == [] assert recovered["carrier"]["carrier_offset_hz"] == pytest.approx( carrier_offset_hz, abs=2.0 )