624 lines
28 KiB
Python
624 lines
28 KiB
Python
"""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()
|