turov_av/SDR-Rover
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Lab043: add hardware diagnostics and HackRF comparison tools

Browse Source
This commit is contained in:
LittleSam129
2026-08-21 17:33:57 +03:00
parent f0fa7e8a46
commit 42b0120eea
42 changed files with 8763 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
tests/test_lab043_bpsk_asymmetry_diagnostics.py Normal file
View File

@@ -0,0 +1,78 @@
import numpy as np
from experiments import lab043_bpsk_asymmetry_diagnostics as diagnostics
def test_known_sequence_phase_fit_removes_known_linear_phase() -> None:
bits = np.tile(np.array([0, 0, 1, 1, 0, 1, 0, 1], dtype=np.uint8), 128)
expected = 2.0 * bits.astype(np.float64) - 1.0
indexes = np.arange(len(expected), dtype=np.float64)
frequency_hz = 0.75
received = expected * np.exp(
1j * (0.37 + 2.0 * np.pi * frequency_hz * indexes / 20_000.0)
)
corrected, estimated_hz, fit_rmse = diagnostics._known_sequence_phase_fit(
received, expected, 20_000.0
)
normalized, _ = diagnostics._normalize_symbols(corrected, expected)
assert abs(estimated_hz - frequency_hz) < 1e-9
assert fit_rmse < 1e-12
assert np.max(np.abs(normalized - expected)) < 1e-12
def test_triplet_rows_count_and_classify_only_actual_decision_crossings() -> None:
bits = np.tile(np.array([0, 0, 0, 1, 1, 1, 0, 1], dtype=np.uint8), 20)
values = 2.0 * bits.astype(np.float64) - 1.0
target = np.flatnonzero(
(bits[:-2] == 0) & (bits[1:-1] == 0) & (bits[2:] == 0)
)[0] + 1
values[target] = 0.1
rows = diagnostics.triplet_rows("synthetic", bits, values.astype(np.complex128))
by_triplet = {row["triplet"]: row for row in rows}
assert sum(row["count"] for row in rows) == len(bits) - 2
assert by_triplet["000"]["bit_errors"] == 1
assert sum(row["bit_errors"] for row in rows) == 1
assert by_triplet["000"]["decision_cross_fraction"] == by_triplet["000"]["ber"]
def test_rrc_audit_has_symmetric_unit_energy_pair() -> None:
impulse_rows, audit = diagnostics.rrc_rows()
assert audit["tx_tap_count"] == 1201
assert audit["rx_tap_count"] == 1201
assert audit["symmetry_max_abs_error"] == 0.0
assert abs(audit["tap_energy"] - 1.0) < 1e-12
by_offset = {row["symbol_offset"]: row["h_normalized"] for row in impulse_rows}
assert by_offset[0] == 1.0
for offset in range(1, 6):
assert abs(by_offset[offset] - by_offset[-offset]) < 1e-15
def test_fractional_delay_preserves_order_and_length() -> None:
source = np.array([0.0, 1.0, 2.0, 3.0], dtype=np.complex128)
shifted = diagnostics._fractional_delay(source, 0.5)
assert len(shifted) == len(source)
assert np.allclose(shifted, [0.0, 0.5, 1.5, 2.5])
def test_triplet_metrics_keep_complex_dc_visible() -> None:
bits = np.tile(np.array([0, 0, 1, 1, 0, 1, 0, 1], dtype=np.uint8), 20)
expected = 2.0 * bits.astype(np.float64) - 1.0
offset = 0.2 - 0.1j
rows = diagnostics.triplet_rows(
"affine", bits, expected.astype(np.complex128) + offset
)
negative = [row for row in rows if row["triplet"][1] == "0" and row["count"]]
positive = [row for row in rows if row["triplet"][1] == "1" and row["count"]]
present = [row for row in rows if row["count"]]
assert all(abs(row["mean_i"] + 0.8) < 1e-12 for row in negative)
assert all(abs(row["mean_i"] - 1.2) < 1e-12 for row in positive)
assert all(abs(row["mean_q"] + 0.1) < 1e-12 for row in present)

33
tests/test_lab043_calibration.py
View File

