"""Lab065: та же лестница уровней при TX VGA 20 дБ. Подготовка, программная проверка и preflight не передают RF. Единственный аппаратный сеанс доступен только отдельной командой с явным разрешением. """ from __future__ import annotations from dataclasses import asdict import argparse import hashlib import json from pathlib import Path import shutil import subprocess import sys import time if __package__ in (None, ""): sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import numpy as np from experiments import lab054_hardware_control as lab054 from experiments import lab056_capture_runner as hardware from experiments import lab064_rf_level_linearity as lab064 ROOT = Path(__file__).resolve().parents[1] RAW = ROOT / "data/raw/lab065" OUTPUT = ROOT / "data/processed/lab065" LAB064_RAW = ROOT / "data/raw/lab064" LAB064_RESULT = ROOT / "data/processed/lab064/rf_result.json" EXPECTED_TX_SHA256 = "fa260719659b2984fe29588238415622cd75c6daefec6cc9130ba1bf851bd3f9" TX_VGA_DB = 20 TX_VGA_DIFFERENCE_DB = 10.0 MEANINGFUL_GAIN_DB = 3.0 MEANINGFUL_RATIO_IMPROVEMENT_DB = 3.0 MAX_LEVEL_TO_LEVEL_GAIN_SPREAD_DB = 3.0 RESIDUAL_TRACKING_MARGIN_DB = 3.0 POINT = hardware.OperatingPoint(**asdict(lab064.POINT)) def save_new(path: Path, value: dict) -> None: """Записать новое свидетельство и не перезаписывать результат опыта.""" path.parent.mkdir(parents=True, exist_ok=True) with path.open("x", encoding="utf-8") as stream: json.dump(value, stream, ensure_ascii=False, indent=2, allow_nan=False) stream.write("\n") def save_text_new(path: Path, value: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("x", encoding="utf-8") as stream: stream.write(value) def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as stream: while block := stream.read(1024 * 1024): digest.update(block) return digest.hexdigest() def parameters() -> dict: result = lab064.parameters() result["tx_vga_db"] = TX_VGA_DB result["geometry_unchanged_from_lab064"] = True result["waveform_source"] = "Lab064 exact CS8 copy" return result def transfer_command(path: Path | str, samples: int, *, transmit: bool = False) -> list[str]: command = lab064.transfer_command(path, samples, transmit=transmit) if transmit: command[command.index("-x") + 1] = str(TX_VGA_DB) return command def validate_lab064_waveform() -> dict: source = LAB064_RAW / "tx_level_staircase.cs8" metadata = json.loads((LAB064_RAW / "tx_metadata.json").read_text(encoding="utf-8")) digest = sha256(source) if ( digest != EXPECTED_TX_SHA256 or metadata.get("tx_sha256") != EXPECTED_TX_SHA256 or source.stat().st_size != 2 * lab064.TX_SAMPLES or metadata.get("segments") != lab064.prepared_segments()[0] ): raise RuntimeError("Waveform Lab064 не совпадает с зафиксированным образцом") return metadata def validate_prepared() -> dict: metadata = json.loads((RAW / "tx_metadata.json").read_text(encoding="utf-8")) path = RAW / "tx_level_staircase.cs8" if ( metadata.get("lab") != "Lab065" or metadata.get("stage") != "PREPARATION_ONLY" or metadata.get("rf_tx_count") != 0 or metadata.get("parameters") != parameters() or metadata.get("tx_sha256") != EXPECTED_TX_SHA256 or metadata.get("source_tx_sha256") != EXPECTED_TX_SHA256 or path.stat().st_size != 2 * lab064.TX_SAMPLES or sha256(path) != EXPECTED_TX_SHA256 ): raise RuntimeError("Подготовленный Lab065 TX или параметры изменились") return metadata def prepare() -> dict: """Скопировать точные байты Lab064 в отдельный каталог Lab065.""" source_metadata = validate_lab064_waveform() RAW.mkdir(parents=True, exist_ok=True) OUTPUT.mkdir(parents=True, exist_ok=True) target = RAW / "tx_level_staircase.cs8" if target.exists(): raise FileExistsError("TX-файл Lab065 уже существует") shutil.copyfile(LAB064_RAW / "tx_level_staircase.cs8", target) if sha256(target) != EXPECTED_TX_SHA256: target.unlink() raise RuntimeError("Копия waveform Lab065 не прошла SHA-256") result = { "lab": "Lab065", "stage": "PREPARATION_ONLY", "rf_tx_count": 0, "head": subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip(), "parameters": parameters(), "segments": source_metadata["segments"], "tx_samples": lab064.TX_SAMPLES, "tx_bytes": 2 * lab064.TX_SAMPLES, "tx_duration_seconds": lab064.TX_SAMPLES / lab064.SAMPLE_RATE, "tx_sha256": sha256(target), "source_tx_sha256": source_metadata["tx_sha256"], "waveform_unchanged_from_lab064": True, "automatic_retry": False, "tx_command_prepared_not_run": transfer_command(target, lab064.TX_SAMPLES, transmit=True), "rx_command_prepared_not_run": transfer_command(RAW / "rx_stream.cs8", lab064.RX_SAMPLES), } save_new(RAW / "tx_metadata.json", result) save_new(OUTPUT / "preparation.json", result) return result def compare_sessions(old_result: dict, new_rows: list[dict]) -> dict: """Сопоставить одинаковые цифровые уровни Lab064 и Lab065.""" old_by_label = {row["label"]: row for row in old_result["rows"]} rows = [] for new in new_rows: old = old_by_label[new["label"]] comparison = { "label": new["label"], "tx_vga_nominal_difference_db": TX_VGA_DIFFERENCE_DB, "rx_matched_power_difference_db": lab064.difference_db( new["matched_signal_power"], old["matched_signal_power"] ), "gain_power_abs_h_squared_difference_db": lab064.difference_db( new["gain_power_abs_h_squared"], old["gain_power_abs_h_squared"] ), "residual_power_difference_db": lab064.difference_db( new["residual_power"], old["residual_power"] ), "matched_to_residual_improvement_db": ( new["matched_to_residual_db"] - old["matched_to_residual_db"] ), } if "complex_gain_h" in new and "residual_spectrum" in new: old_gain = complex( old["complex_gain_h"]["real"], old["complex_gain_h"]["imag"] ) new_gain = complex( new["complex_gain_h"]["real"], new["complex_gain_h"]["imag"] ) comparison["complex_gain_h_old"] = old["complex_gain_h"] comparison["complex_gain_h_new"] = new["complex_gain_h"] comparison["complex_gain_phase_change_degrees"] = float( np.degrees(np.angle(new_gain / old_gain)) ) comparison["residual_spectral_band_difference_db"] = { f"h{harmonic}_{harmonic * lab064.CARRIER_HZ}_hz": lab064.difference_db( new["residual_spectrum"][f"residual_band_h{harmonic}_power_code2"], old["residual_spectrum"][f"residual_band_h{harmonic}_power_code2"], ) for harmonic in (1, 2, 3) } rows.append(comparison) return {"rows": rows, "rules_fixed_before_rf": classification_rules()} def classification_rules() -> dict: return { "internal_linear_max_abs_error_db": 1.0, "compression_upper_step_error_at_most_db": -1.5, "meaningful_matched_gain_median_at_least_db": MEANINGFUL_GAIN_DB, "meaningful_ratio_improvement_median_at_least_db": MEANINGFUL_RATIO_IMPROVEMENT_DB, "maximum_level_to_level_gain_spread_db": MAX_LEVEL_TO_LEVEL_GAIN_SPREAD_DB, "residual_tracks_if_within_matched_gain_db": RESIDUAL_TRACKING_MARGIN_DB, } def classify_result(new_result: dict, cross_comparison: dict) -> dict: """Выбрать один из пяти заранее заданных результатов Lab065.""" if not new_result.get("capture_valid"): return {"case": "INVALID_CAPTURE", "rules_fixed_before_rf": classification_rules()} internal_errors = [row["linearity_error_db"] for row in new_result["comparisons"]] upper_error = internal_errors[-1] cross_rows = cross_comparison["rows"] matched = np.asarray([row["rx_matched_power_difference_db"] for row in cross_rows]) residual = np.asarray([row["residual_power_difference_db"] for row in cross_rows]) improvements = np.asarray([row["matched_to_residual_improvement_db"] for row in cross_rows]) median_matched = float(np.median(matched)) median_residual = float(np.median(residual)) median_improvement = float(np.median(improvements)) spread = float(np.ptp(matched)) linear = max(abs(value) for value in internal_errors) <= 1.0 if upper_error <= -1.5: case = "COMPRESSION_AT_HIGHER_TX_GAIN" elif median_matched < MEANINGFUL_GAIN_DB: case = "NO_MEANINGFUL_GAIN" elif ( linear and spread <= MAX_LEVEL_TO_LEVEL_GAIN_SPREAD_DB and median_improvement >= MEANINGFUL_RATIO_IMPROVEMENT_DB ): case = "LINEAR_GAIN_IMPROVEMENT" elif ( median_residual >= median_matched - RESIDUAL_TRACKING_MARGIN_DB and median_improvement < MEANINGFUL_RATIO_IMPROVEMENT_DB ): case = "RESIDUAL_RISES_WITH_SIGNAL" else: case = "NOT_LOCALIZED" return { "case": case, "summary": { "median_matched_gain_db": median_matched, "median_residual_gain_db": median_residual, "median_matched_to_residual_improvement_db": median_improvement, "matched_gain_level_spread_db": spread, "maximum_internal_linearity_error_db": max(abs(value) for value in internal_errors), "upper_step_linearity_error_db": upper_error, }, "rules_fixed_before_rf": classification_rules(), } def software_gate() -> dict: metadata = validate_prepared() old_result = json.loads(LAB064_RESULT.read_text(encoding="utf-8")) synthetic = lab064.synthetic_linearity_gate() result = { "lab": "Lab065", "stage": "SOFTWARE_GATE", "rf_tx_count": 0, "passed": bool(synthetic["passed"] and old_result["capture_valid"]), "tx_sha256": metadata["tx_sha256"], "waveform_unchanged_from_lab064": metadata["waveform_unchanged_from_lab064"], "lab064_baseline_classification": old_result["classification"]["case"], "classification_rules": classification_rules(), } save_new(OUTPUT / "software_gate.json", result) if not result["passed"]: raise RuntimeError("Lab065 software gate FAIL") return result def preflight(*, runner=None, process_checker=None, clock_checker=None, sleep_function=time.sleep) -> dict: """Проверить общую опору и 0,1 с RX-фона при полном запрете TX.""" metadata = validate_prepared() software = json.loads((OUTPUT / "software_gate.json").read_text(encoding="utf-8")) if not software.get("passed") or software.get("tx_sha256") != metadata["tx_sha256"]: raise RuntimeError("Нет действующего Lab065 software gate") save_new(RAW / "preflight_attempt.json", { "lab": "Lab065", "rf_tx_count": 0, "point": asdict(POINT), "parameters": parameters(), "tx_sha256": metadata["tx_sha256"], }) actual_runner = runner or lab054.clock._default_command_runner process_checker = process_checker or lab054.check_processes clock_checker = clock_checker or hardware.common_clock def rx_only(command: list[str]): if "-t" in command: raise PermissionError("TX запрещён в Lab065 preflight") return actual_runner(command) result = { "lab": "Lab065", "stage": "COMMON_CLOCK_AND_RX_ONLY_PREFLIGHT", "passed": False, "rf_tx_count": 0, "stop_before_tx": True, "point": asdict(POINT), "parameters": parameters(), "tx_sha256": metadata["tx_sha256"], } clock_started = False try: before = process_checker() result["process_before"] = before.to_dict() lab054.require_clear(before) result["inventory"] = lab054.phy.lab045._verify_device_inventory(rx_only) clock_started = True result["clock"] = clock_checker( POINT, rx_only, lab054.clock.TransmitGuard(), sleep_function=sleep_function ) hardware.require_clock_pass(result["clock"]) path = RAW / "preflight_rx.cs8" captured = rx_only(transfer_command(path, lab064.PREFLIGHT_SAMPLES)) stdout, stderr = captured.stdout or "", captured.stderr or "" result["rx_process"] = { "returncode": captured.returncode, "stdout": stdout, "stderr": stderr, } result["rx_counter_summary"] = lab064.counter_summary(stdout + stderr, "overrun") if captured.returncode or not result["rx_counter_summary"]["clean"]: raise RuntimeError("RX preflight: процесс или счётчик overruns не прошёл") if path.stat().st_size != 2 * lab064.PREFLIGHT_SAMPLES: raise RuntimeError("Неверная длина Lab065 preflight RX") level = hardware.receive_level(path) level.pop("symbol_energy_code2_seconds", None) level["symbol_energy"] = "NOT_APPLICABLE_TO_BACKGROUND" level["ac_rms_code"] = float(np.sqrt(max(0.0, level["ac_power_code2"]))) level["unexpected_strong_signal"] = bool( level["ac_rms_code"] >= lab064.STRONG_AC_RMS_CODE or level["max_abs_i"] >= lab064.STRONG_COMPONENT_CODE or level["max_abs_q"] >= lab064.STRONG_COMPONENT_CODE ) result["rx_level"] = level result["rx_sha256"] = sha256(path) result["rx_overruns"] = result["rx_counter_summary"]["final_count"] if level["saturation_detected"] or level["unexpected_strong_signal"]: raise RuntimeError("RX-only уровень не прошёл gate; STOP без изменения gain") result["passed"] = True except Exception as error: result["error"] = str(error) raise finally: errors = lab054.disable_clock_outputs(rx_only) if clock_started else [] try: after = process_checker() result["process_after"] = after.to_dict() lab054.require_clear(after) except Exception as error: errors.append(str(error)) result["cleanup_errors"] = errors result["passed"] = result["passed"] and not errors save_new(RAW / "preflight_result.json", result) OUTPUT.mkdir(parents=True, exist_ok=True) (OUTPUT / "preflight_result.json").write_text( json.dumps(result, ensure_ascii=False, indent=2, allow_nan=False) + "\n", encoding="utf-8", ) if not result["passed"]: raise RuntimeError("Lab065 preflight FAIL") return result def require_preflight(metadata: dict) -> dict: record = json.loads((RAW / "preflight_result.json").read_text(encoding="utf-8")) if ( record.get("passed") is not True or record.get("rf_tx_count") != 0 or record.get("point") != asdict(POINT) or record.get("parameters") != parameters() or record.get("tx_sha256") != metadata["tx_sha256"] ): raise PermissionError("Нет успешного RX-only preflight для конфигурации Lab065") hardware.require_clock_pass(record.get("clock", {})) level = record.get("rx_level", {}) if ( level.get("saturation_detected") is not False or level.get("unexpected_strong_signal") is not False or record.get("rx_overruns") != 0 ): raise PermissionError("RX-only preflight не разрешает передачу") return record def capture_once( *, authorization: bool = False, runner=None, process_starter=None, process_checker=None, clock_checker=None, sleep_function=time.sleep, ) -> dict: """Выполнить один RX-first сеанс после отдельного прямого разрешения.""" if authorization is not True: raise PermissionError("Нужно отдельное разрешение на единственный RF-сеанс Lab065") metadata = validate_prepared() require_preflight(metadata) receive_path = RAW / "rx_stream.cs8" if receive_path.exists() or (RAW / "capture_result.json").exists(): raise FileExistsError("Аппаратный результат Lab065 уже существует") actual_runner = runner or lab054.clock._default_command_runner process_starter = process_starter or lab054.clock._default_process_starter process_checker = process_checker or lab054.check_processes clock_checker = clock_checker or hardware.common_clock save_new(RAW / "hardware_attempt.json", { "lab": "Lab065", "authorized_once": True, "automatic_retry": False, "point": asdict(POINT), "parameters": parameters(), "tx_sha256": metadata["tx_sha256"], }) result = { "lab": "Lab065", "stage": "SINGLE_RF_LEVEL_CAPTURE", "passed": False, "rf_tx_count": 0, "automatic_tx_repeats": 0, "point": asdict(POINT), "parameters": parameters(), "tx_sha256": metadata["tx_sha256"], } guard = lab054.clock.TransmitGuard() process = None clock_started = False try: before = process_checker() result["process_before"] = before.to_dict() lab054.require_clear(before) result["inventory"] = lab054.phy.lab045._verify_device_inventory(actual_runner) clock_started = True result["clock"] = clock_checker( POINT, actual_runner, guard, sleep_function=sleep_function ) hardware.require_clock_pass(result["clock"]) process = process_starter( transfer_command(receive_path, lab064.RX_SAMPLES), RAW / "rx_stdout.txt", RAW / "rx_stderr.txt", ) sleep_function(lab064.RX_LEAD_SECONDS) if process.poll() is not None: raise RuntimeError("RX завершился до TX") before_tx = hardware.receive_level(receive_path, tail_samples=65_536) before_tx.pop("symbol_energy_code2_seconds", None) result["rx_level_before_tx"] = before_tx if before_tx["saturation_detected"]: raise RuntimeError("RX clipping до TX; RF TX = 0") guard.register_transmit() result["rf_tx_count"] = guard.transmit_calls transmitted = actual_runner( transfer_command(RAW / "tx_level_staircase.cs8", lab064.TX_SAMPLES, transmit=True) ) tx_stdout, tx_stderr = transmitted.stdout or "", transmitted.stderr or "" save_text_new(RAW / "tx_stdout.txt", tx_stdout) save_text_new(RAW / "tx_stderr.txt", tx_stderr) result["tx_process"] = { "returncode": transmitted.returncode, "stdout_file": "tx_stdout.txt", "stderr_file": "tx_stderr.txt", } result["tx_counter_summary"] = lab064.counter_summary( tx_stdout + tx_stderr, "underrun" ) if transmitted.returncode: raise RuntimeError("Единственный TX завершился ошибкой; повтор запрещён") if process.wait(lab064.RX_SECONDS + 10): raise RuntimeError("RX завершился ошибкой; повтор запрещён") process.close_logs() result["rx_samples"] = receive_path.stat().st_size // 2 result["rx_bytes"] = receive_path.stat().st_size result["rx_duration_seconds"] = result["rx_samples"] / lab064.SAMPLE_RATE if result["rx_bytes"] != 2 * lab064.RX_SAMPLES: raise RuntimeError("Повреждённая длина RX; повтор запрещён") rx_stdout = (RAW / "rx_stdout.txt").read_text(encoding="utf-8") rx_stderr = (RAW / "rx_stderr.txt").read_text(encoding="utf-8") result["rx_counter_summary"] = lab064.counter_summary( rx_stdout + rx_stderr, "overrun" ) result["rx_sha256"] = sha256(receive_path) level = hardware.receive_level(receive_path) level.pop("symbol_energy_code2_seconds", None) result["rx_level"] = level result["rx_clipping"] = bool(level["saturation_detected"]) result["passed"] = True validity = lab064.capture_validity(result) if not validity["valid"]: result["passed"] = False raise RuntimeError("Запись невалидна; результат сохранён, повтор запрещён") except Exception as error: result["error"] = str(error) raise finally: errors = [] if process is not None: try: process.terminate() except Exception as error: errors.append(str(error)) if clock_started: errors.extend(lab054.disable_clock_outputs(actual_runner)) try: after = process_checker() result["process_after"] = after.to_dict() lab054.require_clear(after) except Exception as error: errors.append(str(error)) result["cleanup_errors"] = errors result["passed"] = result["passed"] and not errors save_new(RAW / "capture_result.json", result) OUTPUT.mkdir(parents=True, exist_ok=True) (OUTPUT / "capture_result.json").write_text( json.dumps(result, ensure_ascii=False, indent=2, allow_nan=False) + "\n", encoding="utf-8", ) if not result["passed"]: raise RuntimeError("Lab065 RF-сеанс завершён с ошибкой; повтор запрещён") return result def analyze_capture() -> dict: """Рассчитать метрики Lab065 и прямое сравнение с Lab064.""" metadata = validate_prepared() rx_path = RAW / "rx_stream.cs8" capture = json.loads((RAW / "capture_result.json").read_text(encoding="utf-8")) if capture.get("rf_tx_count") != 1 or capture.get("rx_sha256") != sha256(rx_path): raise RuntimeError("Нет подтверждённой единственной RX-записи Lab065") tx = lab064.energy.Cs8File(RAW / "tx_level_staircase.cs8") rx = lab064.energy.Cs8File(rx_path) location = lab064.locate_waveform(tx, rx) background_rows = [] for window in lab064.background_windows(): rx_start = location["start_sample"] + window["tx_relative_start_sample"] background_rows.append({ **window, "rx_start_sample": rx_start, "metrics": lab064._window_metrics(rx.read(rx_start, window["samples"])), }) background_by_label = {row["label"]: row for row in background_rows} preceding = {"L0": "INITIAL_GUARD", "L1": "GAP_L0_L1", "L2": "GAP_L1_L2", "L3": "GAP_L2_L3"} rows = [] for segment in metadata["segments"]: tx_start = segment["start_sample"] nominal_rx_start = location["start_sample"] + tx_start fitted = lab064.energy.analyze_rx( lab064.lab056.OffsetSource(tx, tx_start), rx, [nominal_rx_start], lab064.SEGMENT_SAMPLES, lab064.SAMPLES_PER_SYMBOL, )["frames"][0] actual_start = fitted["start"] active_reference = tx.read( tx_start + lab064.TRIM_SAMPLES, lab064.SEGMENT_SAMPLES - 2 * lab064.TRIM_SAMPLES, ) active_received = rx.read( actual_start + lab064.TRIM_SAMPLES, lab064.SEGMENT_SAMPLES - 2 * lab064.TRIM_SAMPLES, ) complex_gain = lab064.lab056.complex_gain( active_reference, active_received, fitted["cfo_hz"] ) matched = fitted["heldout_signal_power"] residual = fitted["heldout_residual_power"] background = background_by_label[preceding[segment["label"]]] rows.append({ "label": segment["label"], "level_db": segment["level_db"], "rx_start_sample": actual_start, "alignment_adjustment_samples": fitted["alignment_adjustment"], "full_iq": lab064._window_metrics(rx.read(actual_start, lab064.SEGMENT_SAMPLES)), "background_label": background["label"], "background": background["metrics"], "matched_signal_power": matched, "residual_power": residual, "matched_to_residual_db": lab064.difference_db(matched, residual), "complex_gain_h": complex_gain, "gain_abs": fitted["gain_abs"], "gain_power_abs_h_squared": fitted["gain_abs"] ** 2, "cfo_hz": fitted["cfo_hz"], "residual_spectrum": lab064.residual_spectral_diagnostics( active_reference, active_received, fitted["cfo_hz"], complex_gain ), }) comparisons = [] for index in range(1, len(rows)): tx_difference = metadata["segments"][index]["tx_input_difference_from_previous_db"] rx_difference = lab064.difference_db( rows[index]["matched_signal_power"], rows[index - 1]["matched_signal_power"] ) comparisons.append({ "comparison": f"{rows[index]['label']}/{rows[index - 1]['label']}", "tx_input_difference_db": tx_difference, "rx_matched_power_difference_db": rx_difference, "linearity_error_db": rx_difference - tx_difference, }) validity = lab064.capture_validity(capture) result = { "lab": "Lab065", "stage": "OFFLINE_RF_LEVEL_LINEARITY_ANALYSIS", "packet_receiver_used": False, "location": location, "rows": rows, "background_windows": background_rows, "background_summary": lab064.summarize_background(background_rows), "comparisons": comparisons, "capture_valid": validity["valid"], "capture_invalid_reasons": validity["invalid_reasons"], "tx_underruns_maximum": capture.get("tx_counter_summary", {}).get("maximum_count"), "rx_overruns_maximum": capture.get("rx_counter_summary", {}).get("maximum_count"), "rx_clipping": any(row["full_iq"]["clipped_samples_fraction"] > 0 for row in rows), } old_result = json.loads(LAB064_RESULT.read_text(encoding="utf-8")) result["lab064_to_lab065"] = compare_sessions(old_result, rows) result["classification"] = classify_result(result, result["lab064_to_lab065"]) save_new(OUTPUT / "rf_result.json", result) return result def run_functional_tests() -> None: assert transfer_command("NUL", 100, transmit=True)[-7:-5] == ["-x", "20"] assert "-t" not in transfer_command("NUL", 100) assert validate_lab064_waveform()["tx_sha256"] == EXPECTED_TX_SHA256 assert lab064.synthetic_linearity_gate()["passed"] try: capture_once(authorization=False) except PermissionError: pass else: raise AssertionError("RF без отдельного разрешения") def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--stage", choices=("prepare", "software", "preflight", "capture", "analyze"), required=True, ) parser.add_argument("--authorize-one-rf", action="store_true") arguments = parser.parse_args() if arguments.stage == "prepare": result = prepare() elif arguments.stage == "software": result = software_gate() elif arguments.stage == "preflight": result = preflight() elif arguments.stage == "capture": result = capture_once(authorization=arguments.authorize_one_rf) else: result = analyze_capture() print(json.dumps(result, ensure_ascii=False, indent=2, allow_nan=False)) print("RF TX NOW = 0") if __name__ == "__main__": main()