Prepare coordinated RF-001A OTA test

tools/rf001a_prepare.py

@@ -0,0 +1,183 @@
+"""Prepare and verify RF-001A HackRF cs8 IQ without opening a radio."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+import sys
+
+import numpy as np
+
+if __package__ in (None, ""):
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from protocol.video003_modem import SAMPLE_RATE_HZ
+from protocol.video005_iq_waveform import NOMINAL_CARRIER_HZ, SYMBOL_RATE
+from tools.qpsk_rf_common import (
+    GAP_MS, PACKET_COUNT, PAYLOAD_BYTES, SEED, TEST_ID, read_iq_file,
+    sha256_file,
+)
+from tools.qpsk_rf_tx import write_dry_run
+
+
+ROOT = Path(__file__).resolve().parents[1]
+SPEC_PATH = ROOT / "configs/rf/rf001a_common_clock.json"
+
+
+def load_spec(path: Path = SPEC_PATH) -> dict:
+    """Reject a spec that diverges from the existing TX and RX PHY."""
+    spec = json.loads(path.read_text(encoding="utf-8-sig"))
+    expected = {
+        "test_id": "RF001A_COMMON_CLOCK_001",
+        "legacy_waveform_test_id": TEST_ID,
+        "center_frequency_hz": 1_400_000_000,
+        "sample_rate_hz": SAMPLE_RATE_HZ,
+        "symbol_rate_sps": SYMBOL_RATE,
+        "modulation": "QPSK",
+        "inner_fec": "2/3",
+        "application_payload_bytes": PAYLOAD_BYTES,
+        "packet_count": PACKET_COUNT,
+        "inter_packet_gap_ms": GAP_MS,
+        "seed": SEED,
+        "digital_carrier_offset_hz": int(NOMINAL_CARRIER_HZ),
+        "clock_mode": "COMMON_10_MHZ",
+        "tx_vga_db": 20,
+        "rx_lna_db": 24,
+        "rx_vga_db": 24,
+        "rf_amp_enabled": False,
+        "antenna_bias_enabled": False,
+        "canonical_format": "c64le",
+        "hardware_format": "cs8",
+        "cs8_scale_codes_per_source_unit": 256,
+        "expected_sample_count": 8_833_828,
+        "rx_capture_seconds": 20,
+        "rx_capture_samples": 160_000_000,
+    }
+    for key, value in expected.items():
+        if type(spec.get(key)) is not type(value) or spec[key] != value:
+            raise ValueError(f"RF-001A spec disagrees with existing PHY: {key}")
+    digest = spec.get("expected_canonical_sha256", "")
+    if len(digest) != 64 or any(char not in "0123456789abcdef" for char in digest):
+        raise ValueError("invalid expected canonical SHA256")
+    hardware_digest = spec.get("expected_hardware_sha256", "")
+    if len(hardware_digest) != 64 or any(char not in "0123456789abcdef" for char in hardware_digest):
+        raise ValueError("invalid expected hardware SHA256")
+    if spec["rx_capture_samples"] != spec["sample_rate_hz"] * spec["rx_capture_seconds"]:
+        raise ValueError("RX capture sample count disagrees with rate and duration")
+    return spec
+
+
+def quantize_cs8(source: np.ndarray, scale: int) -> np.ndarray:
+    """Interleave signed I then Q; fail if scaling would hit either rail."""
+    if source.ndim != 1 or not np.iscomplexobj(source):
+        raise ValueError("expected one-dimensional complex IQ")
+    if not np.all(np.isfinite(source)):
+        raise ValueError("nonfinite IQ")
+    maximum = max(float(np.max(np.abs(source.real))), float(np.max(np.abs(source.imag))))
+    if maximum * scale >= 126.5:
+        raise ValueError("cs8 rail risk")
+    raw = np.empty(source.size * 2, dtype=np.int8)
+    raw[0::2] = np.rint(source.real * scale).astype(np.int8)
+    raw[1::2] = np.rint(source.imag * scale).astype(np.int8)
+    return raw
+
+
+def verify_artifacts(directory: Path, spec: dict) -> dict:
+    """Validate both files and reverse-read the actual HackRF bytes."""
+    canonical = directory / "tx_waveform.c64le"
+    hardware = directory / "tx_waveform.cs8"
+    truth_path = directory / "tx_metadata.json"
+    record_path = directory / "rf001a_metadata.json"
+    truth = json.loads(truth_path.read_text(encoding="utf-8"))
+    record = json.loads(record_path.read_text(encoding="utf-8"))
+    count = spec["expected_sample_count"]
+    if (truth["test_id"] != spec["legacy_waveform_test_id"]
+            or truth["waveform_sample_count"] != count
+            or truth["packet_count"] != spec["packet_count"]
+            or truth["seed"] != spec["seed"]
+            or truth["digital_carrier_offset_hz"] != spec["digital_carrier_offset_hz"]
+            or [row["sequence_number"] for row in truth["expected_packet_hashes"]]
+            != list(range(spec["packet_count"]))):
+        raise ValueError("TX truth disagrees with shared spec")
+    if canonical.stat().st_size != count * 8 or hardware.stat().st_size != count * 2:
+        raise ValueError("IQ byte count disagrees with shared spec")
+    if sha256_file(canonical) != spec["expected_canonical_sha256"]:
+        raise ValueError("canonical waveform SHA256 mismatch")
+    if truth["iq_sha256"] != spec["expected_canonical_sha256"]:
+        raise ValueError("TX metadata SHA256 mismatch")
+    if record["test_id"] != spec["test_id"] or record["cs8_scale"] != spec["cs8_scale_codes_per_source_unit"]:
+        raise ValueError("RF-001A artifact metadata mismatch")
+    if record["hardware_sha256"] != spec["expected_hardware_sha256"] or record["hardware_sha256"] != sha256_file(hardware):
+        raise ValueError("hardware waveform SHA256 mismatch")
+    source, _ = read_iq_file(canonical)
+    observed, clipping = read_iq_file(hardware, "cs8")
+    scale = spec["cs8_scale_codes_per_source_unit"]
+    expected_codes = quantize_cs8(source, scale)
+    actual_codes = np.memmap(hardware, dtype=np.int8, mode="r")
+    if not np.array_equal(actual_codes, expected_codes) or clipping != 0:
+        raise ValueError("cs8 readback or rail-code mismatch")
+    expected_quantized = (expected_codes[0::2].astype(np.float32)
+                          + 1j * expected_codes[1::2].astype(np.float32)) / 128
+    if not np.array_equal(observed, expected_quantized.astype(np.complex64)):
+        raise ValueError("cs8 I/Q order or signed readback mismatch")
+    maximum_error = max(float(np.max(np.abs(source.real - expected_codes[0::2] / scale))),
+                        float(np.max(np.abs(source.imag - expected_codes[1::2] / scale))))
+    if maximum_error > 0.5001 / scale:
+        raise ValueError("quantization error exceeds half a code")
+    return {
+        "samples": count,
+        "duration_s": count / spec["sample_rate_hz"],
+        "canonical_sha256": sha256_file(canonical),
+        "hardware_sha256": record["hardware_sha256"],
+        "hardware_bytes": hardware.stat().st_size,
+        "peak_component": record["peak_component"],
+        "rms_complex": record["rms_complex"],
+        "cs8_rail_codes": clipping,
+        "maximum_component_error_source_units": maximum_error,
+    }
+
+
+def prepare(directory: Path, spec: dict) -> dict:
+    """Create a new ignored output directory, then verify its TX bytes."""
+    canonical, _ = write_dry_run(directory)
+    source, _ = read_iq_file(canonical)
+    scale = spec["cs8_scale_codes_per_source_unit"]
+    raw = quantize_cs8(source, scale)
+    hardware = directory / "tx_waveform.cs8"
+    raw.tofile(hardware)
+    record = {
+        "test_id": spec["test_id"],
+        "shared_spec": "configs/rf/rf001a_common_clock.json",
+        "legacy_truth": "tx_metadata.json",
+        "canonical_sha256": sha256_file(canonical),
+        "hardware_sha256": sha256_file(hardware),
+        "cs8_scale": scale,
+        "peak_component": float(max(np.max(np.abs(source.real)), np.max(np.abs(source.imag)))),
+        "rms_complex": float(np.sqrt(np.mean(np.abs(source) ** 2))),
+        "sample_count": int(source.size),
+        "duration_s": source.size / spec["sample_rate_hz"],
+        "hardware_format": "signed int8, interleaved I/Q",
+        "rf_tx": 0,
+    }
+    (directory / "rf001a_metadata.json").write_text(
+        json.dumps(record, indent=2) + "\n", encoding="utf-8"
+    )
+    return verify_artifacts(directory, spec)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    group = parser.add_mutually_exclusive_group(required=True)
+    group.add_argument("--output-dir", type=Path)
+    group.add_argument("--verify-dir", type=Path)
+    arguments = parser.parse_args()
+    spec = load_spec()
+    result = (prepare(arguments.output_dir, spec) if arguments.output_dir is not None
+              else verify_artifacts(arguments.verify_dir, spec))
+    print(json.dumps(result, indent=2))
+    print("RF TX=0")
+
+
+if __name__ == "__main__":
+    main()