@@ -764,6 +764,39 @@ def test_symbol_sample_phase_is_selected_separately(initial_sample_phase: int) -
assert found["score"] > 0.99
def test_spectral_tone_start_fallback_finds_narrow_tones_without_broadband_jump() -> None:
sample_rate = 240_000.0
rng = np.random.default_rng(43)
samples = (
rng.normal(0.0, 0.08, 240_000)
+ 1j * rng.normal(0.0, 0.08, 240_000)
)
tone_start = 96_000
tone_count = 60_000
indexes = np.arange(tone_count, dtype=np.float64)
samples[tone_start : tone_start + tone_count] += 0.025 * (
np.exp(-1j * 2.0 * np.pi * 50_000.0 * indexes / sample_rate)
+ np.exp(1j * 2.0 * np.pi * 50_000.0 * indexes / sample_rate)
)
with pytest.raises(RuntimeError, match="10"):
lab043.locate_calibration_tone_start(
samples,
sample_rate,
block_samples=2048,
hop_samples=1024,
)
found = lab043.locate_calibration_tone_start_spectral(
samples,
sample_rate,
block_samples=8192,
hop_samples=2048,
required_consecutive_blocks=3,
)
assert tone_start - 8192 <= found <= tone_start + 2048
def test_one_control_packet_passes_crc() -> None:
assert lab043.control_packet_crc_roundtrip(b"known payload")

538
tests/test_lab043_hackrf_counter.py Normal file
View File

@@ -0,0 +1,538 @@
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
)

191
tests/test_lab043_hackrf_gain_ladder.py Normal file
View File

