62 lines
2.2 KiB
Python
62 lines
2.2 KiB
Python
"""Live processing settings model and atomic JSON writer."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import json
|
|
from pathlib import Path
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class ProcessingLiveConfig:
|
|
"""Runtime-adjustable processing parameters shared with data processor."""
|
|
|
|
processor_mode: str = "pass_through"
|
|
gain_db: float = 0.0
|
|
phase_deg: float = 0.0
|
|
bscan_axis: str = "abs"
|
|
bscan_cut_m: float = 0.824
|
|
bscan_max_depth_m: float = 1.0
|
|
bscan_gain: float = 1.0
|
|
bscan_start_freq_mhz: float = 100.0
|
|
bscan_stop_freq_mhz: float = 8800.0
|
|
history_command_seq: int = 0
|
|
history_command: str = "none"
|
|
|
|
def to_dict(self) -> dict[str, float | str | int]:
|
|
"""Convert live config to JSON-serializable dictionary."""
|
|
return {
|
|
"processor_mode": str(self.processor_mode),
|
|
"gain_db": float(self.gain_db),
|
|
"phase_deg": float(self.phase_deg),
|
|
"bscan_axis": str(self.bscan_axis),
|
|
"bscan_cut_m": float(self.bscan_cut_m),
|
|
"bscan_max_depth_m": float(self.bscan_max_depth_m),
|
|
"bscan_gain": float(self.bscan_gain),
|
|
"bscan_start_freq_mhz": float(self.bscan_start_freq_mhz),
|
|
"bscan_stop_freq_mhz": float(self.bscan_stop_freq_mhz),
|
|
"history_command_seq": int(self.history_command_seq),
|
|
"history_command": str(self.history_command),
|
|
}
|
|
|
|
|
|
class ProcessingLiveConfigWriter:
|
|
"""Atomic writer for processing live-config file."""
|
|
|
|
def __init__(self, config_path: Path) -> None:
|
|
"""Create writer targeting `config_path`."""
|
|
self._config_path = config_path
|
|
self._config_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
@property
|
|
def path(self) -> Path:
|
|
"""Return destination config path."""
|
|
return self._config_path
|
|
|
|
def write(self, config: ProcessingLiveConfig) -> Path:
|
|
"""Atomically write config by temp-file replace."""
|
|
temp_path = self._config_path.with_suffix(self._config_path.suffix + ".tmp")
|
|
temp_path.write_text(json.dumps(config.to_dict(), indent=2), encoding="utf-8")
|
|
temp_path.replace(self._config_path)
|
|
return self._config_path
|