76 lines
2.9 KiB
Python
76 lines
2.9 KiB
Python
"""Smoke test for a remote Compact-M K209 server."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
|
if str(PROJECT_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
|
|
from python_app.hardware_full.k209_remote_protocol import DEFAULT_REMOTE_HOST, DEFAULT_REMOTE_PORT
|
|
from python_app.hardware_full.remote_compact_m_k209_service import RemoteCompactMK209Service
|
|
from python_app.models.run_config_model import RadarSweepModel
|
|
|
|
|
|
def _parse_args() -> argparse.Namespace:
|
|
"""Parse command-line options for the remote K209 smoke test."""
|
|
parser = argparse.ArgumentParser(description="Validate remote K209 connection and one sweep.")
|
|
parser.add_argument("--host", default=DEFAULT_REMOTE_HOST, help="K209 remote server host.")
|
|
parser.add_argument("--port", type=int, default=DEFAULT_REMOTE_PORT, help="K209 remote server port.")
|
|
parser.add_argument("--start-hz", type=float, default=10_000_000.0)
|
|
parser.add_argument("--stop-hz", type=float, default=100_000_000.0)
|
|
parser.add_argument("--points", type=int, default=11)
|
|
parser.add_argument("--ifbw-hz", type=float, default=10_000.0)
|
|
parser.add_argument("--power-dbm", type=float, default=-20.0)
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> int:
|
|
"""Connect to a remote K209, run one sweep, and validate its shape and values.
|
|
|
|
Raises on an unexpected point count, non-finite samples, or a non-monotonic frequency
|
|
axis; prints a one-line summary on success.
|
|
"""
|
|
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 = RemoteCompactMK209Service(host=args.host, port=args.port)
|
|
try:
|
|
service.open()
|
|
print(f"K209 IDN: {service.query_identity()}")
|
|
service.configure(sweep)
|
|
result = service.acquire()
|
|
finally:
|
|
service.close()
|
|
|
|
s11 = result.trace("s11")
|
|
s21 = result.trace("s21")
|
|
if result.x.size != args.points or s11.size != args.points or s21.size != args.points:
|
|
raise RuntimeError("Remote K209 sweep returned an unexpected point count")
|
|
if not np.all(np.isfinite(result.x)) or not np.all(np.isfinite(s11)) or not np.all(np.isfinite(s21)):
|
|
raise RuntimeError("Remote K209 sweep contains non-finite values")
|
|
if not np.all(np.diff(result.x) >= 0):
|
|
raise RuntimeError("Remote K209 frequency axis is not monotonic")
|
|
|
|
print(
|
|
"Remote K209 sweep OK: "
|
|
f"points={args.points}, first_hz={result.x[0]:.3f}, last_hz={result.x[-1]:.3f}, "
|
|
f"mean_abs_s11={float(np.mean(np.abs(s11))):.6g}, "
|
|
f"mean_abs_s21={float(np.mean(np.abs(s21))):.6g}"
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|