@@ -0,0 +1,191 @@
import numpy as np
import pytest
from experiments import lab043_hackrf_gain_ladder as ladder
def _tone(
sample_rate: float,
seconds: float,
amplitude: float,
noise: float = 0.0,
second: float = 0.0,
third: float = 0.0,
frequency: float = 50_123.0,
) -> np.ndarray:
rng = np.random.default_rng(43043)
count = round(sample_rate * seconds)
indexes = np.arange(count, dtype=np.float64)
result = amplitude * np.exp(1j * 2.0 * np.pi * frequency * indexes / sample_rate)
result += second * np.exp(1j * 2.0 * np.pi * 2.0 * frequency * indexes / sample_rate)
result += third * np.exp(1j * 2.0 * np.pi * 3.0 * frequency * indexes / sample_rate)
if noise:
result += noise * (rng.normal(size=count) + 1j * rng.normal(size=count))
return result.astype(np.complex64)
def test_fixed_gain_ladder_is_valid_and_monotonic() -> None:
assert ladder.GAIN_LADDER == (
(0, 0), (8, 8), (16, 8), (16, 16), (24, 16), (24, 24)
)
assert [sum(point) for point in ladder.GAIN_LADDER] == sorted(
sum(point) for point in ladder.GAIN_LADDER
)
def test_measure_tone_finds_frequency_margin_and_no_overload() -> None:
sample_rate = 240_000.0
samples = _tone(sample_rate, 3.0, amplitude=0.20, noise=0.002)
metrics = ladder.measure_tone_segment(samples, sample_rate, fft_samples=65_536)
assert metrics["tone_frequency_hz"] == pytest.approx(50_123.0, abs=2.0)
assert metrics["tone_over_noise_db"] > 35.0
assert metrics["rms_amplitude"] == pytest.approx(0.20, abs=0.01)
assert metrics["clipped_component_fraction"] == 0.0
assert not metrics["second_harmonic"]["credible"]
assert not metrics["third_harmonic"]["credible"]
assert not metrics["stop_gain_escalation"]
def test_strong_tone_without_harmonics_is_not_nonlinear() -> None:
sample_rate = 240_000.0
samples = _tone(sample_rate, 3.0, amplitude=0.55, noise=0.002)
metrics = ladder.measure_tone_segment(samples, sample_rate, fft_samples=65_536)
assert not metrics["second_harmonic"]["credible"]
assert not metrics["third_harmonic"]["credible"]
assert not metrics["harmonic_nonlinearity_detected"]
def test_credible_second_harmonic_is_detected() -> None:
sample_rate = 240_000.0
samples = _tone(sample_rate, 3.0, amplitude=0.30, noise=0.001, second=0.03)
metrics = ladder.measure_tone_segment(samples, sample_rate, fft_samples=65_536)
assert metrics["second_harmonic"]["credible"]
assert metrics["second_harmonic"]["nonlinear"]
assert metrics["second_harmonic"]["relative_to_fundamental_db"] == pytest.approx(-20.0, abs=1.0)
assert metrics["stop_gain_escalation"]
def test_credible_third_harmonic_is_detected() -> None:
sample_rate = 240_000.0
samples = _tone(sample_rate, 3.0, amplitude=0.30, noise=0.001, third=0.015)
metrics = ladder.measure_tone_segment(samples, sample_rate, fft_samples=65_536)
assert metrics["third_harmonic"]["credible"]
assert metrics["third_harmonic"]["nonlinear"]
assert metrics["third_harmonic"]["relative_to_fundamental_db"] == pytest.approx(-26.0, abs=1.0)
assert metrics["stop_gain_escalation"]
def test_peak_high_relative_to_weak_tone_but_below_local_margin_is_not_credible() -> None:
sample_rate = 1_000_000.0
fft_samples = 10_000
frequencies = np.fft.fftfreq(fft_samples, d=1.0 / sample_rate)
powers = np.full(fft_samples, 1e-5, dtype=np.float64)
expected = 100_000.0
peak_index = int(np.argmin(np.abs(frequencies - expected)))
powers[peak_index] = 3e-5
result = ladder.analyze_harmonic(
powers,
frequencies,
fundamental_frequency_hz=50_000.0,
fundamental_power=1e-6,
order=2,
sample_rate_hz=sample_rate,
)
assert result["relative_to_fundamental_db"] > 10.0
assert result["above_local_noise_db"] < ladder.HARMONIC_DETECTION_MARGIN_DB
assert not result["credible"]
assert not result["nonlinear"]
def test_measure_tone_stops_on_near_limit_samples() -> None:
sample_rate = 240_000.0
samples = _tone(sample_rate, 2.0, amplitude=0.95)
metrics = ladder.measure_tone_segment(samples, sample_rate, fft_samples=65_536)
assert metrics["near_limit_component_fraction"] > 0.0
assert metrics["stop_gain_escalation"]
def test_harmonic_frequency_aliases_into_nyquist_band() -> None:
assert ladder.alias_frequency_to_nyquist(900_000.0, 1_000_000.0) == pytest.approx(-100_000.0)
assert ladder.alias_frequency_to_nyquist(1_350_000.0, 1_000_000.0) == pytest.approx(350_000.0)
sample_rate = 1_000_000.0
fft_samples = 100_000
frequencies = np.fft.fftfreq(fft_samples, d=1.0 / sample_rate)
powers = np.full(fft_samples, 1e-10, dtype=np.float64)
powers[int(np.argmin(np.abs(frequencies + 100_000.0)))] = 1e-3
powers[int(np.argmin(np.abs(frequencies - 350_000.0)))] = 2.5e-4
second = ladder.analyze_harmonic(
powers, frequencies, 450_000.0, 0.1, 2, sample_rate
)
third = ladder.analyze_harmonic(
powers, frequencies, 450_000.0, 0.1, 3, sample_rate
)
assert second["expected_frequency_hz"] == pytest.approx(-100_000.0)
assert third["expected_frequency_hz"] == pytest.approx(350_000.0)
assert second["credible"]
assert third["credible"]
def test_choose_preferred_gain_uses_lowest_gain_reaching_desirable_margin() -> None:
def row(index: int, lna: int, vga: int, margin: float) -> dict:
return {
"step_index": index,
"lna_gain_db": lna,
"vga_gain_db": vga,
"metrics": {
"tone_over_noise_db": margin,
"maximum_component": 0.4,
"stop_gain_escalation": False,
},
}
rows = [row(0, 0, 0, 20.0), row(1, 8, 8, 38.5), row(2, 16, 8, 40.0)]
selected = ladder.choose_preferred_gain(rows)
assert selected is rows[1]
def test_choose_preferred_gain_rejects_unsafe_rows() -> None:
row = {
"step_index": 0,
"lna_gain_db": 0,
"vga_gain_db": 0,
"metrics": {
"tone_over_noise_db": 60.0,
"maximum_component": 0.95,
"stop_gain_escalation": True,
},
}
assert ladder.choose_preferred_gain([row]) is None
def test_choose_preferred_gain_uses_near_best_when_target_not_reached() -> None:
def row(index: int, lna: int, vga: int, margin: float) -> dict:
return {
"step_index": index,
"lna_gain_db": lna,
"vga_gain_db": vga,
"metrics": {
"tone_over_noise_db": margin,
"maximum_component": 0.4,
"stop_gain_escalation": False,
},
}
rows = [row(0, 0, 0, 20.0), row(1, 8, 8, 31.5), row(2, 16, 8, 33.0)]
assert ladder.choose_preferred_gain(rows) is rows[1]

