"""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()