"""Fixed, receive-only RF-001A capture operations for the HTTP agent. This module has no transmit operation. A capture uses one immutable RF profile, one known RX serial number, a unique directory outside Git, and one finite ``hackrf_transfer`` child process. Failed readiness checks stop that exact process tree and never retry the capture. """ from __future__ import annotations from datetime import datetime, timedelta, timezone import hashlib import json import math import os from pathlib import Path import shutil import subprocess import sys import time from typing import Any, Callable, Mapping, Sequence FIXED_RX_SERIAL = "000000000000000097c063dc2e5327d3" PROFILE_NAME = "RF001A_COMMON_CLOCK_001" MINIMUM_FREE_BYTES = 6_000_000_000 READINESS_TIMEOUT_SECONDS = 5.0 CAPTURE_SECONDS = 300 CAPTURE_SAMPLES = 2_400_000_000 CAPTURE_FILE_NAME = "rf001a_ota_capture.cs8" MANIFEST_FILE_NAME = "capture_manifest.json" STARTED_FILE_NAME = "capture_started.json" RESULT_FILE_NAME = "capture_result.json" REPORT_FILE_NAME = "offline_decode_report.json" MAX_REPORT_BYTES = 192 * 1024 EXPECTED_PROFILE: dict[str, object] = { "test_id": PROFILE_NAME, "center_frequency_hz": 1_400_000_000, "sample_rate_hz": 8_000_000, "symbol_rate_sps": 500_000, "modulation": "QPSK", "inner_fec": "2/3", "application_payload_bytes": 256, "packet_count": 100, "inter_packet_gap_ms": 5, "clock_mode": "COMMON_10_MHZ", "rx_lna_db": 24, "rx_vga_db": 24, "rf_amp_enabled": False, "antenna_bias_enabled": False, "hardware_format": "cs8", "rx_capture_seconds": 20, "rx_capture_samples": 160_000_000, } CommandRunner = Callable[[Sequence[str], float], Any] BackgroundLauncher = Callable[[Sequence[str]], int] ProcessGate = Callable[[], tuple[bool, Mapping[str, object]]] ProcessStopper = Callable[[int], None] def _atomic_json(path: Path, value: Mapping[str, object]) -> None: temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text( json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) os.replace(temporary, path) def _read_json(path: Path) -> dict[str, Any]: value = json.loads(path.read_text(encoding="utf-8-sig")) if not isinstance(value, dict): raise ValueError(f"JSON object required: {path}") return value def load_fixed_profile(repository: Path) -> dict[str, object]: path = repository / "configs" / "rf" / "rf001a_common_clock.json" profile = _read_json(path) mismatches = { key: {"expected": expected, "actual": profile.get(key)} for key, expected in EXPECTED_PROFILE.items() if profile.get(key) != expected } if mismatches: raise ValueError(f"RF-001A capture profile mismatch: {mismatches}") return {key: profile[key] for key in EXPECTED_PROFILE} def resolve_hackrf_tools( hackrf_info_path: Path | None, hackrf_clock_path: Path | None, hackrf_transfer_path: Path | None, ) -> tuple[Path, Path, Path]: if hackrf_info_path is None: raise FileNotFoundError("hackrf_info path is required for RX capture") base = hackrf_info_path.parent info = hackrf_info_path clock = hackrf_clock_path or base / "hackrf_clock.exe" transfer = hackrf_transfer_path or base / "hackrf_transfer.exe" for path in (info, clock, transfer): if not path.is_file(): raise FileNotFoundError(f"required HackRF tool is absent: {path}") return info, clock, transfer def default_process_gate() -> tuple[bool, Mapping[str, object]]: from tools.hackrf_process_check import ProcessState, check_processes check = check_processes() return check.state is ProcessState.CLEAR, check.to_dict() def default_background_launcher(arguments: Sequence[str]) -> int: flags = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) | getattr( subprocess, "CREATE_NO_WINDOW", 0 ) process = subprocess.Popen( list(arguments), stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, creationflags=flags, close_fds=True, ) return int(process.pid) def default_process_stopper(pid: int) -> None: completed = subprocess.run( ["taskkill.exe", "/PID", str(pid), "/T", "/F"], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=15, check=False, creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), ) if completed.returncode not in {0, 128}: raise RuntimeError( f"taskkill failed for pid {pid}: {completed.returncode}: " f"{completed.stderr.strip()}" ) def _iso_utc(value: str) -> datetime: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) if parsed.tzinfo is None: raise ValueError("timestamp must contain a timezone") return parsed.astimezone(timezone.utc) def _bounded_command(result: Any) -> dict[str, object]: return { "arguments": list(result.arguments), "returncode": int(result.returncode), "stdout": str(result.stdout)[-8_000:], "stderr": str(result.stderr)[-8_000:], } def start_rx_capture( repository: Path, work_root: Path, session_id: str, rx_serial: str, *, hackrf_info_path: Path | None, hackrf_clock_path: Path | None, hackrf_transfer_path: Path | None, command_runner: CommandRunner, process_gate: ProcessGate = default_process_gate, background_launcher: BackgroundLauncher = default_background_launcher, process_stopper: ProcessStopper = default_process_stopper, sleep: Callable[[float], None] = time.sleep, monotonic: Callable[[], float] = time.monotonic, ) -> tuple[dict[str, object], list[str]]: """Start one finite RX capture and return only after the READY gate.""" errors: list[str] = [] result: dict[str, object] = { "session_id": session_id, "state": "NOT_READY", "hardware_accessed": False, "common_clock_verified": False, "rf_rx": 0, "rf_tx": 0, "automatic_retry": False, } runner_pid: int | None = None child_pid: int | None = None try: if rx_serial != FIXED_RX_SERIAL: raise ValueError("RX serial does not match the fixed RF-001 receiver") profile = load_fixed_profile(repository) info_path, clock_path, transfer_path = resolve_hackrf_tools( hackrf_info_path, hackrf_clock_path, hackrf_transfer_path, ) session_directory = work_root / "rx_capture" / session_id session_directory.parent.mkdir(parents=True, exist_ok=True) if session_directory.exists(): raise FileExistsError(f"capture session already exists: {session_directory}") free_bytes = shutil.disk_usage(session_directory.parent).free result["free_bytes_before"] = free_bytes if free_bytes < MINIMUM_FREE_BYTES: raise RuntimeError( f"capture volume has {free_bytes} free bytes; " f"{MINIMUM_FREE_BYTES} required" ) clear, process_evidence = process_gate() result["process_gate"] = dict(process_evidence) if not clear: raise RuntimeError("conflicting SDR process gate is not CLEAR") info = command_runner([str(info_path)], 20.0) result["hackrf_info"] = _bounded_command(info) result["hardware_accessed"] = True if info.returncode != 0 or rx_serial.lower() not in ( f"{info.stdout}\n{info.stderr}".lower() ): raise RuntimeError("hackrf_info did not confirm the fixed RX serial") clock = command_runner([str(clock_path), "-d", rx_serial, "-i"], 20.0) result["hackrf_clock"] = _bounded_command(clock) clock_lines = { line.strip().lower() for line in f"{clock.stdout}\n{clock.stderr}".splitlines() } if clock.returncode != 0 or not any( line.endswith("clkin status: clock signal detected") for line in clock_lines ): raise RuntimeError("CLKIN clock signal was not detected") result["common_clock_verified"] = True session_directory.mkdir() capture_path = session_directory / CAPTURE_FILE_NAME arguments_path = session_directory / "hackrf_transfer_arguments.json" stdout_path = session_directory / "hackrf_transfer.stdout.txt" stderr_path = session_directory / "hackrf_transfer.stderr.txt" started_path = session_directory / STARTED_FILE_NAME final_path = session_directory / RESULT_FILE_NAME manifest_path = session_directory / MANIFEST_FILE_NAME runner_path = Path(__file__).resolve().with_name("rf001_capture_runner.py") if not runner_path.is_file(): raise FileNotFoundError(f"capture runner is absent: {runner_path}") transfer_arguments = [ "-d", rx_serial, "-r", str(capture_path), "-f", str(profile["center_frequency_hz"]), "-s", str(profile["sample_rate_hz"]), "-n", str(CAPTURE_SAMPLES), "-l", str(profile["rx_lna_db"]), "-g", str(profile["rx_vga_db"]), "-a", "0", "-p", "0", "-B", ] arguments_path.write_text( json.dumps(transfer_arguments, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) launch_arguments = [ sys.executable, "-B", "-X", "utf8", str(runner_path), "--executable", str(transfer_path), "--arguments-json", str(arguments_path), "--capture-path", str(capture_path), "--stdout-path", str(stdout_path), "--stderr-path", str(stderr_path), "--started-status", str(started_path), "--result-status", str(final_path), ] runner_pid = background_launcher(launch_arguments) result["runner_pid"] = runner_pid result["rf_rx"] = 1 deadline = monotonic() + READINESS_TIMEOUT_SECONDS first_positive_size: int | None = None latest_size = 0 started: dict[str, Any] | None = None while monotonic() < deadline: if final_path.is_file(): final = _read_json(final_path) raise RuntimeError(f"capture exited before READY: {final}") if started_path.is_file(): started = _read_json(started_path) raw_pid = started.get("pid") if type(raw_pid) is int and raw_pid > 0: child_pid = raw_pid if capture_path.is_file(): latest_size = capture_path.stat().st_size if latest_size > 0: if first_positive_size is None: first_positive_size = latest_size elif latest_size > first_positive_size and started is not None: break sleep(0.2) else: raise RuntimeError("capture did not become READY within five seconds") if started is None or child_pid is None or first_positive_size is None: raise RuntimeError("capture readiness evidence is incomplete") capture_started_at = str(started.get("started_at", "")) expected_end = _iso_utc(capture_started_at) + timedelta( seconds=CAPTURE_SECONDS ) expected_bytes = CAPTURE_SAMPLES * 2 manifest: dict[str, object] = { "schema_version": 1, "session_id": session_id, "profile": profile, "capture_seconds": CAPTURE_SECONDS, "capture_samples": CAPTURE_SAMPLES, "capture_duration_reason": "manual AUTHORIZE RF TX gate", "rx_serial": rx_serial, "capture_path": str(capture_path), "capture_started_at": capture_started_at, "expected_capture_end": expected_end.isoformat(), "expected_capture_bytes": expected_bytes, "runner_pid": runner_pid, "child_pid": child_pid, "started_status_path": str(started_path), "result_status_path": str(final_path), "stdout_path": str(stdout_path), "stderr_path": str(stderr_path), "common_clock_verified": True, "automatic_retry": False, "rf_rx": 1, "rf_tx": 0, } _atomic_json(manifest_path, manifest) result.update( { "state": "READY", "session_directory": str(session_directory), "manifest_path": str(manifest_path), "capture_path": str(capture_path), "capture_started_at": capture_started_at, "expected_capture_end": expected_end.isoformat(), "expected_capture_bytes": expected_bytes, "child_pid": child_pid, "capture_bytes_at_ready": latest_size, "common_clock_verified": True, } ) except ( ImportError, OSError, ValueError, RuntimeError, subprocess.SubprocessError, ) as error: errors.append(f"{type(error).__name__}: {error}") for pid in (child_pid, runner_pid): if pid is None: continue try: process_stopper(pid) except (OSError, RuntimeError, subprocess.SubprocessError) as stop_error: errors.append( f"failed to stop pid {pid}: {type(stop_error).__name__}: " f"{stop_error}" ) result["rf_rx"] = 0 return result, errors def read_capture_status( work_root: Path, session_id: str, ) -> tuple[dict[str, object], list[str]]: errors: list[str] = [] result: dict[str, object] = { "session_id": session_id, "hardware_accessed": False, "rf_tx": 0, } try: session_directory = work_root / "rx_capture" / session_id manifest = _read_json(session_directory / MANIFEST_FILE_NAME) if manifest.get("session_id") != session_id or manifest.get("rf_tx") != 0: raise ValueError("capture manifest does not match the requested session") capture_path = Path(str(manifest["capture_path"])) final_path = Path(str(manifest["result_status_path"])) resolved_session = session_directory.resolve() for path in (capture_path, final_path): try: path.resolve().relative_to(resolved_session) except ValueError as error: raise ValueError("capture manifest path escapes the session") from error expected_bytes = int(manifest["expected_capture_bytes"]) current_bytes = capture_path.stat().st_size if capture_path.is_file() else 0 result.update( { "manifest_path": str(session_directory / MANIFEST_FILE_NAME), "capture_path": str(capture_path), "capture_started_at": manifest.get("capture_started_at"), "expected_capture_end": manifest.get("expected_capture_end"), "expected_capture_bytes": expected_bytes, "capture_bytes": current_bytes, "common_clock_verified": manifest.get("common_clock_verified"), } ) expected_end = _iso_utc(str(manifest["expected_capture_end"])) result["capture_window_open"] = datetime.now(timezone.utc) < expected_end if final_path.is_file(): final = _read_json(final_path) result["runner_result"] = final complete = ( final.get("exit_code") == 0 and final.get("runner_error") in {None, ""} and final.get("capture_size") == expected_bytes and current_bytes == expected_bytes ) result["state"] = "CAPTURE_COMPLETE" if complete else "CAPTURE_FAILED" if not complete: errors.append("finite RX capture did not meet the complete-file gate") else: result["state"] = "CAPTURE_RUNNING" except (OSError, ValueError, KeyError, TypeError) as error: errors.append(f"{type(error).__name__}: {error}") result["state"] = "STATUS_FAILED" return result, errors def validate_tx_window( capture_started_at: str, tx_started_at: str, tx_finished_at: str, ) -> None: capture = _iso_utc(capture_started_at) started = _iso_utc(tx_started_at) finished = _iso_utc(tx_finished_at) if not capture <= started <= finished: raise ValueError("TX timestamps are outside the ordered capture window") def decode_completed_capture( repository: Path, work_root: Path, session_id: str, tx_started_at: str, tx_finished_at: str, *, command_runner: CommandRunner, ) -> tuple[dict[str, object], list[str]]: errors: list[str] = [] result: dict[str, object] = { "session_id": session_id, "mode": "OFFLINE_CAPTURE_DECODE", "hardware_accessed": False, "rf_tx": 0, } try: status, status_errors = read_capture_status(work_root, session_id) if status_errors or status.get("state") != "CAPTURE_COMPLETE": raise RuntimeError(f"capture is not complete: {status_errors or status}") capture_started_at = str(status["capture_started_at"]) validate_tx_window(capture_started_at, tx_started_at, tx_finished_at) session_directory = work_root / "rx_capture" / session_id report_path = session_directory / "offline_decode_report.json" if report_path.exists(): raise FileExistsError(f"decode report already exists: {report_path}") decoder_script = Path(__file__).resolve().with_name("rf001_offline_decode.py") truth_path = repository / "configs" / "rf" / "rf001a_truth_metadata.json" decode = command_runner( [ os.fspath(Path(sys.executable)), "-B", "-X", "utf8", str(decoder_script), "--repository", str(repository), "--iq-file", str(status["capture_path"]), "--truth-metadata", str(truth_path), "--report", str(report_path), "--capture-started-at", capture_started_at, "--tx-started-at", tx_started_at, "--tx-finished-at", tx_finished_at, ], 1_800.0, ) result["decode"] = _bounded_command(decode) if decode.returncode != 0: raise RuntimeError(f"offline capture decoder exited with {decode.returncode}") report = _read_json(report_path) diagnostics = read_capture_diagnostics( work_root, session_id, tx_started_at, tx_finished_at, ) if diagnostics[1]: raise RuntimeError( f"capture diagnostics failed: {diagnostics[1]}" ) result.update( { "report_path": str(report_path), "crc_valid_packets": report.get("crc_valid_packets"), "exact_packets": report.get("exact_packets"), "peak_analysis_memory_bytes": report.get("peak_analysis_memory_bytes"), "window_source": report.get("window_source"), "diagnostics": diagnostics[0], } ) except (OSError, ValueError, RuntimeError, KeyError, TypeError) as error: errors.append(f"{type(error).__name__}: {error}") return result, errors def _power_metrics( capture_path: Path, start_sample: int, stop_sample: int, *, chunk_samples: int = 1_000_000, ) -> dict[str, object]: """Measure one fixed cs8 window without loading it all into memory.""" import numpy as np total_samples = capture_path.stat().st_size // 2 if not 0 <= start_sample < stop_sample <= total_samples: raise ValueError("diagnostic window is outside the cs8 capture") raw = np.memmap(capture_path, dtype=np.int8, mode="r") power_sum = 0.0 i_sum = 0.0 q_sum = 0.0 rail_codes = 0 peak_component = 0 for chunk_start in range(start_sample, stop_sample, chunk_samples): chunk_stop = min(stop_sample, chunk_start + chunk_samples) values = np.asarray( raw[chunk_start * 2:chunk_stop * 2], dtype=np.int8, ).reshape(-1, 2) wide = values.astype(np.float64) i_values = wide[:, 0] q_values = wide[:, 1] power_sum += float(np.sum(i_values * i_values + q_values * q_values)) i_sum += float(np.sum(i_values)) q_sum += float(np.sum(q_values)) rail_codes += int(np.count_nonzero((values == -128) | (values == 127))) peak_component = max(peak_component, int(np.max(np.abs(wide)))) sample_count = stop_sample - start_sample mean_power = power_sum / sample_count return { "start_sample": start_sample, "stop_sample": stop_sample, "sample_count": sample_count, "mean_power_code_squared": mean_power, "rms_complex_normalized": math.sqrt(mean_power) / 128.0, "mean_i_code": i_sum / sample_count, "mean_q_code": q_sum / sample_count, "peak_component_code": peak_component, "rail_codes": rail_codes, } def read_capture_diagnostics( work_root: Path, session_id: str, tx_started_at: str, tx_finished_at: str, ) -> tuple[dict[str, object], list[str]]: """Return the fixed report and signal metrics for one completed capture.""" errors: list[str] = [] result: dict[str, object] = { "session_id": session_id, "mode": "OFFLINE_CAPTURE_DIAGNOSTICS", "hardware_accessed": False, "rf_tx": 0, } try: status, status_errors = read_capture_status(work_root, session_id) if status_errors or status.get("state") != "CAPTURE_COMPLETE": raise RuntimeError(f"capture is not complete: {status_errors or status}") capture_started_at = str(status["capture_started_at"]) validate_tx_window(capture_started_at, tx_started_at, tx_finished_at) capture_path = Path(str(status["capture_path"])) session_directory = work_root / "rx_capture" / session_id report_path = session_directory / REPORT_FILE_NAME if not report_path.is_file(): raise FileNotFoundError(f"decode report is absent: {report_path}") report_size = report_path.stat().st_size if report_size <= 0 or report_size > MAX_REPORT_BYTES: raise ValueError( f"decode report size {report_size} is outside the allowed range" ) report_bytes = report_path.read_bytes() report = json.loads(report_bytes.decode("utf-8-sig")) if not isinstance(report, dict) or report.get("rf_tx") != 0: raise ValueError("decode report is invalid or does not preserve rf_tx = 0") runner_result = status.get("runner_result") if not isinstance(runner_result, dict): raise ValueError("capture result is absent") capture_sha256 = str(runner_result.get("sha256", "")) if report.get("iq_sha256") != capture_sha256: raise ValueError("decode report does not match the capture SHA-256") sample_rate = int(EXPECTED_PROFILE["sample_rate_hz"]) total_samples = capture_path.stat().st_size // 2 capture_time = _iso_utc(capture_started_at) tx_start = _iso_utc(tx_started_at) tx_finish = _iso_utc(tx_finished_at) tx_start_sample = round((tx_start - capture_time).total_seconds() * sample_rate) tx_stop_sample = round((tx_finish - capture_time).total_seconds() * sample_rate) tx_start_sample = max(0, min(total_samples, tx_start_sample)) tx_stop_sample = max(0, min(total_samples, tx_stop_sample)) if tx_stop_sample <= tx_start_sample: raise ValueError("TX window does not overlap the capture") tx_samples = tx_stop_sample - tx_start_sample pre_start = max(0, tx_start_sample - tx_samples) post_stop = min(total_samples, tx_stop_sample + tx_samples) signal = _power_metrics(capture_path, tx_start_sample, tx_stop_sample) pre = _power_metrics(capture_path, pre_start, tx_start_sample) post = _power_metrics(capture_path, tx_stop_sample, post_stop) baseline_power = ( float(pre["mean_power_code_squared"]) + float(post["mean_power_code_squared"]) ) / 2.0 signal_power = float(signal["mean_power_code_squared"]) power_ratio = signal_power / baseline_power if baseline_power > 0 else None power_delta_db = ( 10.0 * math.log10(power_ratio) if power_ratio is not None and power_ratio > 0 else None ) result.update( { "report_path": str(report_path), "report_bytes": report_size, "report_sha256": hashlib.sha256(report_bytes).hexdigest(), "capture_sha256": capture_sha256, "report": report, "signal_metrics": { "tx_window": signal, "pre_tx_baseline": pre, "post_tx_baseline": post, "tx_to_baseline_power_ratio": power_ratio, "tx_to_baseline_db": power_delta_db, }, } ) except ( ImportError, OSError, ValueError, RuntimeError, KeyError, TypeError, ) as error: errors.append(f"{type(error).__name__}: {error}") return result, errors