203
tests/test_lab043_hackrf_hardware.py Normal file
View File

@@ -0,0 +1,203 @@
import ctypes
import numpy as np
import pytest
from experiments import lab043_hackrf_hardware as hackrf
from experiments import lab043_hackrf_continuity as continuity
def _block(first: int, complex_samples: int) -> bytes:
values = (np.arange(2 * complex_samples, dtype=np.int16) + first) % 255 - 127
return values.astype(np.int8).tobytes()
def test_signed_iq_bytes_to_complex64_preserves_iq_order() -> None:
raw = np.asarray([-128, 127, -64, 32], dtype=np.int8).tobytes()
result = hackrf.signed_iq_bytes_to_complex64(raw)
assert result.dtype == np.complex64
assert np.allclose(result, [-1.0 + 127j / 128.0, -0.5 + 0.25j])
def test_signed_iq_bytes_rejects_incomplete_complex_sample() -> None:
with pytest.raises(ValueError, match="чётное"):
hackrf.signed_iq_bytes_to_complex64(b"\x01")
def test_callback_copy_uses_only_valid_length_and_survives_source_reuse() -> None:
source = (ctypes.c_uint8 * 8)(1, 2, 3, 4, 90, 91, 92, 93)
copied = hackrf.copy_valid_callback_bytes(source, buffer_length=8, valid_length=4)
for index in range(8):
source[index] = 0
assert copied == b"\x01\x02\x03\x04"
def test_callback_copy_rejects_odd_valid_length() -> None:
source = (ctypes.c_uint8 * 4)(1, 2, 3, 4)
with pytest.raises(ValueError, match="чётным"):
hackrf.copy_valid_callback_bytes(source, buffer_length=4, valid_length=3)
def test_callback_assembler_discards_fixed_warmup_and_keeps_order() -> None:
blocks = [_block(0, 4), _block(20, 4), _block(40, 4), _block(60, 4)]
samples, boundaries, diagnostics = hackrf.assemble_callback_blocks(
blocks, target_sample_count=9, warmup_callback_count=1
)
expected = hackrf.signed_iq_bytes_to_complex64(
blocks[1] + blocks[2] + blocks[3][:2]
)
assert np.array_equal(samples, expected)
assert [row.callback_index for row in boundaries] == [1, 2, 3]
assert [row.accepted_end_sample for row in boundaries] == [4, 8, 9]
assert [row.accepted_bytes for row in boundaries] == [8, 8, 2]
assert [row.buffer_length for row in boundaries] == [8, 8, 8]
assert diagnostics["callback_indexes_sequential"]
assert diagnostics["program_queue_missing_callback_count"] == 0
assert diagnostics["accepted_sample_count"] == 9
def test_callback_assembler_keeps_distinct_blocks_without_duplication() -> None:
blocks = [b"\x01\x02\x03\x04", b"\x05\x06\x07\x08"]
samples, boundaries, diagnostics = hackrf.assemble_callback_blocks(
blocks,
target_sample_count=4,
warmup_callback_count=0,
buffer_lengths=[16, 16],
)
expected = hackrf.signed_iq_bytes_to_complex64(blocks[0] + blocks[1])
assert np.array_equal(samples, expected)
assert [row.callback_index for row in boundaries] == [0, 1]
assert [row.valid_bytes for row in boundaries] == [4, 4]
assert [row.buffer_length for row in boundaries] == [16, 16]
assert diagnostics["accepted_bytes"] == sum(map(len, blocks))
def test_capture_contract_keeps_original_int8_bytes_separate_from_complex64() -> None:
raw = np.asarray([-128, 127, -64, 32], dtype=np.int8).tobytes()
capture = hackrf.HackrfCapture(
samples=hackrf.signed_iq_bytes_to_complex64(raw),
boundaries=(),
diagnostics={},
raw_bytes=raw,
)
assert capture.raw_bytes == raw
assert capture.samples.dtype == np.complex64
assert np.array_equal(
capture.samples,
hackrf.signed_iq_bytes_to_complex64(capture.raw_bytes),
)
@pytest.mark.parametrize("timeout_seconds", [0.0, -1.0, float("nan")])
def test_capture_rejects_invalid_timeout_before_hardware_access(
timeout_seconds: float,
) -> None:
receiver = hackrf.HackrfRxDevice()
with pytest.raises(ValueError, match="Тайм-аут"):
receiver.capture(1, timeout_seconds=timeout_seconds)
def test_callback_assembler_rejects_short_stream() -> None:
with pytest.raises(RuntimeError, match="короче"):
hackrf.assemble_callback_blocks(
[_block(0, 4), _block(20, 4)],
target_sample_count=9,
warmup_callback_count=1,
)
def test_callback_assembler_rejects_odd_callback() -> None:
with pytest.raises(ValueError, match="нечётное"):
hackrf.assemble_callback_blocks(
[b"\x00\x01", b"\x02"],
target_sample_count=1,
warmup_callback_count=1,
)
def test_hackrf_configuration_is_fixed_and_rx_only() -> None:
assert hackrf.HACKRF_CENTER_HZ == 435_000_000
assert hackrf.HACKRF_SAMPLE_RATE_HZ == 2_400_000.0
assert hackrf.HACKRF_RF_AMP == 0
assert hackrf.HACKRF_ANTENNA_BIAS == 0
assert hackrf.HACKRF_LNA_GAIN_DB == 8
assert hackrf.HACKRF_VGA_GAIN_DB == 8
assert not hasattr(hackrf.HackrfRxDevice, "transmit")
@pytest.mark.parametrize(
("lna", "vga"),
[(0, 0), (8, 8), (16, 8), (16, 16), (24, 16), (24, 24), (40, 62)],
)
def test_hackrf_rx_gain_steps_accept_official_values(lna: int, vga: int) -> None:
assert hackrf.validate_hackrf_rx_gains(lna, vga) == (lna, vga)
@pytest.mark.parametrize("lna", [-8, 4, 48])
def test_hackrf_rx_gain_steps_reject_invalid_lna(lna: int) -> None:
with pytest.raises(ValueError, match="LNA"):
hackrf.validate_hackrf_rx_gains(lna, 0)
@pytest.mark.parametrize("vga", [-2, 1, 64])
def test_hackrf_rx_gain_steps_reject_invalid_vga(vga: int) -> None:
with pytest.raises(ValueError, match="VGA"):
hackrf.validate_hackrf_rx_gains(0, vga)
def test_periodic_continuity_tone_has_identical_wrap_phase() -> None:
tone, frequency = continuity.build_periodic_tone(2_400_000.0)
phase_step = np.exp(1j * 2.0 * np.pi * frequency / 2_400_000.0)
assert len(tone) == 48 * continuity.CYCLIC_PERIOD_COUNT
assert np.allclose(tone[0], tone[-1] * phase_step, atol=2e-6)
def test_boundary_analysis_finds_known_phase_step_at_callback_boundary() -> None:
sample_rate = 240_000.0
frequency = 5_000.0
sample_count = 240_000
boundary_sample = sample_count // 2
indexes = np.arange(sample_count, dtype=np.float64)
samples = np.exp(1j * 2.0 * np.pi * frequency * indexes / sample_rate)
samples[boundary_sample:] *= np.exp(0.8j)
boundary = hackrf.HackrfBoundary(
callback_index=1,
valid_bytes=2 * boundary_sample,
accepted_bytes=2 * boundary_sample,
accepted_start_sample=0,
accepted_end_sample=boundary_sample,
monotonic_seconds=0.5,
block_sha256="synthetic",
buffer_length=2 * boundary_sample,
)
rows, summary = continuity.analyze_callback_boundaries(
samples.astype(np.complex64),
(boundary,),
sample_rate,
frequency,
stable_start_sample=0,
stable_end_sample=sample_count,
side_samples=32_768,
block_samples=8_192,
hop_samples=2_048,
)
assert summary["confirmed_jump_count"] == 1
assert rows[0]["additional_phase_jump_rad"] == pytest.approx(0.8, abs=0.01)
assert rows[0]["equivalent_sample_shift_principal"] == pytest.approx(
0.8 / (2.0 * np.pi * frequency / sample_rate), abs=0.05
)
assert rows[0]["relation_to_callback_boundary"] == "consistent_with_exact_boundary"

