132 lines
5.3 KiB
Python
132 lines
5.3 KiB
Python
"""Standalone smoke test for the SN9000 multi-port VISA HiSLIP acquisition path."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
|
|
import numpy as np
|
|
|
|
from python_app.hardware_full.sn9000_service import Sn9000Service
|
|
from python_app.models.dataset_model import SweepCollection
|
|
from python_app.models.run_config_model import RadarSweepModel, RunConfigModel
|
|
|
|
|
|
def _parse_args() -> argparse.Namespace:
|
|
"""Parse command-line options for the SN9000 VISA smoke test."""
|
|
parser = argparse.ArgumentParser(description="Acquire one SN9000 collection through VISA HiSLIP")
|
|
parser.add_argument(
|
|
"--host",
|
|
default="127.0.0.1",
|
|
help="SNVNA HiSLIP server host (default: 127.0.0.1)",
|
|
)
|
|
parser.add_argument(
|
|
"--port",
|
|
type=int,
|
|
default=4880,
|
|
help="SNVNA HiSLIP server TCP port (default: 4880)",
|
|
)
|
|
parser.add_argument("--start-hz", type=float, default=1_000_000.0)
|
|
parser.add_argument("--stop-hz", type=float, default=3_000_000_000.0)
|
|
parser.add_argument("--points", type=int, default=201)
|
|
parser.add_argument("--ifbw-hz", type=float, default=10_000.0)
|
|
parser.add_argument("--power-dbm", type=float, default=-10.0)
|
|
parser.add_argument("--timeout-ms", type=int, default=20_000)
|
|
parser.add_argument("--no-preset", action="store_true")
|
|
parser.add_argument(
|
|
"--visa-library",
|
|
default="@ivi",
|
|
help="PyVISA IVI backend specification, e.g. @ivi or /usr/lib/x86_64-linux-gnu/libvisa.so",
|
|
)
|
|
return parser.parse_args()
|
|
|
|
|
|
def _validate_collection(collection: SweepCollection, expected_points: int) -> None:
|
|
"""Raise unless a SN9000 collection has the expected traces, combos, shapes, and finite values."""
|
|
expected_traces = (
|
|
RunConfigModel.MULTI_DEVICE_INPUT_POSITIONS * RunConfigModel.MULTI_DEVICE_OUTPUT_POSITIONS
|
|
)
|
|
if len(collection.traces) != expected_traces:
|
|
raise RuntimeError(
|
|
f"SN9000 collection has {len(collection.traces)} traces, expected {expected_traces}"
|
|
)
|
|
|
|
reference_frequency = collection.traces[0].frequency_hz
|
|
if reference_frequency.shape != (expected_points,):
|
|
raise RuntimeError(f"Unexpected frequency shape: {reference_frequency.shape}")
|
|
if not np.all(np.isfinite(reference_frequency)):
|
|
raise RuntimeError("Frequency axis contains non-finite values")
|
|
if np.any(np.diff(reference_frequency) < 0.0):
|
|
raise RuntimeError("Frequency axis is not monotonic")
|
|
|
|
expected_combos = RunConfigModel.build_matrix_radar_virtual_combos()
|
|
for trace, expected_combo in zip(collection.traces, expected_combos, strict=True):
|
|
if (
|
|
int(trace.combo.input) != int(expected_combo.input)
|
|
or int(trace.combo.output) != int(expected_combo.output)
|
|
):
|
|
raise RuntimeError(
|
|
f"Trace combo mismatch: got ({trace.combo.input},{trace.combo.output}), "
|
|
f"expected ({expected_combo.input},{expected_combo.output})"
|
|
)
|
|
if trace.s11.shape != (expected_points,) or trace.s21.shape != (expected_points,):
|
|
raise RuntimeError(
|
|
f"Trace ({trace.combo.input},{trace.combo.output}) has shapes "
|
|
f"s11={trace.s11.shape}, s21={trace.s21.shape}"
|
|
)
|
|
if not np.all(np.isfinite(trace.s11.real)) or not np.all(np.isfinite(trace.s11.imag)):
|
|
raise RuntimeError(
|
|
f"Trace ({trace.combo.input},{trace.combo.output}) S11 contains non-finite values"
|
|
)
|
|
if not np.all(np.isfinite(trace.s21.real)) or not np.all(np.isfinite(trace.s21.imag)):
|
|
raise RuntimeError(
|
|
f"Trace ({trace.combo.input},{trace.combo.output}) S21 contains non-finite values"
|
|
)
|
|
|
|
|
|
def main() -> int:
|
|
"""Acquire one SN9000 collection and validate it end to end.
|
|
|
|
Opens the VISA session, configures the sweep, validates the collection, checks the SCPI
|
|
error queue, and prints a summary; raises on any validation or SCPI failure.
|
|
"""
|
|
args = _parse_args()
|
|
sweep = RadarSweepModel(
|
|
start_hz=args.start_hz,
|
|
stop_hz=args.stop_hz,
|
|
points=args.points,
|
|
if_bandwidth_hz=args.ifbw_hz,
|
|
power_dbm=args.power_dbm,
|
|
)
|
|
|
|
service = Sn9000Service(
|
|
host=args.host,
|
|
port=args.port,
|
|
timeout_ms=args.timeout_ms,
|
|
preset_on_open=not args.no_preset,
|
|
visa_library=args.visa_library,
|
|
)
|
|
try:
|
|
service.open()
|
|
print(f"SN9000 IDN: {service.query_identity()}")
|
|
service.configure(sweep)
|
|
collection = service.acquire_collection(collection_id=1)
|
|
_validate_collection(collection, args.points)
|
|
system_error = service.query_system_error()
|
|
if not system_error.startswith("0,"):
|
|
raise RuntimeError(f"SN9000 SCPI error after sweep: {system_error}")
|
|
mean_abs_s21 = np.mean([float(np.mean(np.abs(trace.s21))) for trace in collection.traces])
|
|
print(
|
|
"SN9000 collection OK: "
|
|
f"traces={len(collection.traces)}, points={args.points}, "
|
|
f"first_hz={collection.traces[0].frequency_hz[0]:.3f}, "
|
|
f"last_hz={collection.traces[0].frequency_hz[-1]:.3f}, "
|
|
f"mean_abs_s21={mean_abs_s21:.6g}"
|
|
)
|
|
finally:
|
|
service.close()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|