awe/radar_system
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

bscan added s11

Browse Source
This commit is contained in:
Ayzen
2026-03-26 18:47:07 +03:00
parent 077542cbd0
commit cd13a891ca
5 changed files with 80 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python_app/gui/controllers/app_window_snapshot_mixin.py
+3
View File
@@ -60,6 +60,7 @@ class AppWindowSnapshotMixin:
last_n = int(self._save_count.value())
input_index = int(self._vna_json_input_index.value())
output_index = int(self._vna_json_output_index.value())
channel = self._vna_json_channel.currentText()
output_root = Path(self._save_path_input.text().strip()).expanduser()
output_name = self._save_name_input.text().strip()
output_path, summary = self._store.save_runtime_vna_history_json(
@@ -71,6 +72,7 @@ class AppWindowSnapshotMixin:
last_n,
input_index=input_index,
output_index=output_index,
channel=channel,
primary_stage="preprocessed",
)
self._log(
@@ -85,6 +87,7 @@ class AppWindowSnapshotMixin:
f"anchor={summary.get('anchor_stage', 'unknown')}, "
f"input={input_index}, "
f"output={output_index}, "
f"channel={channel}, "
f"requested_last_n={last_n})"
)
except Exception as exc: # noqa: BLE001
python_app/gui/controllers/sections/data_actions_section.py
+5 -1
View File
@@ -3,7 +3,7 @@
from __future__ import annotations
from PyQt6.QtCore import Qt
from PyQt6.QtWidgets import QGroupBox, QHBoxLayout, QLabel, QLineEdit, QPushButton, QSpinBox, QVBoxLayout
from PyQt6.QtWidgets import QComboBox, QGroupBox, QHBoxLayout, QLabel, QLineEdit, QPushButton, QSpinBox, QVBoxLayout
def build_data_actions_group(owner) -> QGroupBox:
@@ -32,6 +32,8 @@ def build_data_actions_group(owner) -> QGroupBox:
owner._vna_json_output_index.setMinimum(0)
owner._vna_json_output_index.setMaximum(65_535)
owner._vna_json_output_index.setValue(0)
owner._vna_json_channel = QComboBox()
owner._vna_json_channel.addItems(["s21", "s11"])
button_column = QVBoxLayout()
button_column.setSpacing(8)
@@ -54,6 +56,8 @@ def build_data_actions_group(owner) -> QGroupBox:
json_row.addWidget(owner._vna_json_input_index)
json_row.addWidget(QLabel("output"))
json_row.addWidget(owner._vna_json_output_index)
json_row.addWidget(QLabel("channel"))
json_row.addWidget(owner._vna_json_channel)
json_row.addStretch(1)
layout.addLayout(json_row)
python_app/scripts/convert_snapshot_to_vna_history.py
+38 -13
View File
@@ -22,7 +22,7 @@ class TraceRecord:
monotonic_ns: int
stage_index: int
frequency_hz: np.ndarray
s21: np.ndarray
samples: np.ndarray
@dataclass(frozen=True)
@@ -64,12 +64,20 @@ def _pick_trace_meta(meta: dict[str, Any], input_index: int, output_index: int)
return None
def _normalize_channel(channel: str) -> str:
normalized = str(channel).strip().lower()
if normalized not in {"s21", "s11"}:
raise ValueError("channel must be either 's21' or 's11'")
return normalized
def _load_stage_records(
snapshot_dir: Path,
stage: str,
*,
input_index: int,
output_index: int,
channel: str,
) -> list[TraceRecord]:
stage_dir = snapshot_dir / stage
records: list[TraceRecord] = []
@@ -85,20 +93,20 @@ def _load_stage_records(
continue
freq_file = str(trace_meta.get("freq_file", ""))
s21_file = str(trace_meta.get("s21_file", ""))
if not freq_file or not s21_file:
samples_file = str(trace_meta.get(f"{channel}_file", ""))
if not freq_file or not samples_file:
continue
frequency_hz = np.asarray(np.load(collection_dir / freq_file), dtype=np.float64).reshape(-1)
s21 = np.asarray(np.load(collection_dir / s21_file), dtype=np.complex128).reshape(-1)
if frequency_hz.shape != s21.shape:
raise ValueError(f"Shape mismatch in {collection_dir}: freq{frequency_hz.shape} vs s21{s21.shape}")
samples = np.asarray(np.load(collection_dir / samples_file), dtype=np.complex128).reshape(-1)
if frequency_hz.shape != samples.shape:
raise ValueError(f"Shape mismatch in {collection_dir}: freq{frequency_hz.shape} vs {channel}{samples.shape}")
if frequency_hz.size == 0:
continue
if not (
np.isfinite(frequency_hz).all()
and np.isfinite(np.real(s21)).all()
and np.isfinite(np.imag(s21)).all()
and np.isfinite(np.real(samples)).all()
and np.isfinite(np.imag(samples)).all()
):
raise ValueError(f"Non-finite values in {collection_dir}")
@@ -109,7 +117,7 @@ def _load_stage_records(
monotonic_ns=int(meta.get("monotonic_ns", 0)),
stage_index=_parse_stage_index(collection_dir.name, fallback_idx),
frequency_hz=frequency_hz,
s21=s21,
samples=samples,
)
)
@@ -157,6 +165,7 @@ def _build_sweep_history(
raw_records: list[TraceRecord],
preprocessed_records: list[TraceRecord],
*,
channel: str,
primary_stage: str,
) -> list[dict[str, Any]]:
raw_map, raw_order = _index_by_collection_occurrence(raw_records)
@@ -188,11 +197,11 @@ def _build_sweep_history(
history.append(
{
"timestamp": timestamp_sec,
"sweep_points": _complex_to_points(sweep_source.s21),
"calibrated_points": _complex_to_points(calibrated_source.s21),
"sweep_points": _complex_to_points(sweep_source.samples),
"calibrated_points": _complex_to_points(calibrated_source.samples),
"reference_points": [],
"vna_config": {
"mode": "s11",
"mode": channel,
"start_freq": start_freq_hz,
"stop_freq": stop_freq_hz,
"points": int(base.frequency_hz.size),
@@ -257,6 +266,12 @@ def _build_parser() -> argparse.ArgumentParser:
)
parser.add_argument("--input", dest="input_index", type=int, default=0, help="Input switch index to export.")
parser.add_argument("--output-index", dest="output_index", type=int, default=0, help="Output switch index to export.")
parser.add_argument(
"--channel",
choices=("s21", "s11"),
default="s21",
help="Trace channel to export into sweep/calibrated points.",
)
parser.add_argument(
"--primary-stage",
choices=("preprocessed", "raw"),
@@ -275,6 +290,7 @@ def _build_parser() -> argparse.ArgumentParser:
def main() -> None:
parser = _build_parser()
args = parser.parse_args()
channel = _normalize_channel(args.channel)
snapshot_dir = args.snapshot_dir.expanduser().resolve()
if not snapshot_dir.exists():
@@ -291,12 +307,14 @@ def main() -> None:
"raw",
input_index=args.input_index,
output_index=args.output_index,
channel=channel,
)
preprocessed_records = _load_stage_records(
snapshot_dir,
"preprocessed",
input_index=args.input_index,
output_index=args.output_index,
channel=channel,
)
if not raw_records and not preprocessed_records:
raise ValueError(
@@ -304,7 +322,12 @@ def main() -> None:
f"for input={args.input_index}, output={args.output_index}."
)
sweep_history = _build_sweep_history(raw_records, preprocessed_records, primary_stage=args.primary_stage)
sweep_history = _build_sweep_history(
raw_records,
preprocessed_records,
channel=channel,
primary_stage=args.primary_stage,
)
if args.last_n > 0:
sweep_history = sweep_history[-args.last_n :]
if not sweep_history:
@@ -324,6 +347,7 @@ def main() -> None:
"source_snapshot_dir": str(snapshot_dir),
"input_index": int(args.input_index),
"output_index": int(args.output_index),
"channel": channel,
"primary_stage": args.primary_stage,
"raw_record_count": len(raw_records),
"preprocessed_record_count": len(preprocessed_records),
@@ -341,6 +365,7 @@ def main() -> None:
"Converted snapshot to vna_system history JSON:\n"
f" input snapshot: {snapshot_dir}\n"
f" output file: {output_path}\n"
f" channel: {channel}\n"
f" sweeps written: {len(sweep_history)}\n"
f" raw records: {len(raw_records)}\n"
f" pre records: {len(preprocessed_records)}"
python_app/storage/npz/store.py
+3
View File
@@ -225,6 +225,7 @@ class NpzStore(StoreApi):
*,
input_index: int = 0,
output_index: int = 0,
channel: str = "s21",
primary_stage: str = "preprocessed",
) -> tuple[Path, dict[str, Any]]:
"""Save runtime history as vna_system-compatible JSON file."""
@@ -251,6 +252,7 @@ class NpzStore(StoreApi):
selected_results,
input_index=input_index,
output_index=output_index,
channel=channel,
primary_stage=primary_stage,
)
output_path.write_text(
@@ -267,6 +269,7 @@ class NpzStore(StoreApi):
summary["sweep_count"] = len(payload.get("sweep_history", []))
summary["input_index"] = int(input_index)
summary["output_index"] = int(output_index)
summary["channel"] = str(channel)
summary["primary_stage"] = str(primary_stage)
summary["output_path"] = str(output_path)
return output_path, summary
python_app/storage/npz/vna_history_json.py
+31 -11
View File
@@ -21,7 +21,20 @@ class TraceRecord:
monotonic_ns: int
stage_index: int
frequency_hz: np.ndarray
s21: np.ndarray
samples: np.ndarray
def _normalize_channel(channel: str) -> str:
normalized = str(channel).strip().lower()
if normalized not in {"s21", "s11"}:
raise ValueError("channel must be either 's21' or 's11'")
return normalized
def _select_trace_samples(trace: TraceData, channel: str) -> np.ndarray:
if channel == "s11":
return np.asarray(trace.s11, dtype=np.complex128).reshape(-1)
return np.asarray(trace.s21, dtype=np.complex128).reshape(-1)
def _pick_trace(collection: SweepCollection, input_index: int, output_index: int) -> TraceData | None:
@@ -37,6 +50,7 @@ def _build_stage_records(
*,
input_index: int,
output_index: int,
channel: str,
) -> list[TraceRecord]:
records: list[TraceRecord] = []
for stage_index, collection in enumerate(history):
@@ -45,18 +59,18 @@ def _build_stage_records(
continue
frequency_hz = np.asarray(trace.frequency_hz, dtype=np.float64).reshape(-1)
s21 = np.asarray(trace.s21, dtype=np.complex128).reshape(-1)
if frequency_hz.shape != s21.shape:
samples = _select_trace_samples(trace, channel)
if frequency_hz.shape != samples.shape:
raise ValueError(
f"Shape mismatch in {stage} stage for collection_id={collection.collection_id}: "
f"freq{frequency_hz.shape} vs s21{s21.shape}"
f"freq{frequency_hz.shape} vs {channel}{samples.shape}"
)
if frequency_hz.size == 0:
continue
if not (
np.isfinite(frequency_hz).all()
and np.isfinite(np.real(s21)).all()
and np.isfinite(np.imag(s21)).all()
and np.isfinite(np.real(samples)).all()
and np.isfinite(np.imag(samples)).all()
):
raise ValueError(f"Non-finite values in {stage} stage for collection_id={collection.collection_id}")
@@ -67,7 +81,7 @@ def _build_stage_records(
monotonic_ns=int(collection.monotonic_ns),
stage_index=int(stage_index),
frequency_hz=frequency_hz,
s21=s21,
samples=samples,
)
)
return records
@@ -94,6 +108,7 @@ def _build_sweep_history(
raw_records: list[TraceRecord],
preprocessed_records: list[TraceRecord],
*,
channel: str,
primary_stage: str,
) -> list[dict[str, Any]]:
raw_map, raw_order = _index_by_collection_occurrence(raw_records)
@@ -124,11 +139,11 @@ def _build_sweep_history(
history.append(
{
"timestamp": timestamp_sec,
"sweep_points": _complex_to_points(sweep_source.s21),
"calibrated_points": _complex_to_points(calibrated_source.s21),
"sweep_points": _complex_to_points(sweep_source.samples),
"calibrated_points": _complex_to_points(calibrated_source.samples),
"reference_points": [],
"vna_config": {
"mode": "s11",
"mode": channel,
"start_freq": start_freq_hz,
"stop_freq": stop_freq_hz,
"points": int(base.frequency_hz.size),
@@ -178,9 +193,11 @@ def build_vna_history_payload(
*,
input_index: int,
output_index: int,
channel: str = "s21",
primary_stage: str = "preprocessed",
) -> dict[str, Any]:
"""Build vna_system-compatible history JSON payload from runtime histories."""
channel = _normalize_channel(channel)
if primary_stage not in {"preprocessed", "raw"}:
raise ValueError("primary_stage must be either 'preprocessed' or 'raw'")
@@ -189,12 +206,14 @@ def build_vna_history_payload(
raw_history,
input_index=input_index,
output_index=output_index,
channel=channel,
)
preprocessed_records = _build_stage_records(
"preprocessed",
preprocessed_history,
input_index=input_index,
output_index=output_index,
channel=channel,
)
if not raw_records and not preprocessed_records:
raise ValueError(
@@ -205,6 +224,7 @@ def build_vna_history_payload(
sweep_history = _build_sweep_history(
raw_records,
preprocessed_records,
channel=channel,
primary_stage=primary_stage,
)
if not sweep_history:
@@ -217,6 +237,7 @@ def build_vna_history_payload(
"source_snapshot_dir": "<runtime_history>",
"input_index": int(input_index),
"output_index": int(output_index),
"channel": channel,
"primary_stage": str(primary_stage),
"raw_record_count": len(raw_records),
"preprocessed_record_count": len(preprocessed_records),
@@ -228,4 +249,3 @@ def build_vna_history_payload(
payload["alignment_warning"] = alignment_warning
return payload