Lab043: establish reproducible HackRF BPSK working point
This commit is contained in:
LittleSam129
511
tests/test_lab043_hackrf_bpsk.py
Normal file
511
tests/test_lab043_hackrf_bpsk.py
Normal file
@@ -0,0 +1,511 @@
|
||||
"""Проверки fail-closed разбора CLKIN для Lab043."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from experiments import lab043_hackrf_bpsk as bpsk
|
||||
|
||||
|
||||
class FakeCommandRunner:
|
||||
"""Предсказуемый HackRF host без обращения к оборудованию."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
clkin_output: str | list[str] = "CLKIN status: clock signal detected\n",
|
||||
tx_returncode: int = 0,
|
||||
clkin_exception_at: int | None = None,
|
||||
) -> None:
|
||||
self.clkin_outputs = [clkin_output] if isinstance(clkin_output, str) else list(clkin_output)
|
||||
self.tx_returncode = tx_returncode
|
||||
self.clkin_exception_at = clkin_exception_at
|
||||
self.clkin_queries = 0
|
||||
self.commands: list[list[str]] = []
|
||||
|
||||
def __call__(self, command: list[str]) -> subprocess.CompletedProcess:
|
||||
self.commands.append(command)
|
||||
executable = Path(command[0]).name
|
||||
if executable == "hackrf_debug.exe":
|
||||
return subprocess.CompletedProcess(command, 0, "[ 29] -> 0x26\n", "")
|
||||
if executable == "hackrf_clock.exe" and "-i" in command:
|
||||
self.clkin_queries += 1
|
||||
if self.clkin_exception_at == self.clkin_queries:
|
||||
raise subprocess.SubprocessError("mock clock query failure")
|
||||
index = min(self.clkin_queries - 1, len(self.clkin_outputs) - 1)
|
||||
return subprocess.CompletedProcess(command, 0, self.clkin_outputs[index], "")
|
||||
if executable == "hackrf_transfer.exe" and "-t" in command:
|
||||
return subprocess.CompletedProcess(command, self.tx_returncode, "", "")
|
||||
return subprocess.CompletedProcess(command, 0, "", "")
|
||||
|
||||
|
||||
class FakeClock:
|
||||
"""Монотонные часы, которые продвигаются только через sleep."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.seconds = 0.0
|
||||
self.sleeps: list[float] = []
|
||||
|
||||
def monotonic(self) -> float:
|
||||
return self.seconds
|
||||
|
||||
def sleep(self, seconds: float) -> None:
|
||||
self.sleeps.append(seconds)
|
||||
self.seconds += seconds
|
||||
|
||||
|
||||
class FakeRunningCommand:
|
||||
def __init__(self, startup_failed: bool = False, returncode: int = 0) -> None:
|
||||
self.startup_failed = startup_failed
|
||||
self.returncode = returncode
|
||||
self.terminated = False
|
||||
self.logs_closed = False
|
||||
|
||||
def poll(self) -> int | None:
|
||||
if self.startup_failed:
|
||||
return self.returncode or 1
|
||||
return self.returncode if self.terminated else None
|
||||
|
||||
def wait(self, timeout: float) -> int:
|
||||
del timeout
|
||||
self.logs_closed = True
|
||||
return self.returncode
|
||||
|
||||
def terminate(self) -> None:
|
||||
self.terminated = True
|
||||
self.logs_closed = True
|
||||
|
||||
def close_logs(self) -> None:
|
||||
self.logs_closed = True
|
||||
|
||||
|
||||
class FakeProcessStarter:
|
||||
def __init__(self, startup_failed: bool = False, returncode: int = 0) -> None:
|
||||
self.startup_failed = startup_failed
|
||||
self.returncode = returncode
|
||||
self.processes: list[FakeRunningCommand] = []
|
||||
|
||||
def __call__(self, command: list[str], stdout_path: Path, stderr_path: Path) -> FakeRunningCommand:
|
||||
stdout_path.write_text("0 overruns\n", encoding="utf-8")
|
||||
stderr_path.write_text("", encoding="utf-8")
|
||||
raw_path = Path(command[command.index("-r") + 1])
|
||||
raw_path.write_bytes(bytes(2 * bpsk.CAPTURE_SAMPLE_COUNT))
|
||||
process = FakeRunningCommand(self.startup_failed, self.returncode)
|
||||
self.processes.append(process)
|
||||
return process
|
||||
|
||||
|
||||
def fake_analysis() -> dict:
|
||||
return {
|
||||
"timing": {"refined_start_sample": 2},
|
||||
"detection": {"preamble_detected": True},
|
||||
"official_ber": 0.0,
|
||||
"symbol_snr_db": 10.0,
|
||||
"separation": 3.0,
|
||||
"channel": {"abs_h": 0.01},
|
||||
"raw": {
|
||||
"clipping_count": 0,
|
||||
"near_int8_count": 0,
|
||||
"adc_headroom_code": 100.0,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def test_parse_clkin_status_detected() -> None:
|
||||
assert (
|
||||
bpsk.parse_clkin_status("CLKIN status: clock signal detected\n")
|
||||
== bpsk.CLKIN_STATUS_DETECTED
|
||||
)
|
||||
|
||||
|
||||
def test_parse_clkin_status_not_detected() -> None:
|
||||
assert (
|
||||
bpsk.parse_clkin_status("CLKIN status: no clock signal detected\n")
|
||||
== bpsk.CLKIN_STATUS_NOT_DETECTED
|
||||
)
|
||||
|
||||
|
||||
def test_parse_clkin_status_empty_output_is_unknown() -> None:
|
||||
assert bpsk.parse_clkin_status("") == bpsk.CLKIN_STATUS_UNKNOWN
|
||||
|
||||
|
||||
def test_parse_clkin_status_unexpected_output_is_unknown() -> None:
|
||||
assert bpsk.parse_clkin_status("CLKIN status unavailable") == bpsk.CLKIN_STATUS_UNKNOWN
|
||||
|
||||
|
||||
def test_wait_for_slave_clock_detected_on_first_query() -> None:
|
||||
runner = FakeCommandRunner()
|
||||
clock = FakeClock()
|
||||
|
||||
result = bpsk.wait_for_slave_clock(
|
||||
runner, sleep_function=clock.sleep, monotonic_function=clock.monotonic
|
||||
)
|
||||
|
||||
assert result["state"] == bpsk.CLKIN_STATUS_DETECTED
|
||||
assert result["queries"] == 1
|
||||
assert result["time_to_detect_ms"] == pytest.approx(50.0)
|
||||
|
||||
|
||||
def test_wait_for_slave_clock_delayed_detected() -> None:
|
||||
runner = FakeCommandRunner([
|
||||
"CLKIN status: no clock signal detected\n",
|
||||
"CLKIN status: no clock signal detected\n",
|
||||
"CLKIN status: clock signal detected\n",
|
||||
])
|
||||
clock = FakeClock()
|
||||
|
||||
result = bpsk.wait_for_slave_clock(
|
||||
runner, sleep_function=clock.sleep, monotonic_function=clock.monotonic
|
||||
)
|
||||
|
||||
assert result["state"] == bpsk.CLKIN_STATUS_DETECTED
|
||||
assert [poll["state"] for poll in result["polls"]] == [
|
||||
bpsk.CLKIN_STATUS_NOT_DETECTED,
|
||||
bpsk.CLKIN_STATUS_NOT_DETECTED,
|
||||
bpsk.CLKIN_STATUS_DETECTED,
|
||||
]
|
||||
|
||||
|
||||
def test_wait_for_slave_clock_timeout_is_bounded_and_fail_closed() -> None:
|
||||
runner = FakeCommandRunner("CLKIN status: no clock signal detected\n")
|
||||
clock = FakeClock()
|
||||
guard = bpsk.TransmitGuard()
|
||||
|
||||
with pytest.raises(RuntimeError, match="TIMEOUT"):
|
||||
bpsk.common_clock_preflight(
|
||||
runner,
|
||||
guard,
|
||||
sleep_function=clock.sleep,
|
||||
monotonic_function=clock.monotonic,
|
||||
)
|
||||
|
||||
assert runner.clkin_queries == 10
|
||||
assert clock.seconds == pytest.approx(bpsk.CLOCK_TIMEOUT_SECONDS)
|
||||
assert guard.transmit_calls == 0
|
||||
assert guard.tx_allowed is False
|
||||
|
||||
|
||||
def test_wait_for_slave_clock_unknown_is_immediate_fail_closed() -> None:
|
||||
runner = FakeCommandRunner("CLKIN status unavailable\n")
|
||||
clock = FakeClock()
|
||||
guard = bpsk.TransmitGuard()
|
||||
|
||||
with pytest.raises(RuntimeError, match="UNKNOWN"):
|
||||
bpsk.common_clock_preflight(
|
||||
runner,
|
||||
guard,
|
||||
sleep_function=clock.sleep,
|
||||
monotonic_function=clock.monotonic,
|
||||
)
|
||||
|
||||
assert runner.clkin_queries == 1
|
||||
assert guard.transmit_calls == 0
|
||||
|
||||
|
||||
def test_wait_for_slave_clock_subprocess_exception_is_fail_closed() -> None:
|
||||
runner = FakeCommandRunner(clkin_exception_at=1)
|
||||
clock = FakeClock()
|
||||
guard = bpsk.TransmitGuard()
|
||||
|
||||
with pytest.raises(RuntimeError, match="ERROR"):
|
||||
bpsk.common_clock_preflight(
|
||||
runner,
|
||||
guard,
|
||||
sleep_function=clock.sleep,
|
||||
monotonic_function=clock.monotonic,
|
||||
)
|
||||
|
||||
assert guard.transmit_calls == 0
|
||||
assert guard.tx_allowed is False
|
||||
|
||||
|
||||
def test_wait_for_slave_clock_detected_at_timeout_boundary() -> None:
|
||||
runner = FakeCommandRunner([
|
||||
"CLKIN status: no clock signal detected\n",
|
||||
"CLKIN status: no clock signal detected\n",
|
||||
"CLKIN status: no clock signal detected\n",
|
||||
"CLKIN status: clock signal detected\n",
|
||||
])
|
||||
clock = FakeClock()
|
||||
|
||||
result = bpsk.wait_for_slave_clock(
|
||||
runner,
|
||||
initial_settle_s=0.05,
|
||||
poll_interval_s=0.05,
|
||||
timeout_s=0.20,
|
||||
sleep_function=clock.sleep,
|
||||
monotonic_function=clock.monotonic,
|
||||
)
|
||||
|
||||
assert result["state"] == bpsk.CLKIN_STATUS_DETECTED
|
||||
assert result["queries"] == 4
|
||||
assert result["time_to_detect_ms"] == pytest.approx(200.0)
|
||||
|
||||
|
||||
def test_clock_failure_cleanup_disables_both_outputs() -> None:
|
||||
runner = FakeCommandRunner("unexpected")
|
||||
clock = FakeClock()
|
||||
guard = bpsk.TransmitGuard()
|
||||
|
||||
with pytest.raises(RuntimeError):
|
||||
bpsk.common_clock_preflight(
|
||||
runner,
|
||||
guard,
|
||||
sleep_function=clock.sleep,
|
||||
monotonic_function=clock.monotonic,
|
||||
)
|
||||
|
||||
clock_commands = [
|
||||
command for command in runner.commands if Path(command[0]).name == "hackrf_clock.exe"
|
||||
]
|
||||
assert [command[command.index("-d") + 1] for command in clock_commands[-2:]] == [
|
||||
bpsk.MASTER_SERIAL,
|
||||
bpsk.SLAVE_SERIAL,
|
||||
]
|
||||
assert all(command[-2:] == ["-o", "0"] for command in clock_commands[-2:])
|
||||
|
||||
|
||||
def test_common_clock_preflight_is_fail_closed_and_skips_slave_rx() -> None:
|
||||
runner = FakeCommandRunner("CLKIN status: no clock signal detected\n")
|
||||
guard = bpsk.TransmitGuard()
|
||||
|
||||
with pytest.raises(RuntimeError, match="TIMEOUT"):
|
||||
bpsk.common_clock_preflight(runner, guard)
|
||||
|
||||
transfer_commands = [
|
||||
command for command in runner.commands if Path(command[0]).name == "hackrf_transfer.exe"
|
||||
]
|
||||
assert len(transfer_commands) == 1
|
||||
assert bpsk.MASTER_SERIAL in transfer_commands[0]
|
||||
assert guard.tx_allowed is False
|
||||
assert guard.transmit_calls == 0
|
||||
|
||||
|
||||
def test_common_clock_preflight_runs_both_short_rx_before_allowing_tx() -> None:
|
||||
runner = FakeCommandRunner()
|
||||
guard = bpsk.TransmitGuard()
|
||||
|
||||
result = bpsk.common_clock_preflight(runner, guard)
|
||||
|
||||
transfer_commands = [
|
||||
command for command in runner.commands if Path(command[0]).name == "hackrf_transfer.exe"
|
||||
]
|
||||
assert [command[command.index("-d") + 1] for command in transfer_commands] == [
|
||||
bpsk.MASTER_SERIAL,
|
||||
bpsk.SLAVE_SERIAL,
|
||||
]
|
||||
assert result["master_reg29"] == 0x26
|
||||
assert result["slave_reg29"] == 0x26
|
||||
assert result["slave_clkin_status"] == bpsk.CLKIN_STATUS_DETECTED
|
||||
assert guard.tx_allowed is True
|
||||
|
||||
|
||||
def test_failure_before_tx_produces_zero_tx_calls(tmp_path: Path) -> None:
|
||||
runner = FakeCommandRunner("CLKIN status: no clock signal detected\n")
|
||||
|
||||
with pytest.raises(RuntimeError, match="TIMEOUT"):
|
||||
bpsk.run_one_gain_point(24, 16, tmp_path, runner)
|
||||
|
||||
assert not any("-t" in command for command in runner.commands)
|
||||
|
||||
|
||||
def prepare_mock_point(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(bpsk, "CAPTURE_SAMPLE_COUNT", 16)
|
||||
monkeypatch.setattr(bpsk, "analyze_capture", lambda samples: fake_analysis())
|
||||
|
||||
|
||||
def test_waveform_sha_failure_produces_zero_tx_calls(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
runner = FakeCommandRunner()
|
||||
monkeypatch.setattr(bpsk, "build_tx_cs8", lambda: (_ for _ in ()).throw(ValueError("SHA")))
|
||||
|
||||
with pytest.raises(ValueError, match="SHA"):
|
||||
bpsk.run_one_gain_point(24, 24, tmp_path, runner)
|
||||
|
||||
assert runner.commands == []
|
||||
|
||||
|
||||
def test_rx_startup_failure_produces_zero_tx_calls(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
prepare_mock_point(monkeypatch)
|
||||
runner = FakeCommandRunner()
|
||||
starter = FakeProcessStarter(startup_failed=True, returncode=1)
|
||||
|
||||
with pytest.raises(RuntimeError, match="RX завершился до разрешения TX"):
|
||||
bpsk.run_one_gain_point(24, 24, tmp_path, runner, starter, lambda _: None)
|
||||
|
||||
assert not any("-t" in command for command in runner.commands)
|
||||
assert starter.processes[0].terminated is True
|
||||
|
||||
|
||||
def test_successful_experiment_has_one_tx_and_expected_order(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
prepare_mock_point(monkeypatch)
|
||||
events: list[str] = []
|
||||
runner = FakeCommandRunner()
|
||||
original_runner = runner
|
||||
|
||||
def recording_runner(command: list[str]) -> subprocess.CompletedProcess:
|
||||
if "-t" in command:
|
||||
events.append("tx_start_finish")
|
||||
return original_runner(command)
|
||||
|
||||
def preflight(command_runner: bpsk.CommandRunner, guard: bpsk.TransmitGuard) -> dict:
|
||||
del command_runner
|
||||
events.append("preflight")
|
||||
guard.allow()
|
||||
return {"passed": True}
|
||||
|
||||
class RecordingProcess(FakeRunningCommand):
|
||||
def wait(self, timeout: float) -> int:
|
||||
events.append("rx_finish")
|
||||
return super().wait(timeout)
|
||||
|
||||
def starter(command: list[str], stdout_path: Path, stderr_path: Path) -> RecordingProcess:
|
||||
events.append("rx_start")
|
||||
stdout_path.write_text("0 overruns\n", encoding="utf-8")
|
||||
stderr_path.write_text("", encoding="utf-8")
|
||||
Path(command[command.index("-r") + 1]).write_bytes(bytes(2 * bpsk.CAPTURE_SAMPLE_COUNT))
|
||||
return RecordingProcess()
|
||||
|
||||
monkeypatch.setattr(bpsk, "common_clock_preflight", preflight)
|
||||
monkeypatch.setattr(
|
||||
bpsk,
|
||||
"analyze_capture",
|
||||
lambda samples: events.append("analysis") or fake_analysis(),
|
||||
)
|
||||
|
||||
result = bpsk.run_one_gain_point(
|
||||
24, 24, tmp_path, recording_runner, starter, lambda _: None
|
||||
)
|
||||
|
||||
assert events == ["preflight", "rx_start", "tx_start_finish", "rx_finish", "analysis"]
|
||||
assert result["tx_calls"] == 1
|
||||
assert result["passed"] is True
|
||||
assert (tmp_path / "metadata.json").is_file()
|
||||
assert (tmp_path / "analysis.json").is_file()
|
||||
|
||||
|
||||
def test_tx_failure_has_no_retry_and_cleans_process_and_clocks(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
prepare_mock_point(monkeypatch)
|
||||
runner = FakeCommandRunner(tx_returncode=1)
|
||||
starter = FakeProcessStarter()
|
||||
|
||||
with pytest.raises(RuntimeError, match="Единственный TX"):
|
||||
bpsk.run_one_gain_point(24, 24, tmp_path, runner, starter, lambda _: None)
|
||||
|
||||
assert sum("-t" in command for command in runner.commands) == 1
|
||||
assert starter.processes[0].terminated is True
|
||||
clock_commands = [
|
||||
command for command in runner.commands if Path(command[0]).name == "hackrf_clock.exe"
|
||||
]
|
||||
assert [command[command.index("-d") + 1] for command in clock_commands[-2:]] == [
|
||||
bpsk.MASTER_SERIAL,
|
||||
bpsk.SLAVE_SERIAL,
|
||||
]
|
||||
assert all(command[-2:] == ["-o", "0"] for command in clock_commands[-2:])
|
||||
|
||||
|
||||
def test_analysis_failure_has_no_retry_and_cleans_process(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.setattr(bpsk, "CAPTURE_SAMPLE_COUNT", 16)
|
||||
monkeypatch.setattr(
|
||||
bpsk, "analyze_capture", lambda samples: (_ for _ in ()).throw(RuntimeError("analysis"))
|
||||
)
|
||||
runner = FakeCommandRunner()
|
||||
starter = FakeProcessStarter()
|
||||
|
||||
with pytest.raises(RuntimeError, match="analysis"):
|
||||
bpsk.run_one_gain_point(24, 24, tmp_path, runner, starter, lambda _: None)
|
||||
|
||||
assert sum("-t" in command for command in runner.commands) == 1
|
||||
assert starter.processes[0].terminated is True
|
||||
|
||||
|
||||
def test_transmit_guard_allows_exactly_one_tx() -> None:
|
||||
guard = bpsk.TransmitGuard()
|
||||
with pytest.raises(RuntimeError, match="общий clock-gate"):
|
||||
guard.register_transmit()
|
||||
guard.allow()
|
||||
guard.register_transmit()
|
||||
with pytest.raises(RuntimeError, match="Повторная передача"):
|
||||
guard.register_transmit()
|
||||
assert guard.transmit_calls == 1
|
||||
|
||||
|
||||
def test_gain_ladder_first_delta_uses_scientific_24_16_baseline(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
previous_values: list[float | None] = []
|
||||
|
||||
def fake_point(
|
||||
lna_gain_db: int,
|
||||
vga_gain_db: int,
|
||||
output_directory: Path,
|
||||
runner: bpsk.CommandRunner | None = None,
|
||||
process_starter: bpsk.ProcessStarter | None = None,
|
||||
sleep_function: bpsk.SleepFunction = bpsk.time.sleep,
|
||||
previous_abs_h: float | None = None,
|
||||
) -> dict:
|
||||
del output_directory, runner, process_starter, sleep_function
|
||||
previous_values.append(previous_abs_h)
|
||||
return {
|
||||
"lna_db": lna_gain_db,
|
||||
"vga_db": vga_gain_db,
|
||||
"tx_calls": 1,
|
||||
"passed": True,
|
||||
"stop_reason": "BPSK_PASS",
|
||||
"analysis": {"channel": {"abs_h": 0.004}},
|
||||
}
|
||||
|
||||
monkeypatch.setattr(bpsk, "run_one_gain_point", fake_point)
|
||||
|
||||
report = bpsk.run_gain_ladder(tmp_path)
|
||||
|
||||
assert previous_values == [bpsk.BASELINE_24_16_ABS_H]
|
||||
assert report["classification"] == "A"
|
||||
assert report["baseline"]["abs_h"] == bpsk.BASELINE_24_16_ABS_H
|
||||
|
||||
|
||||
def test_deterministic_waveform_sha256() -> None:
|
||||
assert hashlib.sha256(bpsk.build_tx_cs8()).hexdigest() == bpsk.EXPECTED_TX_SHA256
|
||||
|
||||
|
||||
def test_direct_invocation_smoke() -> None:
|
||||
project_root = Path(__file__).resolve().parents[1]
|
||||
result = subprocess.run(
|
||||
[sys.executable, str(project_root / "experiments" / "lab043_hackrf_bpsk.py"), "--help"],
|
||||
cwd=project_root,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
)
|
||||
assert result.returncode == 0, result.stderr
|
||||
assert "--analyze-existing" in result.stdout
|
||||
|
||||
|
||||
def test_offline_synthetic_bpsk() -> None:
|
||||
tx = bpsk.signed_iq_bytes_to_complex64(bpsk.build_tx_cs8())
|
||||
capture = np.concatenate((
|
||||
np.zeros(80_000, dtype=np.complex64),
|
||||
tx,
|
||||
np.zeros(80_000, dtype=np.complex64),
|
||||
))
|
||||
|
||||
analysis = bpsk.analyze_capture(capture, include_oracle=False)
|
||||
|
||||
assert analysis["detection"]["preamble_detected"] is True
|
||||
assert analysis["detection"]["payload_used_for_start"] is False
|
||||
assert analysis["official_ber"] == 0.0
|
||||
assert analysis["synchronization"]["preamble_errors"] == 0
|
||||
Reference in New Issue
Block a user