428
tests/test_lab043_hackrf_zero_if_compare.py Normal file
View File

@@ -0,0 +1,428 @@
import ast
import inspect
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_hackrf_zero_if_compare as comparison
from experiments import lab043_prbs11_hardware as pluto_hardware
def test_comparison_uses_dedicated_output_directory() -> None:
assert comparison.PROCESSED_DIRECTORY == Path(
"data/processed/lab043_hackrf_zero_if_compare"
)
def test_composite_waveform_has_seven_fixed_sections_and_identical_counters(
monkeypatch: pytest.MonkeyPatch,
) -> None:
sample_rate = 240_000.0
base_counter = np.asarray(
[0.35 + 0.0j, 0.0 + 0.35j, -0.35 + 0.0j, 0.0 - 0.35j],
dtype=np.complex64,
)
monkeypatch.setattr(
counter,
"build_counter_waveform",
lambda _sample_rate: (base_counter.copy(), {"samples_per_symbol": 120}),
)
waveform, metadata = comparison.build_composite_waveform(sample_rate)
names = [row["name"] for row in metadata["sections"]]
assert names == [
"leading_zero",
"tone_plus_50khz",
"guard_before_shifted_counter",
"counter_plus_50khz",
"guard_after_shifted_counter",
"counter_zero_if",
"trailing_zero",
]
sections = {row["name"]: row for row in metadata["sections"]}
shifted_row = sections["counter_plus_50khz"]
centered_row = sections["counter_zero_if"]
shifted = waveform[
shifted_row["start_sample"] : shifted_row["end_sample_exclusive"]
]
centered = waveform[
centered_row["start_sample"] : centered_row["end_sample_exclusive"]
]
indexes = np.arange(len(shifted), dtype=np.float64)
restored = shifted * np.exp(
-2j * np.pi * comparison.TONE_OFFSET_HZ * indexes / sample_rate
)
assert np.allclose(restored, centered, atol=2e-7)
assert metadata["digital_sequences_bit_identical"] is True
assert (
metadata["counter_symbol_ids_sha256_shifted"]
== metadata["counter_symbol_ids_sha256_centered"]
)
assert metadata["counter_amplitude"] == metadata["tone_amplitude"] == 0.35
for name in (
"leading_zero",
"guard_before_shifted_counter",
"guard_after_shifted_counter",
"trailing_zero",
):
row = sections[name]
assert np.count_nonzero(waveform[row["start_sample"] : row["end_sample_exclusive"]]) == 0
def test_pluto_scaling_is_in_place_and_does_not_clip() -> None:
waveform = np.asarray([0.35 + 0.0j, 0.0 - 0.35j], dtype=np.complex64)
identity = id(waveform)
result = comparison.scale_pluto_samples_in_place(waveform)
assert id(result) == identity
assert result.dtype == np.complex64
assert np.max(np.abs(result)) == pytest.approx(0.35 * 2**14, rel=1e-6)
def test_pluto_validation_includes_rf_bandwidth() -> None:
parameters = {
"actual_center_frequency_hz": 435_000_000,
"actual_sample_rate_hz": 2_400_000,
"actual_gain_db": -30.0,
"actual_rf_bandwidth_hz": 200_000,
}
comparison.validate_pluto_working_point(parameters)
parameters["actual_rf_bandwidth_hz"] = 100_000
with pytest.raises(RuntimeError, match="RF bandwidth"):
comparison.validate_pluto_working_point(parameters)
def test_dual_raw_is_hashed_reread_and_exactly_converted(tmp_path: Path) -> None:
raw = np.asarray([-128, 127, -64, 32, 0, -1], dtype=np.int8).tobytes()
samples = hackrf.signed_iq_bytes_to_complex64(raw)
int8_path, iq_path, metadata_path, document = comparison.save_dual_raw_before_processing(
tmp_path / "capture",
raw,
samples,
{"capture_status": "captured", "processing_status": "pending"},
)
assert int8_path.read_bytes() == raw
assert np.array_equal(np.load(iq_path, allow_pickle=False), samples)
assert document["raw_verified"] is True
assert document["int8_to_complex64"]["verified"] is True
assert document["int8_to_complex64"]["byte_order"] == "interleaved I then Q"
assert document["int8_sha256"] == comparison._hash_file(int8_path)
assert document["iq_sha256"] == comparison._hash_file(iq_path)
assert metadata_path.exists()
def test_background_metrics_keep_dc_and_signed_code_distributions() -> None:
codes = np.asarray([[-2, 4], [-2, 4], [-1, 3], [-2, 4]], dtype=np.int8)
samples = hackrf.signed_iq_bytes_to_complex64(codes.tobytes())
metrics = comparison.background_metrics(samples, codes, 2_400_000.0)
assert metrics["i_code_distribution"] == {"-2": 3, "-1": 1}
assert metrics["q_code_distribution"] == {"3": 1, "4": 3}
assert metrics["mean_i"] == pytest.approx(-1.75 / 128.0)
assert metrics["mean_q"] == pytest.approx(3.75 / 128.0)
assert metrics["dc_correction_applied"] is False
def test_known_counter_analysis_recovers_shifted_synthetic_without_dc_correction(
monkeypatch: pytest.MonkeyPatch,
) -> None:
sample_rate = 240_000.0
symbol_count = 1_000
samples_per_symbol = 120
carrier_hz = 50_200.0
expected_start = 80_037
monkeypatch.setattr(counter, "TRANSMITTED_SYMBOL_COUNT", symbol_count)
monkeypatch.setattr(comparison, "MATCH_REFERENCE_START_SYMBOL", 100)
monkeypatch.setattr(comparison, "MATCH_REFERENCE_SYMBOL_COUNT", 300)
symbols = counter.counter_qpsk_symbols(0, symbol_count)
base = np.repeat(symbols, samples_per_symbol)
indexes = np.arange(len(base), dtype=np.float64)
signal = (0.02 * base * np.exp(2j * np.pi * carrier_hz * indexes / sample_rate)).astype(
np.complex64
)
raw = np.full(320_000, 0.01 + 0.02j, dtype=np.complex64)
raw[expected_start : expected_start + len(signal)] += signal
result = comparison.analyze_counter_section(
raw,
expected_start + 40,
carrier_hz,
sample_rate,
background_power=0.01**2 + 0.02**2,
)
assert result["start_sample"] == expected_start
assert result["measured_carrier_hz"] == pytest.approx(carrier_hz, abs=0.1)
assert result["symbol_error_rate"] == 0.0
assert result["dc_correction_applied"] is False
assert result["new_filter_applied"] is False
assert result["equalizer_applied"] is False
def test_hardware_runner_contains_one_tx_call_and_no_forbidden_experiment_calls() -> None:
tree = ast.parse(inspect.getsource(comparison.run_one_zero_if_comparison))
calls = [node for node in ast.walk(tree) if isinstance(node, ast.Call)]
attributes = [node.func.attr for node in calls if isinstance(node.func, ast.Attribute)]
assert attributes.count("transmit_noncyclic_buffer_once") == 1
forbidden = {
"run_one_hackrf_s",
"run_one_hardware_counter",
"run_one_hardware_prbs11",
"run_packet",
"run_jpeg",
}
assert forbidden.isdisjoint(attributes)
def test_hardware_orchestration_transmits_once_and_saves_both_raw_before_analysis(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
events: list[str] = []
transmitted: list[np.ndarray] = []
class FakeLibrary:
@staticmethod
def hackrf_is_streaming(_device: object) -> int:
return 1
class FakeReceiver:
library = FakeLibrary()
device = object()
def __enter__(self):
return self
def __exit__(self, *_arguments: object) -> None:
return None
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:
assert (lna_gain_db, vga_gain_db) == (8, 8)
return {
"rf_amp_enabled": False,
"antenna_bias_enabled": False,
"lna_gain_db": lna_gain_db,
"vga_gain_db": vga_gain_db,
}
def capture(self, target_sample_count: int, timeout_seconds: float):
events.append("capture")
assert timeout_seconds == comparison.RX_TIMEOUT_SECONDS
raw = bytes(2 * target_sample_count)
return SimpleNamespace(
samples=hackrf.signed_iq_bytes_to_complex64(raw),
raw_bytes=raw,
boundaries=(),
diagnostics={
"callback_count_total": 1,
"stream_started_monotonic": 0.0,
},
)
monkeypatch.setattr(comparison, "RX_LEADING_SECONDS", 0.0)
monkeypatch.setattr(comparison, "RX_TRAILING_SECONDS", 0.0)
monkeypatch.setattr(comparison, "TX_UPLOAD_MARGIN_SECONDS", 0.0)
monkeypatch.setattr(hackrf, "HACKRF_EXPECTED_CALLBACK_BYTES", 0)
monkeypatch.setattr(hackrf, "HackrfRxDevice", FakeReceiver)
monkeypatch.setattr(
hackrf,
"preflight_hackrf",
lambda lna, vga: {
"set_center_frequency_hz": 435_000_000,
"set_sample_rate_hz": 2_400_000.0,
"rf_amp_enabled": False,
"lna_gain_db": lna,
"vga_gain_db": vga,
"antenna_bias_enabled": False,
"serial_number": "serial",
"firmware_version": "firmware",
"libhackrf_version": "library",
},
)
monkeypatch.setattr(pluto_hardware, "discover_pluto_usb_uri", lambda: "usb:1.2.3")
monkeypatch.setattr(
pluto_hardware,
"preflight_pluto",
lambda uri: {"uri": uri, "context_opened": True},
)
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,
"actual_rf_bandwidth_hz": 200_000,
},
),
)
monkeypatch.setattr(
comparison,
"build_composite_waveform",
lambda _sample_rate: (
np.asarray([0.1 + 0.1j] * 4, dtype=np.complex64),
{"total_duration_seconds": 4 / 2_400_000.0, "sections": [], "total_sample_count": 4},
),
)
def fake_transmit(_device: object, samples: np.ndarray, _sample_rate: float) -> float:
events.append("transmit")
transmitted.append(np.asarray(samples).copy())
return 0.0
monkeypatch.setattr(pluto_hardware, "transmit_noncyclic_buffer_once", fake_transmit)
real_save = comparison.save_dual_raw_before_processing
def save_raw(*arguments: object, **keyword_arguments: object):
events.append("save_raw")
return real_save(*arguments, **keyword_arguments)
monkeypatch.setattr(comparison, "save_dual_raw_before_processing", save_raw)
def analyze(*_arguments: object) -> dict:
events.append("analyze")
return {
"counter_plus_50khz": {},
"counter_zero_if": {},
}
monkeypatch.setattr(comparison, "analyze_saved_capture", analyze)
monkeypatch.setattr(comparison, "save_processed_artifacts", lambda *_arguments: ())
result = comparison.run_one_zero_if_comparison(tmp_path, tmp_path / "processed")
assert result["processing_status"] == "success"
assert events.count("transmit") == 1
assert len(transmitted) == 1
assert events.index("save_raw") < events.index("analyze")
assert Path(result["int8_path"]).suffix == ".int8"
assert Path(result["iq_path"]).suffix == ".npy"
def test_rx_failure_before_stream_confirmation_prevents_tx(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
transmit_calls = 0
class FailedLibrary:
@staticmethod
def hackrf_is_streaming(_device: object) -> int:
return 0
class FailedReceiver:
library = FailedLibrary()
device = object()
def __enter__(self):
return self
def __exit__(self, *_arguments: object) -> None:
return None
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:
return {
"rf_amp_enabled": False,
"antenna_bias_enabled": False,
"lna_gain_db": lna_gain_db,
"vga_gain_db": vga_gain_db,
}
@staticmethod
def capture(_target_sample_count: int, timeout_seconds: float):
raise RuntimeError(f"synthetic RX failure before streaming ({timeout_seconds})")
monkeypatch.setattr(hackrf, "HackrfRxDevice", FailedReceiver)
monkeypatch.setattr(
hackrf,
"preflight_hackrf",
lambda lna, vga: {
"set_center_frequency_hz": 435_000_000,
"set_sample_rate_hz": 2_400_000.0,
"rf_amp_enabled": False,
"lna_gain_db": lna,
"vga_gain_db": vga,
"antenna_bias_enabled": False,
"serial_number": "serial",
"firmware_version": "firmware",
"libhackrf_version": "library",
},
)
monkeypatch.setattr(pluto_hardware, "discover_pluto_usb_uri", lambda: "usb:1.2.3")
monkeypatch.setattr(
pluto_hardware,
"preflight_pluto",
lambda uri: {"uri": uri, "context_opened": True},
)
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,
"actual_rf_bandwidth_hz": 200_000,
},
),
)
monkeypatch.setattr(
comparison,
"build_composite_waveform",
lambda _sample_rate: (
np.asarray([0.1 + 0.1j], dtype=np.complex64),
{"total_duration_seconds": 1 / 2_400_000.0},
),
)
def forbidden_transmit(*_arguments: object) -> float:
nonlocal transmit_calls
transmit_calls += 1
return 0.0
monkeypatch.setattr(
pluto_hardware,
"transmit_noncyclic_buffer_once",
forbidden_transmit,
)
with pytest.raises(RuntimeError, match="до TX; Pluto TX не включён"):
comparison.run_one_zero_if_comparison(tmp_path, tmp_path / "processed")
assert transmit_calls == 0