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

Lab046C: record first continuous RF object attempt

Browse Source
This commit is contained in:
LittleSam129
2026-09-01 15:42:00 +03:00
parent 39fa4b139d
commit b6654671b4
4 changed files with 1441 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

350
tests/test_lab046c_hackrf_bpsk_object.py Normal file
View File

@@ -0,0 +1,350 @@
"""Аппаратно-независимые проверки Lab046C preparation и fail-closed runner."""
from __future__ import annotations
import inspect
import json
import subprocess
import sys
from pathlib import Path
import numpy as np
import pytest
from experiments import lab046c_hackrf_bpsk_object as lab046c
def small_waveform_setup(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(lab046c, "OBJECT_SIZE_BYTES", 300)
monkeypatch.setattr(lab046c, "_validate_fixed_phy", lambda: None)
monkeypatch.setattr(
lab046c.lab045,
"build_tx_cs8",
lambda plan: bytes([plan.payload[12] & 0x7F, 0]) * plan.waveform_sample_count,
)
def test_frame_and_stream_accounting_matches_actual_code() -> None:
frames = lab046c.lab046.build_object_frames(bytes(lab046c.OBJECT_SIZE_BYTES))
assert len(frames) == 68
assert frames[-1].encoded_payload_length - 12 == 36
assert {frame.plan.waveform_sample_count for frame in frames} == {2_306_400}
assert sum(frame.plan.waveform_sample_count for frame in frames) == 156_835_200
assert 2 * sum(frame.plan.waveform_sample_count for frame in frames) == 313_670_400
assert sum(frame.plan.airtime_seconds for frame in frames) == pytest.approx(19.6044)
def test_preparation_writes_one_stream_sequentially_and_hashes(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
import hashlib
small_waveform_setup(monkeypatch)
output = tmp_path / "prepared"
prepared = lab046c.prepare_transmission(
output,
object_factory=lambda count: bytes((index * 17 + 3) % 256 for index in range(count)),
)
metadata = json.loads(prepared.metadata_path.read_text(encoding="utf-8"))
assert prepared.object_path.stat().st_size == 300
assert prepared.frame_count == 2
assert prepared.fragment_count == 2
assert prepared.stream_path.stat().st_size == prepared.stream_file_bytes
assert hashlib.sha256(prepared.object_path.read_bytes()).hexdigest() == prepared.object_sha256
assert hashlib.sha256(prepared.stream_path.read_bytes()).hexdigest() == prepared.stream_sha256
assert metadata["rf_tx_count"] == 0
assert metadata["object"]["deterministic_seed"] is None
assert metadata["tx_stream"]["inter_frame_samples_added"] == 0
assert metadata["recovered_object"] is None
def test_receiver_api_has_no_tx_oracle() -> None:
parameters = set(inspect.signature(lab046c.receive_object_from_cs8).parameters)
assert parameters == {
"capture_path",
"preamble_threshold",
"max_frames",
"max_search_iterations",
"chunk_samples",
}
forbidden = {
"tx_starts",
"frame_starts",
"frame_lengths",
"expected_fragment_count",
"tx_object",
"tx_packets",
"tx_coded_bits",
}
assert parameters.isdisjoint(forbidden)
def test_cs8_window_is_complex64_and_exact(tmp_path: Path) -> None:
path = tmp_path / "capture.cs8"
path.write_bytes(np.asarray([64, -64, -128, 127, 0, 1], dtype=np.int8).tobytes())
raw = np.memmap(path, dtype=np.int8, mode="r")
samples = lab046c._cs8_window(raw, 0, 3)
assert samples.dtype == np.complex64
assert samples[0] == pytest.approx(0.5 - 0.5j)
assert samples[1].real == pytest.approx(-1.0)
assert samples[1].imag == pytest.approx(127 / 128)
def fake_frame(local_start: int, local_end: int) -> lab046c.lab046b.IqFrameReception:
reception = lab046c.lab045.IndependentReception(
{"fec_decode_pass": True, "crc": lab046c.lab045.CRC_PASS},
b"packet",
bytes(lab046c.lab045.PAYLOAD_SIZE_BYTES),
)
return lab046c.lab046b.IqFrameReception(
detection_number=1,
coarse_start_sample=local_start,
refined_start_sample=local_start,
frame_end_sample=local_end,
preamble_score=0.9,
over_median_db=20.0,
over_second_db=0.0,
timing_offset_samples=0,
symbol_sample_phase=0,
nominal_sample=0,
estimated_cfo_hz=0.0,
estimated_phase_rad=0.0,
bpsk_polarity="resolved from preamble",
preamble_errors=0,
snr_db=20.0,
post_preamble_soft=np.empty(0),
reception=reception,
)
def test_windowed_scanner_owns_each_candidate_once_and_cursor_moves_forward(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
path = tmp_path / "capture.cs8"
path.write_bytes(bytes(2 * 80))
monkeypatch.setattr(lab046c, "maximum_phy_waveform_samples", lambda: 16)
monkeypatch.setattr(lab046c, "REFINEMENT_MARGIN_SAMPLES", 0)
monkeypatch.setattr(lab046c.lab044.lab043, "DETECTION_DECIMATION", 8)
def detect(samples, threshold):
del threshold
starts = [8, 24, 40] if len(samples) == 48 else [8]
return tuple(
lab046c.lab046b.IqPreambleCandidate(start, 0.9, 20.0, 0.0)
for start in starts
), 0
def decode(samples, candidate, detection_number):
del samples
start = candidate.coarse_start_sample
frame = fake_frame(start, start + 4)
return lab046c.lab046b.CandidateDecode(
start,
start + 4,
True,
True,
lab046c.lab045.CRC_PASS,
True,
"CRC_PASS",
frame,
)
monkeypatch.setattr(lab046c.lab046b, "_detect_iq_preamble_candidates", detect)
monkeypatch.setattr(lab046c.lab046b, "_decode_iq_candidate", decode)
result = lab046c.scan_cs8_capture(path, chunk_samples=48)
starts = [frame.refined_start_sample for frame in result.accepted_frames]
assert starts == [8, 24, 40, 56, 72]
assert len(starts) == len(set(starts))
assert all(after > before for before, after in zip(result.cursor_history, result.cursor_history[1:]))
assert result.processing_dtype == "complex64"
assert result.maximum_window_samples <= 48
def test_windowed_scanner_safety_limits_fail_closed(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
path = tmp_path / "capture.cs8"
path.write_bytes(bytes(2 * 80))
monkeypatch.setattr(lab046c, "maximum_phy_waveform_samples", lambda: 16)
monkeypatch.setattr(lab046c, "REFINEMENT_MARGIN_SAMPLES", 0)
monkeypatch.setattr(lab046c.lab044.lab043, "DETECTION_DECIMATION", 8)
monkeypatch.setattr(
lab046c.lab046b,
"_detect_iq_preamble_candidates",
lambda samples, threshold: (
tuple(lab046c.lab046b.IqPreambleCandidate(start, 0.9, 20.0, 0.0) for start in (8, 16, 24)),
0,
),
)
monkeypatch.setattr(
lab046c.lab046b,
"_decode_iq_candidate",
lambda samples, candidate, number: lab046c.lab046b.CandidateDecode(
candidate.coarse_start_sample,
candidate.coarse_start_sample + 1,
False,
False,
lab046c.lab045.CRC_NOT_REACHED,
False,
"BAD",
None,
),
)
result = lab046c.scan_cs8_capture(path, chunk_samples=48, max_search_iterations=1)
assert result.search_iterations == 1
assert result.termination_reason == "MAX_SEARCH_ITERATIONS_REACHED"
class FakeRunner:
def __init__(self) -> None:
self.commands: list[list[str]] = []
def __call__(self, command: list[str]):
self.commands.append(command)
return subprocess.CompletedProcess(command, 0, "", "")
def test_preflight_only_never_registers_tx(monkeypatch: pytest.MonkeyPatch) -> None:
runner = FakeRunner()
monkeypatch.setattr(lab046c.lab045, "_verify_device_inventory", lambda runner: {"passed": True})
def preflight(runner, guard):
guard.allow()
return {"passed": True, "master_reg29": 0x26, "slave_reg29": 0x26}
monkeypatch.setattr(lab046c.lab044.lab043, "common_clock_preflight", preflight)
monkeypatch.setattr(lab046c.lab044.lab043, "_disable_clock_outputs", lambda runner: [])
result = lab046c.run_common_clock_preflight_only(runner, process_checker=lambda: [])
assert result["rf_tx_count"] == 0
assert not any("-t" in command for command in runner.commands)
@pytest.mark.parametrize(
("clock_errors", "processes", "message"),
[
(["CLKOUT OFF failed"], [], "CLKOUT cleanup"),
([], ["pid 123"], "hackrf_transfer"),
],
)
def test_preflight_cleanup_failure_is_fail_closed(
monkeypatch: pytest.MonkeyPatch,
clock_errors: list[str],
processes: list[str],
message: str,
) -> None:
monkeypatch.setattr(lab046c.lab045, "_verify_device_inventory", lambda runner: {"passed": True})
def preflight(runner, guard):
guard.allow()
return {"passed": True}
calls = 0
def process_checker():
nonlocal calls
calls += 1
return [] if calls == 1 else processes
monkeypatch.setattr(lab046c.lab044.lab043, "common_clock_preflight", preflight)
monkeypatch.setattr(
lab046c.lab044.lab043,
"_disable_clock_outputs",
lambda runner: clock_errors,
)
with pytest.raises(RuntimeError, match=message):
lab046c.run_common_clock_preflight_only(FakeRunner(), process_checker=process_checker)
def test_prepared_metadata_cannot_self_declare_short_stream(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
import hashlib
output = tmp_path / "forged"
output.mkdir()
object_path = output / "tx_object.bin"
stream_path = output / "tx_stream.cs8"
object_path.write_bytes(bytes(lab046c.OBJECT_SIZE_BYTES))
stream_path.write_bytes(b"\x00\x00")
metadata = {
"stage": "LAB046C_PREPARATION_ONLY",
"rf_tx_count": 0,
"object": {
"path": str(object_path),
"sha256": hashlib.sha256(object_path.read_bytes()).hexdigest(),
"object_id": lab046c.OBJECT_ID,
},
"fragmentation": {
"fragment_data_bytes": lab046c.lab046.FRAGMENT_DATA_SIZE,
"fragment_count": 1,
"last_fragment_data_bytes": 1,
},
"tx_stream": {
"path": str(stream_path),
"sha256": hashlib.sha256(stream_path.read_bytes()).hexdigest(),
"frame_count": 1,
"samples_per_frame": 1,
"sample_count": 1,
"file_bytes": 2,
"duration_seconds": 1 / lab046c.SAMPLE_RATE_HZ,
"inter_frame_samples_added": 0,
},
"hardware": lab046c.hardware_parameters(),
"receiver_oracles": {"tx_start_offset": False},
}
(output / "tx_metadata.json").write_text(json.dumps(metadata), encoding="utf-8")
monkeypatch.setattr(lab046c, "_sha256_file", lambda path: hashlib.sha256(path.read_bytes()).hexdigest())
with pytest.raises(RuntimeError, match="accounting"):
lab046c._load_and_validate_prepared(output)
def test_prepared_metadata_requires_exact_oracle_declaration() -> None:
exact = {key: False for key in lab046c.RECEIVER_ORACLE_KEYS}
assert lab046c._receiver_oracle_declaration_is_safe(exact) is True
assert lab046c._receiver_oracle_declaration_is_safe(
{"tx_start_offset": False}
) is False
with_enabled_oracle = exact | {"tx_start_offset": True}
assert lab046c._receiver_oracle_declaration_is_safe(with_enabled_oracle) is False
def test_hardware_entry_point_requires_two_explicit_keys() -> None:
project_root = Path(__file__).resolve().parents[1]
script = project_root / "experiments" / "lab046c_hackrf_bpsk_object.py"
result = subprocess.run(
[sys.executable, str(script), "--hardware"],
cwd=project_root,
capture_output=True,
text=True,
timeout=60,
)
assert result.returncode == 2
assert "RF TX = 0" in result.stdout
def test_hardware_sequence_has_rx_before_one_tx_and_cleanup_in_finally() -> None:
source = inspect.getsource(lab046c.run_hardware_experiment)
assert source.index("process_starter(rx_command") < source.index("guard.register_transmit()")
assert source.count("guard.register_transmit()") == 1
assert "finally:" in source
assert "_disable_clock_outputs" in source
assert '"-d", MASTER_SERIAL' in source
assert '"-d", SLAVE_SERIAL' in source
def test_direct_invocation_is_software_only() -> None:
project_root = Path(__file__).resolve().parents[1]
result = subprocess.run(
[sys.executable, str(project_root / "experiments" / "lab046c_hackrf_bpsk_object.py")],
cwd=project_root,
capture_output=True,
text=True,
timeout=60,
)
assert result.returncode == 0, result.stderr
assert "LAB046C SOFTWARE PREFLIGHT = PASS" in result.stdout
assert "RF TX = 0" in result.stdout