91 lines
3.2 KiB
Python
91 lines
3.2 KiB
Python
"""Path and naming helpers for NPZ snapshot storage."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from collections.abc import Sequence
|
|
|
|
|
|
def radar_key_from_config(
|
|
model_name: str,
|
|
serial: str,
|
|
sweep_start_hz: float,
|
|
sweep_stop_hz: float,
|
|
sweep_points: int,
|
|
ifbw_hz: float,
|
|
power_dbm: float,
|
|
extra_serials: Sequence[str] | None = None,
|
|
) -> str:
|
|
"""Build deterministic key for calibration/reference set lookup."""
|
|
serial_parts = [serial or "no_serial"]
|
|
if extra_serials:
|
|
serial_parts.extend(str(value).strip() or "no_serial" for value in extra_serials)
|
|
serial_part = "_".join(sanitize_path_component(value) for value in serial_parts)
|
|
start_token = format_float_for_key(sweep_start_hz)
|
|
stop_token = format_float_for_key(sweep_stop_hz)
|
|
points_token = "adc" if model_name.strip().lower() == "kamil_adc" else str(int(sweep_points))
|
|
ifbw_token = format_float_for_key(ifbw_hz)
|
|
power_token = format_float_for_key(power_dbm)
|
|
return (
|
|
f"{model_name}_{serial_part}"
|
|
f"_st{start_token}_sp{stop_token}"
|
|
f"_p{points_token}_if{ifbw_token}_pw{power_token}"
|
|
)
|
|
|
|
|
|
def radar_config_filename_prefix(
|
|
model_name: str,
|
|
sweep_start_hz: float,
|
|
sweep_stop_hz: float,
|
|
sweep_points: int,
|
|
power_dbm: float,
|
|
) -> str:
|
|
"""Build a readable, filesystem-safe prefix describing the radar acquisition setup.
|
|
|
|
Encodes the radar model, sweep frequency span, point count, and power so a saved
|
|
dataset's name carries the configuration it was captured with. ADC-mode radars have
|
|
no fixed sweep, so the point count is reported as ``adc``.
|
|
"""
|
|
points_token = "adc" if model_name.strip().lower() == "kamil_adc" else f"{int(sweep_points)}pts"
|
|
tokens = (
|
|
model_name,
|
|
_format_frequency_span(sweep_start_hz, sweep_stop_hz),
|
|
points_token,
|
|
f"{power_dbm:g}dBm",
|
|
)
|
|
return sanitize_path_component("_".join(tokens))
|
|
|
|
|
|
def _format_frequency_span(start_hz: float, stop_hz: float) -> str:
|
|
"""Format a frequency span compactly, both bounds sharing the larger bound's unit."""
|
|
unit, scale = _frequency_unit(max(start_hz, stop_hz))
|
|
return f"{start_hz / scale:g}-{stop_hz / scale:g}{unit}"
|
|
|
|
|
|
def _frequency_unit(hz: float) -> tuple[str, float]:
|
|
"""Pick the largest unit (GHz/MHz/kHz/Hz) that keeps the magnitude at or above one."""
|
|
for unit, scale in (("GHz", 1e9), ("MHz", 1e6), ("kHz", 1e3)):
|
|
if hz >= scale:
|
|
return unit, scale
|
|
return "Hz", 1.0
|
|
|
|
|
|
def collection_dir_name(index: int, collection_id: int, monotonic_ns: int) -> str:
|
|
"""Build canonical collection directory name for snapshot stages."""
|
|
return f"{index:04d}_id{int(collection_id)}_ns{int(monotonic_ns)}"
|
|
|
|
|
|
def sanitize_path_component(name: str) -> str:
|
|
"""Convert arbitrary string into filesystem-safe component."""
|
|
cleaned = re.sub(r"[^A-Za-z0-9._-]+", "_", name.strip())
|
|
cleaned = cleaned.strip("._")
|
|
return cleaned or "snapshot"
|
|
|
|
|
|
def format_float_for_key(value: float) -> str:
|
|
"""Format float compactly for deterministic key serialization."""
|
|
integer = int(round(value))
|
|
if abs(value - float(integer)) < 1e-6:
|
|
return str(integer)
|
|
return f"{value:.6f}".rstrip("0").rstrip(".")
|