155 lines
5.2 KiB
Python
155 lines
5.2 KiB
Python
"""Helpers for loading radar sweep variants from a directory of JSON files."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from python_app.models.run_config_model import RunConfigModel
|
|
from python_app.storage.npz_store import radar_key_from_config
|
|
|
|
_RADAR_SWEEP_KEYS = (
|
|
"start_hz",
|
|
"stop_hz",
|
|
"points",
|
|
"if_bandwidth_hz",
|
|
"stimulus_power_dbm",
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class RadarConfigVariant:
|
|
"""One capture-ready radar sweep variant loaded from JSON."""
|
|
|
|
source_path: Path
|
|
display_name: str
|
|
config: RunConfigModel
|
|
radar_key: str
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class RadarConfigScanSummary:
|
|
"""Scan results for one radar-config directory refresh."""
|
|
|
|
directory_path: str
|
|
json_file_count: int
|
|
valid_variant_count: int
|
|
skipped_file_count: int
|
|
duplicate_variant_count: int
|
|
issues: tuple[str, ...]
|
|
|
|
|
|
def scan_radar_config_variants(
|
|
directory_path: str,
|
|
*,
|
|
base_config: RunConfigModel,
|
|
) -> tuple[list[RadarConfigVariant], RadarConfigScanSummary]:
|
|
"""Load valid radar sweep variants from `directory_path` using `base_config` as the baseline."""
|
|
normalized_path = str(directory_path).strip()
|
|
if not normalized_path:
|
|
return [], RadarConfigScanSummary(
|
|
directory_path="",
|
|
json_file_count=0,
|
|
valid_variant_count=0,
|
|
skipped_file_count=0,
|
|
duplicate_variant_count=0,
|
|
issues=(),
|
|
)
|
|
|
|
directory = Path(normalized_path).expanduser()
|
|
if not directory.exists():
|
|
return [], RadarConfigScanSummary(
|
|
directory_path=str(directory),
|
|
json_file_count=0,
|
|
valid_variant_count=0,
|
|
skipped_file_count=0,
|
|
duplicate_variant_count=0,
|
|
issues=(f"Directory does not exist: {directory}",),
|
|
)
|
|
if not directory.is_dir():
|
|
return [], RadarConfigScanSummary(
|
|
directory_path=str(directory),
|
|
json_file_count=0,
|
|
valid_variant_count=0,
|
|
skipped_file_count=0,
|
|
duplicate_variant_count=0,
|
|
issues=(f"Path is not a directory: {directory}",),
|
|
)
|
|
|
|
variants: list[RadarConfigVariant] = []
|
|
issues: list[str] = []
|
|
duplicate_variant_count = 0
|
|
seen_radar_keys: set[str] = set()
|
|
json_paths = sorted(path for path in directory.glob("*.json") if path.is_file())
|
|
|
|
for path in json_paths:
|
|
try:
|
|
variant = _load_radar_config_variant(path, base_config=base_config)
|
|
except Exception as exc: # noqa: BLE001
|
|
issues.append(f"{path.name}: {type(exc).__name__}: {exc}")
|
|
continue
|
|
if variant.radar_key in seen_radar_keys:
|
|
duplicate_variant_count += 1
|
|
issues.append(
|
|
f"{path.name}: duplicate radar variant key {variant.radar_key}; keeping the first matching file only"
|
|
)
|
|
continue
|
|
seen_radar_keys.add(variant.radar_key)
|
|
variants.append(variant)
|
|
|
|
return variants, RadarConfigScanSummary(
|
|
directory_path=str(directory),
|
|
json_file_count=len(json_paths),
|
|
valid_variant_count=len(variants),
|
|
skipped_file_count=max(0, len(json_paths) - len(variants) - duplicate_variant_count),
|
|
duplicate_variant_count=duplicate_variant_count,
|
|
issues=tuple(issues),
|
|
)
|
|
|
|
|
|
def _load_radar_config_variant(path: Path, *, base_config: RunConfigModel) -> RadarConfigVariant:
|
|
"""Load one radar sweep variant by overlaying JSON `radar.sweep` onto `base_config`."""
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
if not isinstance(payload, dict):
|
|
raise ValueError("JSON root must be an object")
|
|
|
|
radar_payload = payload.get("radar")
|
|
if not isinstance(radar_payload, dict):
|
|
raise ValueError("Missing radar section")
|
|
|
|
sweep_payload = radar_payload.get("sweep")
|
|
if not isinstance(sweep_payload, dict):
|
|
raise ValueError("Missing radar.sweep section")
|
|
if not any(key in sweep_payload for key in _RADAR_SWEEP_KEYS):
|
|
raise ValueError("radar.sweep does not contain any supported sweep keys")
|
|
|
|
config = base_config.clone()
|
|
sweep = config.radar.sweep
|
|
if "start_hz" in sweep_payload:
|
|
sweep.start_hz = float(sweep_payload["start_hz"])
|
|
if "stop_hz" in sweep_payload:
|
|
sweep.stop_hz = float(sweep_payload["stop_hz"])
|
|
if "points" in sweep_payload:
|
|
sweep.points = int(sweep_payload["points"])
|
|
if "if_bandwidth_hz" in sweep_payload:
|
|
sweep.if_bandwidth_hz = float(sweep_payload["if_bandwidth_hz"])
|
|
if "stimulus_power_dbm" in sweep_payload:
|
|
sweep.power_dbm = float(sweep_payload["stimulus_power_dbm"])
|
|
|
|
radar_key = radar_key_from_config(
|
|
model_name=config.radar.model,
|
|
serial=config.radar.serial,
|
|
sweep_start_hz=config.radar.sweep.start_hz,
|
|
sweep_stop_hz=config.radar.sweep.stop_hz,
|
|
sweep_points=config.radar.sweep.points,
|
|
ifbw_hz=config.radar.sweep.if_bandwidth_hz,
|
|
power_dbm=config.radar.sweep.power_dbm,
|
|
)
|
|
return RadarConfigVariant(
|
|
source_path=path,
|
|
display_name=path.stem,
|
|
config=config,
|
|
radar_key=radar_key,
|
|
)
|