"""Lab048: измерение ускоренного автономного BPSK-приёмника без RF. Единственный основной режим обрабатывает сохранённую RX-запись Lab047 с уже зафиксированным порогом. Независимый приёмник не получает данные передатчика. JPEG сравнивается с TX только после завершённого независимого вердикта. """ from __future__ import annotations import argparse from contextlib import contextmanager from dataclasses import asdict, dataclass from functools import wraps import json import math from pathlib import Path import sys import time from typing import Callable, Iterator import numpy as np if __package__ in (None, ""): sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from experiments import lab043_hackrf_bpsk as lab043 from experiments import lab045_hackrf_bpsk_fec as lab045 from experiments import lab046c_offline_recovery as offline from experiments import lab047_jpeg_object_transport as lab047 OUTPUT_DIRECTORY = Path("data/processed/lab048") ATTEMPT_NAME = "final_processing_attempt.json" RESULT_NAME = "result.json" RECOVERED_IMAGE_NAME = "recovered_image.jpg" BASELINE_CAPTURE_SECONDS = 15.0 BASELINE_FULL_SECONDS = 37.423910000012256 BASELINE_STRICT_SECONDS = 32.45996900013415 BASELINE_AVERAGE_STRICT_SECONDS = 0.47043433333527757 BASELINE_APPROXIMATE_PEAK_BYTES = 1_545_755_352 FIXED_THRESHOLD = lab047.FIXED_HARDWARE_CANDIDATE_THRESHOLD @dataclass class StageMeasurement: calls: int = 0 cumulative_seconds: float = 0.0 def _timed_wrapper( function: Callable, measurement: StageMeasurement, ) -> Callable: @wraps(function) def wrapped(*args: object, **kwargs: object) -> object: started = time.perf_counter() try: return function(*args, **kwargs) finally: measurement.calls += 1 measurement.cumulative_seconds += time.perf_counter() - started return wrapped @contextmanager def measure_receiver_stages() -> Iterator[dict[str, StageMeasurement]]: """Временно измерить крупные стадии, не изменяя их входы и результаты.""" measurements = { "cs8_to_complex64": StageMeasurement(), "refine_start": StageMeasurement(), "matched_filter": StageMeasurement(), "timing": StageMeasurement(), "cfo_phase_soft": StageMeasurement(), "phy_header": StageMeasurement(), "post_preamble_fec_crc": StageMeasurement(), "viterbi": StageMeasurement(), } patches = ( (offline.lab046c, "_cs8_window", "cs8_to_complex64"), (lab043, "_refine_frame_start", "refine_start"), (lab043, "_matched_waveform", "matched_filter"), (lab043, "_preamble_timing_offset", "timing"), (lab043, "_evaluate_symbols", "cfo_phase_soft"), (lab045, "decode_phy_header_soft", "phy_header"), (lab045, "analyze_post_preamble_soft", "post_preamble_fec_crc"), (lab045.conv, "viterbi_decode_soft", "viterbi"), ) originals: list[tuple[object, str, Callable]] = [] for owner, name, measurement_name in patches: original = getattr(owner, name) originals.append((owner, name, original)) setattr(owner, name, _timed_wrapper(original, measurements[measurement_name])) try: yield measurements finally: for owner, name, original in reversed(originals): setattr(owner, name, original) def run_profiled_receiver( capture_path: Path, original_result_path: Path, ) -> tuple[offline.IndependentRecoveryResult, dict[str, StageMeasurement]]: """Обработать RX без TX-данных и измерить крупные стадии.""" rule, calibration, control = lab047.load_fixed_hardware_candidate_rule( capture_path, original_result_path, ) if rule.threshold != FIXED_THRESHOLD: raise RuntimeError("Порог Lab047 был изменён") with measure_receiver_stages() as measurements: started = time.perf_counter() independent = offline._run_with_fixed_candidate_rule( capture_path, rule, calibration, control, full_started=started, ) return independent, measurements def memory_array_inventory( result: offline.IndependentRecoveryResult, ) -> list[dict]: """Описать крупнейшие явные массивы и консервативную оценку пика.""" coarse_samples = result.coarse.maximum_window_samples decimation = offline.lab046c.lab044.lab043.DETECTION_DECIMATION decimated_samples = math.ceil(coarse_samples / decimation) reference_samples = offline.lab046c.lab044._preamble_reference_samples().size reference_decimated = math.ceil(reference_samples / decimation) score_samples = max(0, decimated_samples - reference_decimated + 1) strict_samples = result.strict.maximum_local_window_samples matched_samples = strict_samples + lab043._matched_filter_taps().size - 1 definitions = ( ("coarse_iq_window", "complex64", coarse_samples), ("coarse_decimated", "complex64", decimated_samples), ("coarse_downconversion_vector", "complex64", decimated_samples), ("coarse_correlation_scores", "float32", score_samples), ("strict_full_iq_window", "complex64", strict_samples), ("strict_cached_carrier_vector", "complex64", strict_samples), ("strict_baseband", "complex64", strict_samples), ("strict_matched_output", "complex64", matched_samples), ("matched_filter_taps", "float32", lab043._matched_filter_taps().size), ) inventory = [] for name, dtype, count in definitions: size_bytes = int(count * np.dtype(dtype).itemsize) inventory.append( { "name": name, "dtype": dtype, "shape": [int(count)], "bytes": size_bytes, "mib": size_bytes / (1024**2), } ) return inventory def result_summary( independent: offline.IndependentRecoveryResult, measurements: dict[str, StageMeasurement], diagnostics: dict, ) -> dict: """Собрать машинно-читаемый итог после разрешённой TX-диагностики.""" return { "stage": "LAB048_OFFLINE_RECEIVER_PERFORMANCE", "baseline": { "capture_seconds": BASELINE_CAPTURE_SECONDS, "full_processing_seconds": BASELINE_FULL_SECONDS, "strict_stage_seconds": BASELINE_STRICT_SECONDS, "average_strict_candidate_seconds": BASELINE_AVERAGE_STRICT_SECONDS, "approximate_peak_bytes": BASELINE_APPROXIMATE_PEAK_BYTES, }, "optimized": { "capture_seconds": independent.capture_duration_seconds, "full_processing_seconds": independent.full_elapsed_seconds, "coarse_search_seconds": independent.coarse.elapsed_seconds, "strict_stage_seconds": independent.strict.elapsed_seconds, "average_strict_candidate_seconds": independent.strict.average_candidate_seconds, "approximate_peak_bytes": independent.coarse.estimated_maximum_temporary_bytes, "stage_measurements": { name: asdict(value) for name, value in measurements.items() }, "memory_arrays": memory_array_inventory(independent), }, "speedup": BASELINE_FULL_SECONDS / independent.full_elapsed_seconds, "processing_to_capture_ratio": ( independent.full_elapsed_seconds / independent.capture_duration_seconds ), "real_time_capable": ( independent.full_elapsed_seconds < independent.capture_duration_seconds ), "threshold": independent.rule.threshold, "threshold_changed": independent.rule.threshold != FIXED_THRESHOLD, "coarse": { "total_local_peaks": independent.coarse.total_local_peaks, "first_stage_candidates": independent.coarse.candidate_count, "skipped_as_owned_event": ( independent.coarse.candidates_skipped_as_same_local_event ), "exact_duplicates": independent.coarse.exact_duplicate_candidates_skipped, }, "strict": asdict(independent.strict.counters), "object": { "valid_fragments": independent.object_verdict.valid_fragments, "missing_fragments": independent.object_verdict.missing_fragments, "object_complete": independent.object_verdict.object_complete, }, "after_independent_verdict": diagnostics, "receiver_oracles": lab047.receiver_oracle_declaration(), "phy_changed": False, "fec_changed": False, "bpsk_changed": False, "rf_tx": 0, } def run_and_write( capture_path: Path, original_result_path: Path, tx_image_path: Path, output_directory: Path, ) -> dict: """Однократно выполнить итоговый проход и после вердикта сравнить JPEG.""" output_directory = Path(output_directory) output_directory.mkdir(parents=True, exist_ok=True) attempt_path = output_directory / ATTEMPT_NAME result_path = output_directory / RESULT_NAME recovered_path = output_directory / RECOVERED_IMAGE_NAME if attempt_path.exists() or result_path.exists() or recovered_path.exists(): raise FileExistsError("Итоговый проход Lab048 уже запускался") rule, _, _ = lab047.load_fixed_hardware_candidate_rule( capture_path, original_result_path, ) attempt_path.write_text( json.dumps( { "capture_path": str(Path(capture_path)), "original_result_path": str(Path(original_result_path)), "threshold": rule.threshold, "automatic_retry": False, "receiver_oracles": lab047.receiver_oracle_declaration(), "rf_tx": 0, }, ensure_ascii=False, indent=2, ), encoding="utf-8", ) independent, measurements = run_profiled_receiver( capture_path, original_result_path, ) if not independent.completed or not independent.object_verdict.object_complete: raise RuntimeError("Итоговый независимый проход Lab048 не завершён") recovered = independent.object_verdict.recovered_object if recovered is None: raise RuntimeError("OBJECT COMPLETE без восстановленного JPEG") recovered_path.write_bytes(recovered) diagnostics = lab047.diagnose_hardware_jpeg_after_independent_verdict( independent, tx_image_path, recovered_path, ) summary = result_summary(independent, measurements, diagnostics) result_path.write_text( json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8", ) return summary def run_functional_tests() -> None: assert BASELINE_FULL_SECONDS > BASELINE_CAPTURE_SECONDS assert BASELINE_STRICT_SECONDS < BASELINE_FULL_SECONDS assert BASELINE_APPROXIMATE_PEAK_BYTES > 1024**3 assert FIXED_THRESHOLD == 0.030964473262429237 assert not any(lab047.receiver_oracle_declaration().values()) def main() -> int: parser = argparse.ArgumentParser(description="Lab048 offline receiver performance") parser.add_argument("--run-saved-capture", action="store_true") parser.add_argument("--capture", type=Path) parser.add_argument("--original-result", type=Path) parser.add_argument("--tx-image", type=Path) parser.add_argument("--output-directory", type=Path, default=OUTPUT_DIRECTORY) arguments = parser.parse_args() if arguments.run_saved_capture: if arguments.capture is None or arguments.original_result is None: parser.error("Требуются --capture и --original-result") if arguments.tx_image is None: parser.error("TX reference требуется только для послерезультатной диагностики") summary = run_and_write( arguments.capture, arguments.original_result, arguments.tx_image, arguments.output_directory, ) print(json.dumps(summary, ensure_ascii=False, indent=2)) return 0 run_functional_tests() print("Lab048 functional tests: PASS; RF TX = 0") return 0 if __name__ == "__main__": raise SystemExit(main())