429 lines
15 KiB
Python
429 lines
15 KiB
Python
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
|