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

added b-scan

Browse Source
This commit is contained in:
Ayzen
2025-09-29 19:07:26 +03:00
parent 3b1aa988de
commit f4e223ca96
15 changed files with 1945 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files

232
vna_system/api/endpoints/settings.py
View File

@ -19,6 +19,9 @@ from vna_system.api.models.settings import (
SetCalibrationRequest,
RemoveStandardRequest,
WorkingCalibrationModel,
ReferenceModel,
CreateReferenceRequest,
SetReferenceRequest,
)
router = APIRouter(prefix="/api/v1/settings", tags=["settings"])
@ -36,19 +39,10 @@ async def get_status() -> dict[str, Any]:
@router.get("/presets", response_model=List[PresetModel])
async def get_presets(mode: str | None = None) -> list[PresetModel]:
async def get_presets() -> list[PresetModel]:
"""Get all available configuration presets, optionally filtered by mode."""
try:
if mode:
from vna_system.core.settings.preset_manager import VNAMode
try:
vna_mode = VNAMode(mode.lower())
except ValueError:
raise HTTPException(status_code=400, detail=f"Invalid mode: {mode}")
presets = singletons.settings_manager.get_presets_by_mode(vna_mode)
else:
presets = singletons.settings_manager.get_available_presets()
presets = singletons.settings_manager.get_available_presets()
return [
PresetModel(
@ -61,8 +55,6 @@ async def get_presets(mode: str | None = None) -> list[PresetModel]:
)
for p in presets
]
except HTTPException:
raise
except Exception as exc: # noqa: BLE001
logger.error("Failed to list presets")
raise HTTPException(status_code=500, detail=str(exc))
@ -415,3 +407,217 @@ async def get_working_calibration_standards_plots() -> dict[str, Any]:
except Exception as exc: # noqa: BLE001
logger.error("Failed to build working calibration standards plots")
raise HTTPException(status_code=500, detail=str(exc))
# ================================================================ #
# Reference Management Endpoints
# ================================================================ #
@router.get("/references", response_model=List[ReferenceModel])
async def get_references(preset_filename: str | None = None) -> List[dict[str, Any]]:
"""Get available references for a preset."""
try:
settings_manager = singletons.settings_manager
if preset_filename:
preset = next(
(p for p in settings_manager.get_available_presets() if p.filename == preset_filename),
None
)
if not preset:
raise HTTPException(status_code=404, detail=f"Preset '{preset_filename}' not found")
else:
preset = settings_manager.get_current_preset()
if not preset:
raise HTTPException(status_code=400, detail="No current preset set and no preset_filename provided")
references = settings_manager.get_available_references(preset)
return [
{
"name": ref.name,
"timestamp": ref.timestamp.isoformat(),
"preset_filename": ref.preset_filename,
"description": ref.description,
"metadata": ref.metadata
}
for ref in references
]
except HTTPException:
raise
except Exception as exc: # noqa: BLE001
logger.error("Failed to get references", error=repr(exc))
raise HTTPException(status_code=500, detail=str(exc))
@router.get("/reference/current", response_model=ReferenceModel | None)
async def get_current_reference(preset_filename: str | None = None) -> dict[str, Any] | None:
"""Get the currently selected reference."""
try:
settings_manager = singletons.settings_manager
if preset_filename:
preset = next(
(p for p in settings_manager.get_available_presets() if p.filename == preset_filename),
None
)
if not preset:
raise HTTPException(status_code=404, detail=f"Preset '{preset_filename}' not found")
else:
preset = settings_manager.get_current_preset()
if not preset:
raise HTTPException(status_code=400, detail="No current preset set and no preset_filename provided")
current_ref = settings_manager.get_current_reference(preset)
if not current_ref:
return None
return {
"name": current_ref.name,
"timestamp": current_ref.timestamp.isoformat(),
"preset_filename": current_ref.preset_filename,
"description": current_ref.description,
"metadata": current_ref.metadata
}
except HTTPException:
raise
except Exception as exc: # noqa: BLE001
logger.error("Failed to get current reference", error=repr(exc))
raise HTTPException(status_code=500, detail=str(exc))
@router.post("/reference/create")
async def create_reference(request: CreateReferenceRequest) -> dict[str, Any]:
"""Create a new reference from current sweep data."""
try:
settings_manager = singletons.settings_manager
data_acquisition = singletons.vna_data_acquisition_instance
if request.preset_filename:
preset = next(
(p for p in settings_manager.get_available_presets() if p.filename == request.preset_filename),
None
)
if not preset:
raise HTTPException(status_code=404, detail=f"Preset '{request.preset_filename}' not found")
else:
preset = settings_manager.get_current_preset()
if not preset:
raise HTTPException(status_code=400, detail="No current preset set and no preset_filename provided")
# Create reference using the new capture method
reference_info = settings_manager.capture_reference_from_acquisition(
reference_name=request.name,
data_acquisition=data_acquisition,
description=request.description,
metadata=request.metadata,
preset=preset
)
return {
"success": True,
"message": f"Reference '{request.name}' created successfully",
"reference": {
"name": reference_info.name,
"timestamp": reference_info.timestamp.isoformat(),
"preset_filename": reference_info.preset_filename,
"description": reference_info.description,
"metadata": reference_info.metadata
}
}
except HTTPException:
raise
except Exception as exc: # noqa: BLE001
logger.error("Failed to create reference", name=request.name, error=repr(exc))
raise HTTPException(status_code=500, detail=str(exc))
@router.post("/reference/set")
async def set_current_reference(request: SetReferenceRequest) -> dict[str, Any]:
"""Set the current reference."""
try:
settings_manager = singletons.settings_manager
if request.preset_filename:
preset = next(
(p for p in settings_manager.get_available_presets() if p.filename == request.preset_filename),
None
)
if not preset:
raise HTTPException(status_code=404, detail=f"Preset '{request.preset_filename}' not found")
else:
preset = settings_manager.get_current_preset()
if not preset:
raise HTTPException(status_code=400, detail="No current preset set and no preset_filename provided")
success = settings_manager.set_current_reference(request.name, preset)
if not success:
raise HTTPException(status_code=404, detail=f"Reference '{request.name}' not found")
return {
"success": True,
"message": f"Reference '{request.name}' set as current"
}
except HTTPException:
raise
except Exception as exc: # noqa: BLE001
logger.error("Failed to set current reference", name=request.name, error=repr(exc))
raise HTTPException(status_code=500, detail=str(exc))
@router.delete("/reference/current")
async def clear_current_reference() -> dict[str, Any]:
"""Clear the current reference selection."""
try:
settings_manager = singletons.settings_manager
success = settings_manager.clear_current_reference()
if not success:
raise HTTPException(status_code=500, detail="Failed to clear current reference")
return {
"success": True,
"message": "Current reference cleared"
}
except HTTPException:
raise
except Exception as exc: # noqa: BLE001
logger.error("Failed to clear current reference", error=repr(exc))
raise HTTPException(status_code=500, detail=str(exc))
@router.delete("/reference/{reference_name}")
async def delete_reference(reference_name: str, preset_filename: str | None = None) -> dict[str, Any]:
"""Delete a reference."""
try:
settings_manager = singletons.settings_manager
if preset_filename:
preset = next(
(p for p in settings_manager.get_available_presets() if p.filename == preset_filename),
None
)
if not preset:
raise HTTPException(status_code=404, detail=f"Preset '{preset_filename}' not found")
else:
preset = settings_manager.get_current_preset()
if not preset:
raise HTTPException(status_code=400, detail="No current preset set and no preset_filename provided")
success = settings_manager.delete_reference(reference_name, preset)
if not success:
raise HTTPException(status_code=404, detail=f"Reference '{reference_name}' not found")
return {
"success": True,
"message": f"Reference '{reference_name}' deleted successfully"
}
except HTTPException:
raise
except Exception as exc: # noqa: BLE001
logger.error("Failed to delete reference", name=reference_name, error=repr(exc))
raise HTTPException(status_code=500, detail=str(exc))

23
vna_system/api/models/settings.py
View File

@ -56,4 +56,25 @@ class WorkingCalibrationModel(BaseModel):
progress: str | None = None
is_complete: bool | None = None
completed_standards: List[str] | None = None
missing_standards: List[str] | None = None
missing_standards: List[str] | None = None
# Reference models
class ReferenceModel(BaseModel):
name: str
timestamp: str
preset_filename: str
description: str
metadata: Dict[str, Any]
class CreateReferenceRequest(BaseModel):
name: str
description: str = ""
metadata: Dict[str, Any] = {}
preset_filename: str | None = None
class SetReferenceRequest(BaseModel):
name: str
preset_filename: str | None = None

2
vna_system/binary_input/current_input.bin
View File

@ -1 +1 @@
config_inputs/s21_start100_stop8800_points1000_bw1khz.bin
config_inputs/s11_start100_stop8800_points1000_bw1khz.bin

1
vna_system/calibration/current_calibration
View File

@ -1 +0,0 @@
s21_start100_stop8800_points1000_bw1khz/bambambum

2
vna_system/core/acquisition/data_acquisition.py
View File

@ -75,7 +75,7 @@ class VNADataAcquisition:
@property
def is_running(self) -> bool:
"""Return True if the acquisition thread is alive."""
return self._running and (self._thread is not None and self._thread.is_alive())
return self._running and (self._thread is not None and self._thread.is_alive()) and not self._paused
@property
def sweep_buffer(self) -> SweepBuffer:

5
vna_system/core/processors/base_processor.py
View File

@ -299,7 +299,7 @@ class BaseProcessor:
# --------------------------------------------------------------------- #
# Data path: accept new sweep, recompute, produce result
# --------------------------------------------------------------------- #
def add_sweep_data(self, sweep_data: Any, calibrated_data: Any, vna_config: ConfigPreset | None):
def add_sweep_data(self, sweep_data: Any, calibrated_data: Any, vna_config: ConfigPreset | None, reference_data: Any = None):
"""
Add the latest sweep to the in-memory history and trigger recalculation.
@ -311,6 +311,8 @@ class BaseProcessor:
Data post-calibration (structure is processor-specific).
vna_config:
Snapshot of VNA settings (dataclass or pydantic model supported).
reference_data:
Open air reference sweep data for background subtraction/normalization.
Returns
-------
@ -323,6 +325,7 @@ class BaseProcessor:
"sweep_data": sweep_data,
"calibrated_data": calibrated_data,
"vna_config": asdict(vna_config) if vna_config is not None else {},
"reference_data": reference_data,
"timestamp": datetime.now().timestamp(),
}
)

12
vna_system/core/processors/configs/bscan_config.json Normal file
View File

@ -0,0 +1,12 @@
{
"open_air": true,
"axis": "real",
"data_limitation": null,
"cut": 0.824,
"max": 1.1,
"gain": 1.2,
"start_freq": 100.0,
"stop_freq": 8800.0,
"clear_history": false,
"data_limit": 500
}

575
vna_system/core/processors/implementations/bscan_processor.py Normal file
View File

@ -0,0 +1,575 @@
from __future__ import annotations
from pathlib import Path
from typing import Any, Final
import numpy as np
from numpy.typing import NDArray
from vna_system.core.logging.logger import get_component_logger
from vna_system.core.processors.base_processor import BaseProcessor, UIParameter
from vna_system.core.acquisition.sweep_buffer import SweepData
logger = get_component_logger(__file__)
class BScanProcessor(BaseProcessor):
"""
B-Scan processor for time-domain analysis of VNA measurements.
Features
-------
- Optional open-air reference subtraction
- Configurable axis: "real", "abs", "phase"
- IFFT with frequency range filtering
- Depth windowing with gain shaping
- Plot history accumulation for multi-sweep heatmaps
"""
# Physical and display-related constants
SPEED_OF_LIGHT_M_S: Final[float] = 299_700_000.0 # m/s
def __init__(self, config_dir: Path) -> None:
super().__init__("bscan", config_dir)
# Increase history size for multi-sweep plotting
self.max_history = 10
# Local plot history (separate from sweep history maintained by BaseProcessor)
self._plot_history: list[dict[str, Any]] = []
self._max_plot_history = 50
logger.info("BScanProcessor initialized", processor_id=self.processor_id)
# -------------------------------------------------------------------------
# Configuration
# -------------------------------------------------------------------------
def _get_default_config(self) -> dict[str, Any]:
"""Return default configuration values."""
return {
"open_air": False, # Toggle for reference usage
"axis": "abs", # "real", "abs", or "phase"
"data_limitation": None, # None, "ph_only_1", "ph_only_2"
"cut": 0.824, # Cut parameter (meters)
"max": 1.0, # Max depth (meters)
"gain": 1.0, # Gain exponent
"start_freq": 100.0, # Start frequency (MHz)
"stop_freq": 8800.0, # Stop frequency (MHz)
"clear_history": False, # UI button; not persisted
}
def get_ui_parameters(self) -> list[UIParameter]:
"""Return UI parameter schema for configuration."""
cfg = self._config # Local alias for brevity
return [
UIParameter(
name="open_air",
label="Open Air Reference",
type="toggle",
value=cfg["open_air"],
),
UIParameter(
name="axis",
label="Axis",
type="select",
value=cfg["axis"],
options={"choices": ["real", "abs", "phase"]},
),
UIParameter(
name="data_limitation",
label="Data Limitation",
type="select",
value=cfg["data_limitation"],
options={"choices": [None, "ph_only_1", "ph_only_2"]},
),
UIParameter(
name="cut",
label="Cut (m)",
type="slider",
value=cfg["cut"],
options={"min": 0.0, "max": 2.0, "step": 0.001, "dtype": "float"},
),
UIParameter(
name="max",
label="Max Depth (m)",
type="slider",
value=cfg["max"],
options={"min": 0.1, "max": 5.0, "step": 0.1, "dtype": "float"},
),
UIParameter(
name="gain",
label="Gain Factor",
type="slider",
value=cfg["gain"],
options={"min": 0.0, "max": 3.0, "step": 0.1, "dtype": "float"},
),
UIParameter(
name="start_freq",
label="Start Frequency (MHz)",
type="slider",
value=cfg["start_freq"],
options={"min": 100.0, "max": 8800.0, "step": 10.0, "dtype": "float"},
),
UIParameter(
name="stop_freq",
label="Stop Frequency (MHz)",
type="slider",
value=cfg["stop_freq"],
options={"min": 100.0, "max": 8800.0, "step": 10.0, "dtype": "float"},
),
UIParameter(
name="clear_history",
label="Clear Plot History",
type="button",
value=False,
options={"action": "Clear accumulated plot history"},
),
]
def update_config(self, updates: dict[str, Any]) -> None:
"""Handle config updates, including the 'clear history' button action."""
if updates.get("clear_history"):
self._clear_plot_history()
# Remove button flag before delegating to base class
updates = {k: v for k, v in updates.items() if k != "clear_history"}
if updates:
super().update_config(updates)
def _clear_plot_history(self) -> None:
"""Clear the accumulated plot history."""
with self._lock:
self._plot_history.clear()
logger.info("Plot history cleared", processor_id=self.processor_id)
# -------------------------------------------------------------------------
# Processing
# -------------------------------------------------------------------------
def process_sweep(
self,
sweep_data: SweepData,
calibrated_data: SweepData | None,
vna_config: dict[str, Any],
) -> dict[str, Any]:
"""
Process a single sweep and prepare B-scan data.
Returns
-------
dict
Keys: time_domain_data, distance_data, frequency_range, reference_used,
axis_type, data_limitation, points_processed, plot_history_count
Or: {"error": "..."} on failure.
"""
try:
# Choose calibrated data when provided
data_to_process = calibrated_data or sweep_data
complex_data = self._get_complex_s11(data_to_process)
if complex_data is None or complex_data.size == 0:
logger.warning("No valid complex data for B-scan processing")
return {"error": "No valid complex data"}
# Optional reference subtraction (latest stored reference in sweep history)
reference_data: SweepData | None = None
if self._config["open_air"] and self._sweep_history:
latest_history = self._sweep_history[-1]
reference_data = latest_history.get("reference_data")
if self._config["open_air"] and reference_data is not None:
reference_complex = self._get_complex_s11(reference_data)
if reference_complex is not None and reference_complex.size:
complex_data = self._subtract_reference(complex_data, reference_complex)
logger.debug("Applied open-air reference subtraction")
# Keep frequency controls in sync with the current VNA config
self._update_frequency_ranges(vna_config)
# Perform the main analysis pipeline
analysis = self._perform_data_analysis(complex_data, vna_config)
if analysis is None:
logger.warning("Data analysis failed")
return {"error": "Data analysis failed"}
plot_record = {
"time_domain_data": analysis["time_data"].tolist(),
"distance_data": analysis["distance"].tolist(),
"sweep_number": sweep_data.sweep_number,
"timestamp": sweep_data.timestamp,
"frequency_range": analysis["freq_range"],
}
with self._lock:
self._plot_history.append(plot_record)
if len(self._plot_history) > self._max_plot_history:
self._plot_history = self._plot_history[-self._max_plot_history :]
return {
"time_domain_data": analysis["time_data"].tolist(),
"distance_data": analysis["distance"].tolist(),
"frequency_range": analysis["freq_range"],
"reference_used": bool(self._config["open_air"] and reference_data is not None),
"axis_type": self._config["axis"],
"data_limitation": self._config["data_limitation"],
"points_processed": int(complex_data.size),
"plot_history_count": len(self._plot_history),
}
except Exception as exc: # noqa: BLE001
logger.error("B-scan processing failed", error=repr(exc))
return {"error": str(exc)}
# -------------------------------------------------------------------------
# Visualization
# -------------------------------------------------------------------------
def generate_plotly_config(
self,
processed_data: dict[str, Any],
vna_config: dict[str, Any], # noqa: ARG002 - reserved for future layout tweaks
) -> dict[str, Any]:
"""
Produce a Plotly-compatible heatmap configuration from accumulated sweeps.
"""
if "error" in processed_data:
return {
"data": [],
"layout": {
"title": "B-Scan Analysis - Error",
"annotations": [
{
"text": f"Error: {processed_data['error']}",
"x": 0.5,
"y": 0.5,
"xref": "paper",
"yref": "paper",
"showarrow": False,
}
],
"template": "plotly_dark",
},
}
with self._lock:
history = list(self._plot_history)
if not history:
return {
"data": [],
"layout": {
"title": "B-Scan Analysis - No Data",
"xaxis": {"title": "Sweep Number"},
"yaxis": {"title": "Depth (m)"},
"template": "plotly_dark",
},
}
# Build scatter-like heatmap (irregular grid) from history
x_coords: list[int] = []
y_coords: list[float] = []
z_values: list[float] = []
for item in history:
sweep_num = item["sweep_number"]
depths = item["distance_data"]
amps = item["time_domain_data"]
for d, a in zip(depths, amps, strict=False):
x_coords.append(sweep_num)
y_coords.append(d)
z_values.append(a)
# Colorscale selection
if self._config["axis"] == "abs":
colorscale = "Viridis"
heatmap_kwargs: dict[str, Any] = {}
else:
colorscale = "RdBu"
heatmap_kwargs = {"zmid": 0}
heatmap_trace = {
"type": "heatmap",
"x": x_coords,
"y": y_coords,
"z": z_values,
"colorscale": colorscale,
"colorbar": {"title": "Amplitude"},
"hovertemplate": (
"Sweep: %{x}<br>"
"Depth: %{y:.3f} m<br>"
"Amplitude: %{z:.3f}<br>"
"<extra></extra>"
),
**heatmap_kwargs,
}
freq_start, freq_stop = processed_data.get("frequency_range", [0.0, 0.0])
config_info = (
f"Freq: {freq_start/1e6:.1f}-{freq_stop/1e6:.1f} MHz | "
f"Gain: {self._config['gain']:.1f} | "
f"Cut: {self._config['cut']:.3f} m | "
f"Max: {self._config['max']:.1f} m | "
f"Axis: {self._config['axis']} | "
f"Sweeps: {len(history)}"
)
if processed_data.get("reference_used", False):
config_info += " | Open Air: ON"
if self._config["data_limitation"]:
config_info += f" | Limit: {self._config['data_limitation']}"
layout = {
"title": f"B-Scan Heatmap - {config_info}",
"xaxis": {"title": "Sweep Number", "side": "top"},
"yaxis": {"title": "Depth (m)", "autorange": "reversed"},
"hovermode": "closest",
"height": 600,
"width": 500,
"template": "plotly_dark",
}
return {"data": [heatmap_trace], "layout": layout}
# -------------------------------------------------------------------------
# Low-level helpers
# -------------------------------------------------------------------------
def _get_complex_s11(self, sweep_data: SweepData) -> NDArray[np.complex128] | None:
"""Extract complex S11 array from sweep points."""
try:
if not sweep_data.points:
return None
# Each point is expected as a pair (real, imag)
arr = np.asarray(sweep_data.points, dtype=float)
if arr.ndim != 2 or arr.shape[1] != 2:
raise ValueError("Expected Nx2 array for (real, imag) points")
complex_data = arr[:, 0] + 1j * arr[:, 1]
return complex_data.astype(np.complex128, copy=False)
except Exception as exc: # noqa: BLE001
logger.error("Failed to extract complex S11 data", error=repr(exc))
return None
def _subtract_reference(
self,
signal: NDArray[np.complex128],
reference: NDArray[np.complex128],
) -> NDArray[np.complex128]:
"""Subtract reference from signal (complex subtraction), length-safe."""
try:
n = min(signal.size, reference.size)
if n == 0:
return signal
result = signal[:n] - reference[:n]
logger.debug("Reference subtraction completed", points=n)
return result
except Exception as exc: # noqa: BLE001
logger.error("Reference subtraction failed", error=repr(exc))
return signal # Non-fatal; continue with original signal
def _update_frequency_ranges(self, vna_config: dict[str, Any]) -> None:
"""Clamp configured frequency sliders to VNA limits."""
if not vna_config:
return
start_freq_hz = float(vna_config.get("start_freq", 100e6))
stop_freq_hz = float(vna_config.get("stop_freq", 8.8e9))
start_mhz = start_freq_hz / 1e6
stop_mhz = stop_freq_hz / 1e6
current_start = float(self._config["start_freq"])
current_stop = float(self._config["stop_freq"])
# Ensure order and clamp to VNA bounds
new_start = max(start_mhz, min(stop_mhz, current_start))
new_stop = max(new_start, min(stop_mhz, current_stop))
self._config["start_freq"] = new_start
self._config["stop_freq"] = new_stop
def _perform_data_analysis(
self,
complex_data: NDArray[np.complex128],
vna_config: dict[str, Any],
) -> dict[str, Any] | None:
"""Full analysis pipeline: limit -> IFFT -> depth shaping."""
try:
# Determine effective frequency range (Hz)
if vna_config:
freq_start = max(float(vna_config.get("start_freq", 100e6)), self._config["start_freq"] * 1e6)
freq_stop = min(float(vna_config.get("stop_freq", 8.8e9)), self._config["stop_freq"] * 1e6)
else:
freq_start = self._config["start_freq"] * 1e6
freq_stop = self._config["stop_freq"] * 1e6
# Frequency vector over current data length
freq_axis = np.linspace(freq_start, freq_stop, complex_data.size, dtype=float)
# Optionally normalize amplitude (phase-only modes)
limited = self._apply_data_limitations(complex_data)
# IFFT to time domain
depth_m, time_response = self._perform_ifft(limited, freq_axis, axis=self._config["axis"])
# Depth windowing and gain shaping
depth_out, time_out = self._apply_depth_processing(depth_m, time_response)
return {
"time_data": time_out,
"distance": depth_out,
"freq_range": [freq_start, freq_stop],
"complex_time": limited,
}
except Exception as exc: # noqa: BLE001
logger.error("Data analysis failed", error=repr(exc))
return None
def _apply_data_limitations(self, s: NDArray[np.complex128]) -> NDArray[np.complex128]:
"""
Apply optional amplitude normalization to emphasize phase information.
Modes
-----
- None: passthrough
- "ph_only_1": normalize by magnitude
- "ph_only_2": same normalization (kept for behavioral parity)
"""
try:
mode = self._config.get("data_limitation")
if mode in {"ph_only_1", "ph_only_2"}:
# Avoid division by zero
return s / (np.abs(s) + 1e-12)
return s
except Exception as exc: # noqa: BLE001
logger.error("Data limitation processing failed", error=repr(exc))
return s
def _perform_ifft(
self,
s_array: NDArray[np.complex128],
frequencies_hz: NDArray[np.floating],
*,
axis: str = "abs",
) -> tuple[NDArray[np.floating], NDArray[np.floating]]:
"""
Frequency-to-time conversion with zero-padding and frequency offset handling.
Returns
-------
depth_m : ndarray
One-way distance in meters corresponding to the time axis.
y_fin : ndarray
Time-domain response (abs/real/phase per `axis`).
"""
try:
# Apply frequency band mask from current config (Hz)
start_hz = float(self._config.get("start_freq", 100.0)) * 1e6
stop_hz = float(self._config.get("stop_freq", 8800.0)) * 1e6
mask = (frequencies_hz >= start_hz) & (frequencies_hz <= stop_hz)
f = frequencies_hz[mask]
s = s_array[mask]
n = f.size
if n < 2:
raise ValueError("Not enough frequency points after filtering")
df = (f[-1] - f[0]) / (n - 1)
if df <= 0.0:
raise ValueError("Non-increasing frequency grid")
# Frequency offset index
k0 = int(np.round(f[0] / df))
# FFT size (next power of two for 2*(k0 + n - 1))
min_len = 2 * (k0 + n - 1)
N_fft = 1 << int(np.ceil(np.log2(min_len)))
dt = 1.0 / (N_fft * df)
t_sec = np.arange(N_fft, dtype=float) * dt
# Build shifted spectrum
H = np.zeros(N_fft, dtype=np.complex128)
H[k0 : k0 + n] = s
y = np.fft.ifft(H)
# Convert time (s) to one-way distance (m): d = c * t
depth_m = t_sec * self.SPEED_OF_LIGHT_M_S
if axis == "abs":
y_fin = np.abs(y)
elif axis == "real":
y_fin = np.real(y)
elif axis == "phase":
y_fin = np.angle(y)
else:
raise ValueError(f"Invalid axis parameter: {axis!r}")
# Truncate output to computed length (optional: keep full)
return depth_m[: y_fin.size], y_fin.astype(float, copy=False)
except Exception as exc: # noqa: BLE001
logger.error("IFFT failed", error=repr(exc))
# Graceful fallback: synthetic axes with magnitude of input
depth_fallback = np.linspace(0.0, 1.0, s_array.size, dtype=float)
return depth_fallback, np.abs(s_array).astype(float, copy=False)
def _apply_depth_processing(
self,
depth_m: NDArray[np.floating],
response: NDArray[np.floating],
) -> tuple[NDArray[np.floating], NDArray[np.floating]]:
"""
Apply depth windowing and gain shaping.
Window
------
Keep samples where:
depth in [2*cut, 2*cut + 2*max]
Then map depth to:
depth' = (depth - 2*cut) / 2
Gain
----
response' = response * (depth' ** gain)
"""
try:
cut = float(self._config["cut"])
max_depth = float(self._config["max"])
gain = float(self._config["gain"])
# Window: account for two-way path in raw depth (factor 2)
lo = 2.0 * cut
hi = lo + 2.0 * max_depth
mask = (depth_m >= lo) & (depth_m <= hi)
if not np.any(mask):
# Nothing in window; return empty arrays
return depth_m[:0], response[:0]
depth_win = depth_m[mask]
resp_win = response[mask]
# Convert to one-way depth relative to the cut
depth_out = (depth_win - lo) / 2.0
# Depth-dependent gain (safe for zero depth with exponent >= 0)
with np.errstate(invalid="ignore"):
gain_shape = np.power(depth_out, gain, dtype=float)
# Replace NaN/inf (e.g., 0**negative) with 0
gain_shape[~np.isfinite(gain_shape)] = 0.0
response_out = resp_win * gain_shape
return depth_out, response_out
except Exception as exc: # noqa: BLE001
logger.error("Depth processing failed", error=repr(exc))
return depth_m, response

9
vna_system/core/processors/manager.py
View File

@ -99,7 +99,7 @@ class ProcessorManager:
# --------------------------------------------------------------------- #
# Main processing actions
# --------------------------------------------------------------------- #
def process_sweep(self, sweep_data: SweepData, calibrated_data: Any, vna_config: ConfigPreset | None) -> dict[str, ProcessedResult]:
def process_sweep(self, sweep_data: SweepData, calibrated_data: Any, vna_config: ConfigPreset | None, reference_data: Any = None) -> dict[str, ProcessedResult]:
"""
Feed a sweep into all processors and dispatch results to callbacks.
@ -113,7 +113,7 @@ class ProcessorManager:
for processor_id, processor in processors_items:
try:
result = processor.add_sweep_data(sweep_data, calibrated_data, vna_config)
result = processor.add_sweep_data(sweep_data, calibrated_data, vna_config, reference_data)
if result:
results[processor_id] = result
for cb in callbacks:
@ -215,7 +215,8 @@ class ProcessorManager:
if latest and latest.sweep_number > self._last_processed_sweep:
calibrated = self._apply_calibration(latest)
vna_cfg = self.settings_manager.get_current_preset()
self.process_sweep(latest, calibrated, vna_cfg)
reference_data = self.settings_manager.get_current_reference_sweep(vna_cfg)
self.process_sweep(latest, calibrated, vna_cfg, reference_data)
self._last_processed_sweep = latest.sweep_number
# Light-duty polling to reduce wakeups
@ -267,9 +268,11 @@ class ProcessorManager:
"""
try:
from .implementations import MagnitudeProcessor, PhaseProcessor, SmithChartProcessor
from .implementations.bscan_processor import BScanProcessor
self.register_processor(MagnitudeProcessor(self.config_dir))
self.register_processor(PhaseProcessor(self.config_dir))
self.register_processor(BScanProcessor(self.config_dir))
# self.register_processor(SmithChartProcessor(self.config_dir))
logger.info("Default processors registered", count=len(self._processors))

464
vna_system/core/settings/reference_manager.py Normal file
View File

@ -0,0 +1,464 @@
"""
Open Air Reference Manager
Manages open air reference sweep data for background subtraction and normalization.
References are stored per preset configuration and can be captured from live sweeps.
"""
import json
import shutil
from datetime import datetime
from pathlib import Path
from vna_system.core import config as cfg
from vna_system.core.logging.logger import get_component_logger
from vna_system.core.acquisition.sweep_buffer import SweepData
from vna_system.core.settings.preset_manager import ConfigPreset
logger = get_component_logger(__file__)
class ReferenceInfo:
"""Information about a stored reference."""
def __init__(
self,
name: str,
timestamp: datetime,
preset_filename: str,
description: str = "",
metadata: dict | None = None
):
self.name = name
self.timestamp = timestamp
self.preset_filename = preset_filename
self.description = description
self.metadata = metadata or {}
def to_dict(self) -> dict:
"""Convert to dictionary for JSON serialization."""
return {
"name": self.name,
"timestamp": self.timestamp.isoformat(),
"preset_filename": self.preset_filename,
"description": self.description,
"metadata": self.metadata
}
@classmethod
def from_dict(cls, data: dict) -> "ReferenceInfo":
"""Create from dictionary loaded from JSON."""
return cls(
name=data["name"],
timestamp=datetime.fromisoformat(data["timestamp"]),
preset_filename=data["preset_filename"],
description=data.get("description", ""),
metadata=data.get("metadata", {})
)
class ReferenceManager:
"""
Manages open air reference sweep data.
References are stored in the following structure:
references/
├── current_reference -> symlink to active reference
├── <preset_name>/
│ ├── <reference_name>/
│ │ ├── info.json # Reference metadata
│ │ └── sweep_data.json # Raw sweep data
│ └── ...
└── ...
"""
def __init__(self):
self.references_dir = Path(cfg.BASE_DIR) / "references"
self.current_reference_symlink = self.references_dir / "current_reference"
# Ensure directory structure exists
self.references_dir.mkdir(exist_ok=True)
logger.info("ReferenceManager initialized", references_dir=str(self.references_dir))
# ------------------------------------------------------------------ #
# Reference Creation and Storage
# ------------------------------------------------------------------ #
def create_reference(
self,
name: str,
preset: ConfigPreset,
sweep_data: SweepData,
description: str = "",
metadata: dict | None = None
) -> ReferenceInfo:
"""
Create a new reference from sweep data.
Args:
name: Unique name for the reference
preset: Current configuration preset
sweep_data: Raw sweep data to store
description: Optional description
metadata: Optional additional metadata
Returns:
ReferenceInfo: Created reference information
Raises:
ValueError: If reference name already exists for this preset
OSError: If file operations fail
"""
if not name or not name.strip():
raise ValueError("Reference name cannot be empty")
preset_dir = self._get_preset_dir(preset)
reference_dir = preset_dir / name
if reference_dir.exists():
raise ValueError(f"Reference '{name}' already exists for preset {preset.filename}")
# Create reference directory
reference_dir.mkdir(parents=True, exist_ok=True)
# Create reference info
reference_info = ReferenceInfo(
name=name,
timestamp=datetime.now(),
preset_filename=preset.filename,
description=description.strip(),
metadata=metadata or {}
)
try:
# Save reference metadata
info_file = reference_dir / "info.json"
with info_file.open("w", encoding="utf-8") as f:
json.dump(reference_info.to_dict(), f, indent=2, ensure_ascii=False)
# Save sweep data
sweep_file = reference_dir / "sweep_data.json"
sweep_dict = {
"sweep_number": sweep_data.sweep_number,
"timestamp": sweep_data.timestamp,
"points": sweep_data.points,
"total_points": sweep_data.total_points
}
with sweep_file.open("w", encoding="utf-8") as f:
json.dump(sweep_dict, f, indent=2)
logger.info(
"Created reference",
name=name,
preset=preset.filename,
sweep_number=sweep_data.sweep_number,
points=len(sweep_data.points)
)
return reference_info
except Exception as exc:
# Clean up on failure
if reference_dir.exists():
shutil.rmtree(reference_dir, ignore_errors=True)
logger.error("Failed to create reference", name=name, error=repr(exc))
raise
def delete_reference(self, name: str, preset: ConfigPreset) -> bool:
"""
Delete a reference.
Args:
name: Reference name to delete
preset: Configuration preset
Returns:
bool: True if deleted successfully
"""
preset_dir = self._get_preset_dir(preset)
reference_dir = preset_dir / name
if not reference_dir.exists():
logger.warning("Reference not found for deletion", name=name, preset=preset.filename)
return False
try:
# Check if this is the current reference
current_ref = self.get_current_reference(preset)
if current_ref and current_ref.name == name:
self.clear_current_reference()
shutil.rmtree(reference_dir)
logger.info("Deleted reference", name=name, preset=preset.filename)
return True
except Exception as exc:
logger.error("Failed to delete reference", name=name, error=repr(exc))
return False
# ------------------------------------------------------------------ #
# Reference Selection and Retrieval
# ------------------------------------------------------------------ #
def set_current_reference(self, name: str, preset: ConfigPreset) -> bool:
"""
Set the current active reference.
Args:
name: Reference name to activate
preset: Configuration preset
Returns:
bool: True if set successfully
"""
preset_dir = self._get_preset_dir(preset)
reference_dir = preset_dir / name
if not reference_dir.exists():
raise ValueError(f"Reference '{name}' not found for preset {preset.filename}")
try:
# Remove existing symlink
if self.current_reference_symlink.exists() or self.current_reference_symlink.is_symlink():
self.current_reference_symlink.unlink()
# Create relative symlink for portability
try:
relative_target = reference_dir.relative_to(self.references_dir)
except ValueError:
relative_target = reference_dir
self.current_reference_symlink.symlink_to(relative_target)
logger.info("Set current reference", name=name, preset=preset.filename)
return True
except Exception as exc:
logger.error("Failed to set current reference", name=name, error=repr(exc))
return False
def get_current_reference(self, preset: ConfigPreset) -> ReferenceInfo | None:
"""
Get information about the currently active reference.
Args:
preset: Configuration preset to check against
Returns:
ReferenceInfo | None: Current reference info, or None if no reference is set
or if the current reference doesn't match the preset
"""
if not self.current_reference_symlink.exists():
return None
try:
target_dir = self.current_reference_symlink.resolve()
if not target_dir.exists():
logger.warning("Current reference symlink points to non-existent directory")
return None
# Load reference info
info_file = target_dir / "info.json"
if not info_file.exists():
logger.warning("Current reference missing info.json")
return None
with info_file.open("r", encoding="utf-8") as f:
info_data = json.load(f)
reference_info = ReferenceInfo.from_dict(info_data)
# Check if reference matches current preset
if reference_info.preset_filename != preset.filename:
logger.debug(
"Current reference is for different preset",
reference_preset=reference_info.preset_filename,
current_preset=preset.filename
)
return None
return reference_info
except Exception as exc:
logger.warning("Failed to get current reference", error=repr(exc))
return None
def clear_current_reference(self) -> bool:
"""
Clear the current reference selection.
Returns:
bool: True if cleared successfully
"""
try:
if self.current_reference_symlink.exists() or self.current_reference_symlink.is_symlink():
self.current_reference_symlink.unlink()
logger.info("Cleared current reference")
return True
except Exception as exc:
logger.error("Failed to clear current reference", error=repr(exc))
return False
def get_current_reference_sweep(self, preset: ConfigPreset) -> SweepData | None:
"""
Get the sweep data for the currently active reference.
Args:
preset: Configuration preset
Returns:
SweepData | None: Current reference sweep data, or None if no reference is active
"""
current_ref = self.get_current_reference(preset)
if not current_ref:
return None
return self.get_reference_sweep(current_ref.name, preset)
def get_reference_sweep(self, name: str, preset: ConfigPreset) -> SweepData | None:
"""
Get sweep data for a specific reference.
Args:
name: Reference name
preset: Configuration preset
Returns:
SweepData | None: Reference sweep data, or None if not found
"""
preset_dir = self._get_preset_dir(preset)
reference_dir = preset_dir / name
sweep_file = reference_dir / "sweep_data.json"
if not sweep_file.exists():
logger.warning("Reference sweep data not found", name=name, preset=preset.filename)
return None
try:
with sweep_file.open("r", encoding="utf-8") as f:
sweep_dict = json.load(f)
return SweepData(
sweep_number=sweep_dict["sweep_number"],
timestamp=sweep_dict["timestamp"],
points=sweep_dict["points"],
total_points=sweep_dict["total_points"]
)
except Exception as exc:
logger.error("Failed to load reference sweep data", name=name, error=repr(exc))
return None
# ------------------------------------------------------------------ #
# Reference Listing and Information
# ------------------------------------------------------------------ #
def list_references(self, preset: ConfigPreset) -> list[ReferenceInfo]:
"""
List all references for a given preset.
Args:
preset: Configuration preset
Returns:
list[ReferenceInfo]: List of available references, sorted by timestamp (newest first)
"""
preset_dir = self._get_preset_dir(preset)
if not preset_dir.exists():
return []
references = []
for ref_dir in preset_dir.iterdir():
if not ref_dir.is_dir():
continue
info_file = ref_dir / "info.json"
if not info_file.exists():
logger.warning("Reference missing info.json", reference_dir=str(ref_dir))
continue
try:
with info_file.open("r", encoding="utf-8") as f:
info_data = json.load(f)
reference_info = ReferenceInfo.from_dict(info_data)
references.append(reference_info)
except Exception as exc:
logger.warning("Failed to load reference info", reference_dir=str(ref_dir), error=repr(exc))
# Sort by timestamp, newest first
references.sort(key=lambda r: r.timestamp, reverse=True)
return references
def get_reference_info(self, name: str, preset: ConfigPreset) -> ReferenceInfo | None:
"""
Get information about a specific reference.
Args:
name: Reference name
preset: Configuration preset
Returns:
ReferenceInfo | None: Reference information, or None if not found
"""
preset_dir = self._get_preset_dir(preset)
reference_dir = preset_dir / name
info_file = reference_dir / "info.json"
if not info_file.exists():
return None
try:
with info_file.open("r", encoding="utf-8") as f:
info_data = json.load(f)
return ReferenceInfo.from_dict(info_data)
except Exception as exc:
logger.warning("Failed to load reference info", name=name, error=repr(exc))
return None
# ------------------------------------------------------------------ #
# Helper Methods
# ------------------------------------------------------------------ #
def _get_preset_dir(self, preset: ConfigPreset) -> Path:
"""Get directory path for a preset's references."""
# Use preset filename without extension as directory name
preset_name = Path(preset.filename).stem
return self.references_dir / preset_name
def cleanup_orphaned_references(self) -> int:
"""
Clean up reference directories that don't have valid preset files.
Returns:
int: Number of orphaned references cleaned up
"""
cleaned_count = 0
for preset_dir in self.references_dir.iterdir():
if not preset_dir.is_dir() or preset_dir.name == "current_reference":
continue
# Check if corresponding preset file exists
preset_filename = f"{preset_dir.name}.bin"
preset_path = Path(cfg.BASE_DIR) / "binary_input" / "config_inputs" / preset_filename
if not preset_path.exists():
try:
shutil.rmtree(preset_dir)
cleaned_count += 1
logger.info("Cleaned up orphaned reference directory", preset=preset_dir.name)
except Exception as exc:
logger.warning("Failed to clean up orphaned references", preset=preset_dir.name, error=repr(exc))
if cleaned_count > 0:
logger.info("Reference cleanup completed", cleaned_count=cleaned_count)
return cleaned_count

160
vna_system/core/settings/settings_manager.py
View File

@ -11,6 +11,7 @@ from vna_system.core.settings.calibration_manager import (
CalibrationSet,
CalibrationStandard,
)
from vna_system.core.settings.reference_manager import ReferenceManager, ReferenceInfo
logger = get_component_logger(__file__)
@ -38,6 +39,7 @@ class VNASettingsManager:
# Sub-managers
self.preset_manager = PresetManager(self.base_dir / "binary_input")
self.calibration_manager = CalibrationManager(self.base_dir)
self.reference_manager = ReferenceManager()
logger.debug(
"VNASettingsManager initialized",
@ -50,11 +52,7 @@ class VNASettingsManager:
def get_available_presets(self) -> list[ConfigPreset]:
"""Return all available configuration presets."""
return self.preset_manager.get_available_presets()
def get_presets_by_mode(self, mode: VNAMode) -> list[ConfigPreset]:
"""Return presets filtered by VNA mode."""
return [p for p in self.get_available_presets() if p.mode == mode]
def set_current_preset(self, preset: ConfigPreset) -> ConfigPreset:
"""Set the current configuration preset (updates the symlink)."""
chosen = self.preset_manager.set_current_preset(preset)
@ -148,6 +146,48 @@ class VNASettingsManager:
return [CalibrationStandard.THROUGH]
return []
# ------------------------------------------------------------------ #
# Reference Management
# ------------------------------------------------------------------ #
def get_available_references(self, preset: ConfigPreset | None = None) -> list[ReferenceInfo]:
"""Return all available references for the given preset (or current preset)."""
current_preset = preset or self.get_current_preset()
if not current_preset:
return []
return self.reference_manager.list_references(current_preset)
def get_current_reference(self, preset: ConfigPreset | None = None) -> ReferenceInfo | None:
"""Return the currently selected reference for the given preset (or current preset)."""
current_preset = preset or self.get_current_preset()
if not current_preset:
return None
return self.reference_manager.get_current_reference(current_preset)
def set_current_reference(self, reference_name: str, preset: ConfigPreset | None = None) -> bool:
"""Set the current reference for the given preset (or current preset)."""
current_preset = preset or self.get_current_preset()
if not current_preset:
raise ValueError("No current preset available")
return self.reference_manager.set_current_reference(reference_name, current_preset)
def clear_current_reference(self) -> bool:
"""Clear the current reference selection."""
return self.reference_manager.clear_current_reference()
def get_current_reference_sweep(self, preset: ConfigPreset | None = None) -> SweepData | None:
"""Get sweep data for the currently selected reference."""
current_preset = preset or self.get_current_preset()
if not current_preset:
return None
return self.reference_manager.get_current_reference_sweep(current_preset)
def delete_reference(self, reference_name: str, preset: ConfigPreset | None = None) -> bool:
"""Delete a reference for the given preset (or current preset)."""
current_preset = preset or self.get_current_preset()
if not current_preset:
raise ValueError("No current preset available")
return self.reference_manager.delete_reference(reference_name, current_preset)
# ------------------------------------------------------------------ #
# Acquisition integration
# ------------------------------------------------------------------ #
@ -169,7 +209,18 @@ class VNASettingsManager:
RuntimeError
If no sweep is available in the acquisition buffer.
"""
latest = data_acquisition.sweep_buffer.get_latest_sweep()
# Get current sweep number before any operations
current_sweep = data_acquisition.sweep_buffer.get_latest_sweep()
current_sweep_number = current_sweep.sweep_number if current_sweep else -1
# Check if acquisition is running, if not - trigger single sweep
if not data_acquisition.is_running:
logger.info("Acquisition not running, triggering single sweep for calibration")
data_acquisition.trigger_single_sweep()
# Wait for new sweep to appear (regardless of acquisition state)
latest = self._wait_for_new_sweep(data_acquisition.sweep_buffer, current_sweep_number)
if latest is None:
raise RuntimeError("No sweep data available in acquisition buffer")
@ -181,6 +232,66 @@ class VNASettingsManager:
)
return latest.sweep_number
def capture_reference_from_acquisition(
self,
reference_name: str,
data_acquisition: VNADataAcquisition,
description: str = "",
metadata: dict | None = None,
preset: ConfigPreset | None = None,
) -> ReferenceInfo:
"""
Capture the latest sweep from acquisition as an open air reference.
Args:
reference_name: Name for the new reference
data_acquisition: Data acquisition instance to capture from
description: Optional description for the reference
metadata: Optional additional metadata
preset: Preset to use (defaults to current preset)
Returns:
ReferenceInfo: Created reference information
Raises:
RuntimeError: If no sweep is available in the acquisition buffer
ValueError: If no current preset is available
"""
# Get current sweep number before any operations
current_sweep = data_acquisition.sweep_buffer.get_latest_sweep()
current_sweep_number = current_sweep.sweep_number if current_sweep else -1
# Check if acquisition is running, if not - trigger single sweep
if not data_acquisition.is_running:
logger.info("Acquisition not running, triggering single sweep for reference")
data_acquisition.trigger_single_sweep()
# Wait for new sweep to appear (regardless of acquisition state)
latest = self._wait_for_new_sweep(data_acquisition.sweep_buffer, current_sweep_number)
if latest is None:
raise RuntimeError("No sweep data available in acquisition buffer")
current_preset = preset or self.get_current_preset()
if not current_preset:
raise ValueError("No current preset available")
reference_info = self.reference_manager.create_reference(
name=reference_name,
preset=current_preset,
sweep_data=latest,
description=description,
metadata=metadata or {}
)
logger.info(
"Captured open air reference from acquisition",
reference_name=reference_name,
preset=current_preset.filename,
sweep_number=latest.sweep_number,
)
return reference_info
def get_status_summary(self) -> dict[str, object]:
"""Get comprehensive status of current configuration and calibration"""
current_preset = self.get_current_preset()
@ -223,4 +334,39 @@ class VNASettingsManager:
"missing_standards": [s.value for s in working_calibration.get_missing_standards()]
}
return summary
return summary
def _wait_for_new_sweep(self, sweep_buffer, current_sweep_number: int, timeout_seconds: float = 5.0) -> SweepData | None:
"""
Wait for a new sweep to appear in the buffer with a higher sweep number.
Args:
sweep_buffer: The sweep buffer to monitor
current_sweep_number: Current sweep number to wait beyond
timeout_seconds: Maximum time to wait
Returns:
SweepData | None: New sweep data, or None if timeout occurred
"""
import time
start_time = time.time()
while time.time() - start_time < timeout_seconds:
latest = sweep_buffer.get_latest_sweep()
if latest and latest.sweep_number > current_sweep_number:
logger.debug(
"New sweep detected",
old_sweep=current_sweep_number,
new_sweep=latest.sweep_number
)
return latest
# Short sleep to avoid busy waiting
time.sleep(0.01)
logger.warning(
"Timeout waiting for new sweep",
current_sweep=current_sweep_number,
timeout=timeout_seconds
)
return None

1
vna_system/references/current_reference Symbolic link
View File

@ -0,0 +1 @@
s11_start100_stop8800_points1000_bw1khz/test1

134
vna_system/web_ui/static/css/settings.css
View File

@ -489,4 +489,138 @@
.calibration-standard-btn.btn--warning:hover:not(:disabled) {
background-color: var(--color-warning-600);
border-color: var(--color-warning-600);
}
/* ========================================
Reference Management Styles
======================================== */
/* Reference Creation */
.reference-creation {
display: flex;
flex-direction: column;
gap: var(--space-3);
margin-bottom: var(--space-4);
}
.reference-name-input,
.reference-description-input {
padding: var(--space-2) var(--space-3);
border: 1px solid var(--color-border);
border-radius: var(--radius-md);
font-size: var(--text-sm);
background: var(--color-bg);
color: var(--color-text);
transition: all 0.2s ease;
}
.reference-name-input:focus,
.reference-description-input:focus {
outline: none;
border-color: var(--color-primary-500);
box-shadow: 0 0 0 3px var(--color-primary-100);
}
.reference-name-input::placeholder,
.reference-description-input::placeholder {
color: var(--color-text-secondary);
}
/* Reference Actions */
.reference-actions {
display: flex;
gap: var(--space-3);
align-items: stretch;
margin-top: var(--space-3);
flex-wrap: wrap;
}
.reference-dropdown {
flex: 1;
min-width: 200px;
padding: var(--space-2) var(--space-3);
border: 1px solid var(--color-border);
border-radius: var(--radius-md);
font-size: var(--text-sm);
background: var(--color-bg);
color: var(--color-text);
}
.reference-dropdown:disabled {
opacity: 0.6;
cursor: not-allowed;
background: var(--color-gray-50);
}
/* Current Reference Info */
.current-reference-info {
margin-top: var(--space-4);
padding-top: var(--space-4);
border-top: 1px solid var(--color-border);
}
.reference-info-card {
background: var(--color-gray-800);
border: 1px solid var(--color-gray-600);
border-radius: var(--radius-lg);
padding: var(--space-4);
}
.reference-info-card h5 {
margin: 0 0 var(--space-3) 0;
font-size: var(--text-sm);
font-weight: var(--weight-semibold);
color: var(--color-gray-300);
text-transform: uppercase;
letter-spacing: 0.05em;
}
.reference-details {
display: flex;
flex-direction: column;
gap: var(--space-2);
background: var(--color-gray-900);
padding: var(--space-3);
border-radius: var(--radius-md);
border: 1px solid var(--color-gray-700);
}
.reference-name {
font-size: var(--text-base);
font-weight: var(--weight-semibold);
color: var(--color-white);
}
.reference-timestamp {
font-size: var(--text-sm);
color: var(--color-gray-300);
font-family: var(--font-mono);
}
.reference-description {
margin: 0;
font-size: var(--text-sm);
color: var(--color-gray-400);
font-style: italic;
}
.reference-description:empty::before {
content: "No description provided";
opacity: 0.7;
}
/* Responsive adjustments */
@media (max-width: 768px) {
.reference-creation {
gap: var(--space-2);
}
.reference-actions {
flex-direction: column;
gap: var(--space-2);
}
.reference-dropdown {
min-width: unset;
}
}

399
vna_system/web_ui/static/js/modules/settings.js
View File

@ -90,12 +90,11 @@ export class SettingsManager {
this.currentPreset = null;
this.currentCalibration = null;
this.workingCalibration = null;
this.availableReferences = [];
this.currentReference = null;
// Калибровка: ожидание свипа
this.waitingForSweep = false;
this.pendingStandard = null;
// Калибровка: состояние захвата
this.disabledStandards = new Set();
this.calibrationTimeout = null;
// DOM cache
this.elements = {};
@ -105,7 +104,6 @@ export class SettingsManager {
this.reqGuard = new RequestGuard();
// Единственный bound-обработчик, чтобы корректно отписываться
this._boundHandleSweepForCalibration = this.handleSweepForCalibration.bind(this);
// Bind UI handlers
this.handlePresetChange = this.handlePresetChange.bind(this);
@ -118,6 +116,13 @@ export class SettingsManager {
this.handleViewPlots = this.handleViewPlots.bind(this);
this.handleViewCurrentPlots = this.handleViewCurrentPlots.bind(this);
// Reference handlers
this.handleCreateReference = this.handleCreateReference.bind(this);
this.handleReferenceChange = this.handleReferenceChange.bind(this);
this.handleSetReference = this.handleSetReference.bind(this);
this.handleClearReference = this.handleClearReference.bind(this);
this.handleDeleteReference = this.handleDeleteReference.bind(this);
// Пакет данных для модалки с графиками
this.currentPlotsData = null;
}
@ -179,6 +184,19 @@ export class SettingsManager {
plotsGrid: document.getElementById('plotsGrid'),
downloadAllBtn: document.getElementById('downloadAllBtn'),
// References
referenceNameInput: document.getElementById('referenceNameInput'),
referenceDescriptionInput: document.getElementById('referenceDescriptionInput'),
createReferenceBtn: document.getElementById('createReferenceBtn'),
referenceDropdown: document.getElementById('referenceDropdown'),
setReferenceBtn: document.getElementById('setReferenceBtn'),
clearReferenceBtn: document.getElementById('clearReferenceBtn'),
deleteReferenceBtn: document.getElementById('deleteReferenceBtn'),
currentReferenceInfo: document.getElementById('currentReferenceInfo'),
currentReferenceName: document.getElementById('currentReferenceName'),
currentReferenceTimestamp: document.getElementById('currentReferenceTimestamp'),
currentReferenceDescription: document.getElementById('currentReferenceDescription'),
// Status
presetCount: document.getElementById('presetCount'),
calibrationCount: document.getElementById('calibrationCount'),
@ -199,6 +217,13 @@ export class SettingsManager {
this.elements.viewPlotsBtn?.addEventListener('click', this.handleViewPlots);
this.elements.viewCurrentPlotsBtn?.addEventListener('click', this.handleViewCurrentPlots);
// References
this.elements.createReferenceBtn?.addEventListener('click', this.handleCreateReference);
this.elements.referenceDropdown?.addEventListener('change', this.handleReferenceChange);
this.elements.setReferenceBtn?.addEventListener('click', this.handleSetReference);
this.elements.clearReferenceBtn?.addEventListener('click', this.handleClearReference);
this.elements.deleteReferenceBtn?.addEventListener('click', this.handleDeleteReference);
// Name input → enables Save
this.elements.calibrationNameInput?.addEventListener('input', () => {
const hasName = this.elements.calibrationNameInput.value.trim().length > 0;
@ -217,6 +242,13 @@ export class SettingsManager {
this.elements.viewPlotsBtn?.removeEventListener('click', this.handleViewPlots);
this.elements.viewCurrentPlotsBtn?.removeEventListener('click', this.handleViewCurrentPlots);
// References
this.elements.createReferenceBtn?.removeEventListener('click', this.handleCreateReference);
this.elements.referenceDropdown?.removeEventListener('change', this.handleReferenceChange);
this.elements.setReferenceBtn?.removeEventListener('click', this.handleSetReference);
this.elements.clearReferenceBtn?.removeEventListener('click', this.handleClearReference);
this.elements.deleteReferenceBtn?.removeEventListener('click', this.handleDeleteReference);
// WebSocket
if (this.websocket) {
this.websocket.off?.('processor_result', this._boundHandleSweepForCalibration);
@ -229,7 +261,8 @@ export class SettingsManager {
await Promise.all([
this._loadPresets(),
this._loadStatus(),
this._loadWorkingCalibration()
this._loadWorkingCalibration(),
this._loadReferences()
]);
}
@ -464,7 +497,11 @@ export class SettingsManager {
}
if (this.elements.saveCalibrationBtn) this.elements.saveCalibrationBtn.disabled = true;
if (this.elements.progressText) this.elements.progressText.textContent = '0/0';
console.log('🔄 Calibration UI reset after preset change');
// Reset reference state
this.availableReferences = [];
this.currentReference = null;
console.log('🔄 Calibration and reference UI reset after preset change');
}
/* ----------------------------- Event Handlers (UI) ----------------------------- */
@ -554,39 +591,30 @@ export class SettingsManager {
this.disabledStandards.add(standard);
const btn = document.querySelector(`[data-standard="${standard}"]`);
ButtonState.set(btn, { state: 'loading', icon: 'clock', text: 'Waiting for next sweep...' });
ButtonState.set(btn, { state: 'loading', icon: 'upload', text: 'Capturing...' });
// Если acquisition не работает — попросим один свип
const running = this.acquisition?.isRunning?.() ?? false;
if (!running) {
this._notify('info', 'Triggering Sweep', `Requesting single sweep for ${standard.toUpperCase()} standard`);
try {
await this.acquisition?.triggerSingleSweep?.();
} catch (e) {
console.error('Trigger sweep failed:', e);
this._notify('error', 'Sweep Error', 'Failed to trigger single sweep for calibration');
this._resetCalibrationCaptureState(standard);
return;
}
} else {
// Если уже работает — просим пользователя запустить новый свип (или просто ждём)
this._notify('info', 'Waiting for Sweep', `Please trigger a new sweep to capture ${standard.toUpperCase()} standard`);
}
this._notify('info', 'Capturing Standard', `Capturing ${standard.toUpperCase()} standard...`);
// Ожидаем следующий свип через WebSocket; подписка единым обработчиком
this.waitingForSweep = true;
this.pendingStandard = standard;
this.websocket?.on?.('processor_result', this._boundHandleSweepForCalibration);
// Прямой вызов API - бэкенд сам обработает ожидание и триггеринг
const r = await fetch('/api/v1/settings/calibration/add-standard', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ standard })
});
// Таймаут ожидания (5с)
this._clearCalibrationTimeout();
this.calibrationTimeout = setTimeout(() => {
this._notify('warning', 'Calibration Timeout', `No sweep received within 5 seconds for ${standard.toUpperCase()}. Please try again.`);
this._resetCalibrationCaptureState(standard);
}, 5000);
if (!r.ok) throw new Error(`HTTP ${r.status}`);
const result = await r.json();
this._notify('success', 'Standard Captured', result.message);
// Сброс состояния
this._resetCalibrationCaptureState();
// Обновить рабочую калибровку
await this._loadWorkingCalibration();
} catch (e) {
console.error('Start standard capture failed:', e);
this._notify('error', 'Calibration Error', 'Failed to start calibration standard capture');
console.error('Capture standard failed:', e);
this._notify('error', 'Calibration Error', 'Failed to capture calibration standard');
this._resetCalibrationCaptureState(standard);
}
}), 500
@ -716,65 +744,16 @@ export class SettingsManager {
/* ----------------------------- WebSocket sweep capture ----------------------------- */
async handleSweepForCalibration() {
// реагируем только если действительно ждём свип
if (!this.waitingForSweep || !this.pendingStandard) return;
try {
console.log(`📡 New sweep → capture ${this.pendingStandard}...`);
// Сразу отключаем подписку и таймер
this.websocket?.off?.('processor_result', this._boundHandleSweepForCalibration);
this._clearCalibrationTimeout();
const btn = document.querySelector(`[data-standard="${this.pendingStandard}"]`);
ButtonState.set(btn, { state: 'loading', icon: 'upload', text: 'Capturing...' });
const r = await fetch('/api/v1/settings/calibration/add-standard', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ standard: this.pendingStandard })
});
if (!r.ok) throw new Error(`HTTP ${r.status}`);
const result = await r.json();
this._notify('success', 'Standard Captured', result.message);
// Сброс
this._resetCalibrationCaptureState();
// Обновить рабочую калибровку
await this._loadWorkingCalibration();
} catch (e) {
console.error('Capture standard failed:', e);
this._notify('error', 'Calibration Error', 'Failed to capture calibration standard');
this._resetCalibrationCaptureState();
}
}
_resetCalibrationCaptureState(standard = null) {
if (standard) this.disabledStandards.delete(standard);
else this.disabledStandards.clear();
if (!standard || standard === this.pendingStandard) {
this.waitingForSweep = false;
this.pendingStandard = null;
this.websocket?.off?.('processor_result', this._boundHandleSweepForCalibration);
this._clearCalibrationTimeout();
}
if (this.workingCalibration) {
this._renderStandardButtons(this.workingCalibration);
}
}
_clearCalibrationTimeout() {
if (this.calibrationTimeout) {
clearTimeout(this.calibrationTimeout);
this.calibrationTimeout = null;
}
}
/* ----------------------------- Plots Modal ----------------------------- */
@ -1230,26 +1209,242 @@ export class SettingsManager {
/* ----------------------------- Helpers ----------------------------- */
/* ----------------------------- Reference Management ----------------------------- */
async _loadReferences() {
try {
if (!this.currentPreset) {
this._renderReferencesDropdown([]);
this._updateCurrentReferenceInfo(null);
return;
}
// Load available references
const referencesResponse = await fetch(`/api/v1/settings/references?preset_filename=${encodeURIComponent(this.currentPreset.filename)}`);
if (!referencesResponse.ok) throw new Error(`HTTP ${referencesResponse.status}`);
this.availableReferences = await referencesResponse.json();
// Load current reference
const currentResponse = await fetch(`/api/v1/settings/reference/current?preset_filename=${encodeURIComponent(this.currentPreset.filename)}`);
if (!currentResponse.ok) throw new Error(`HTTP ${currentResponse.status}`);
this.currentReference = await currentResponse.json();
this._renderReferencesDropdown(this.availableReferences);
this._updateCurrentReferenceInfo(this.currentReference);
} catch (error) {
console.error('Failed to load references:', error);
this._renderReferencesDropdown([]);
this._updateCurrentReferenceInfo(null);
}
}
_renderReferencesDropdown(references) {
if (!this.elements.referenceDropdown) return;
// Clear existing options
this.elements.referenceDropdown.innerHTML = '';
if (references.length === 0) {
this.elements.referenceDropdown.innerHTML = '<option value="">No references available</option>';
this.elements.referenceDropdown.disabled = true;
this.elements.setReferenceBtn.disabled = true;
this.elements.clearReferenceBtn.disabled = true;
this.elements.deleteReferenceBtn.disabled = true;
} else {
this.elements.referenceDropdown.innerHTML = '<option value="">Select a reference...</option>';
references.forEach(ref => {
const option = document.createElement('option');
option.value = ref.name;
option.textContent = `${ref.name} (${new Date(ref.timestamp).toLocaleDateString()})`;
this.elements.referenceDropdown.appendChild(option);
});
this.elements.referenceDropdown.disabled = false;
// Select current reference if any
if (this.currentReference) {
this.elements.referenceDropdown.value = this.currentReference.name;
}
}
this._updateReferenceButtons();
}
_updateReferenceButtons() {
const hasSelection = this.elements.referenceDropdown.value !== '';
const hasCurrent = this.currentReference !== null;
this.elements.setReferenceBtn.disabled = !hasSelection;
this.elements.deleteReferenceBtn.disabled = !hasSelection;
this.elements.clearReferenceBtn.disabled = !hasCurrent;
}
_updateCurrentReferenceInfo(reference) {
if (!this.elements.currentReferenceInfo) return;
if (!reference) {
this.elements.currentReferenceInfo.style.display = 'none';
return;
}
this.elements.currentReferenceInfo.style.display = 'block';
if (this.elements.currentReferenceName) {
this.elements.currentReferenceName.textContent = reference.name;
}
if (this.elements.currentReferenceTimestamp) {
this.elements.currentReferenceTimestamp.textContent = new Date(reference.timestamp).toLocaleString();
}
if (this.elements.currentReferenceDescription) {
this.elements.currentReferenceDescription.textContent = reference.description || '';
}
}
async handleCreateReference() {
const name = this.elements.referenceNameInput.value.trim();
if (!name) {
this._notify('warning', 'Missing Name', 'Please enter a name for the reference');
return;
}
const description = this.elements.referenceDescriptionInput.value.trim();
this.debouncer.debounce('create-reference', () =>
this.reqGuard.runExclusive('create-reference', async () => {
try {
ButtonState.set(this.elements.createReferenceBtn, { state: 'loading', icon: 'loader', text: 'Creating...' });
const response = await fetch('/api/v1/settings/reference/create', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ name, description })
});
if (!response.ok) throw new Error(`HTTP ${response.status}`);
const result = await response.json();
this._notify('success', 'Reference Created', result.message);
// Clear inputs
this.elements.referenceNameInput.value = '';
this.elements.referenceDescriptionInput.value = '';
// Reload references
await this._loadReferences();
} catch (error) {
console.error('Create reference failed:', error);
this._notify('error', 'Reference Error', 'Failed to create reference');
} finally {
ButtonState.set(this.elements.createReferenceBtn, { state: 'normal', icon: 'target', text: 'Capture Reference' });
}
}), 500
);
}
handleReferenceChange() {
this._updateReferenceButtons();
}
async handleSetReference() {
const referenceName = this.elements.referenceDropdown.value;
if (!referenceName) return;
this.debouncer.debounce('set-reference', () =>
this.reqGuard.runExclusive('set-reference', async () => {
try {
ButtonState.set(this.elements.setReferenceBtn, { state: 'loading', icon: 'loader', text: 'Setting...' });
const response = await fetch('/api/v1/settings/reference/set', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ name: referenceName })
});
if (!response.ok) throw new Error(`HTTP ${response.status}`);
const result = await response.json();
this._notify('success', 'Reference Set', result.message);
// Reload references to update current
await this._loadReferences();
} catch (error) {
console.error('Set reference failed:', error);
this._notify('error', 'Reference Error', 'Failed to set reference');
} finally {
ButtonState.set(this.elements.setReferenceBtn, { state: 'normal', icon: 'check', text: 'Set Active' });
}
}), 400
);
}
async handleClearReference() {
this.debouncer.debounce('clear-reference', () =>
this.reqGuard.runExclusive('clear-reference', async () => {
try {
ButtonState.set(this.elements.clearReferenceBtn, { state: 'loading', icon: 'loader', text: 'Clearing...' });
const response = await fetch('/api/v1/settings/reference/current', {
method: 'DELETE'
});
if (!response.ok) throw new Error(`HTTP ${response.status}`);
const result = await response.json();
this._notify('success', 'Reference Cleared', result.message);
// Reload references to update current
await this._loadReferences();
} catch (error) {
console.error('Clear reference failed:', error);
this._notify('error', 'Reference Error', 'Failed to clear reference');
} finally {
ButtonState.set(this.elements.clearReferenceBtn, { state: 'normal', icon: 'x', text: 'Clear' });
}
}), 400
);
}
async handleDeleteReference() {
const referenceName = this.elements.referenceDropdown.value;
if (!referenceName) return;
// Confirm deletion
if (!confirm(`Are you sure you want to delete reference "${referenceName}"? This action cannot be undone.`)) {
return;
}
this.debouncer.debounce('delete-reference', () =>
this.reqGuard.runExclusive('delete-reference', async () => {
try {
ButtonState.set(this.elements.deleteReferenceBtn, { state: 'loading', icon: 'loader', text: 'Deleting...' });
const response = await fetch(`/api/v1/settings/reference/${encodeURIComponent(referenceName)}`, {
method: 'DELETE'
});
if (!response.ok) throw new Error(`HTTP ${response.status}`);
const result = await response.json();
this._notify('success', 'Reference Deleted', result.message);
// Reload references
await this._loadReferences();
} catch (error) {
console.error('Delete reference failed:', error);
this._notify('error', 'Reference Error', 'Failed to delete reference');
} finally {
ButtonState.set(this.elements.deleteReferenceBtn, { state: 'normal', icon: 'trash-2', text: 'Delete' });
}
}), 400
);
}
_notify(type, title, message) {
this.notifications?.show?.({ type, title, message });
}
_resetCalibrationCaptureState(standard = null) {
// публичная версия для внешних вызовов
this._resetCalibrationCaptureStateInternal(standard);
}
_resetCalibrationCaptureStateInternal(standard = null) {
if (standard) this.disabledStandards.delete(standard);
else this.disabledStandards.clear();
if (!standard || standard === this.pendingStandard) {
this.waitingForSweep = false;
this.pendingStandard = null;
this.websocket?.off?.('processor_result', this._boundHandleSweepForCalibration);
this._clearCalibrationTimeout();
}
if (this.workingCalibration) this._renderStandardButtons(this.workingCalibration);
}
}

56
vna_system/web_ui/templates/index.html
View File

@ -244,6 +244,62 @@
</div>
</div>
<!-- Open Air Reference -->
<div class="settings-card">
<h3 class="settings-card-title">Open Air Reference</h3>
<div class="settings-card-content">
<!-- Create Reference -->
<div class="workflow-section">
<h4 class="workflow-title">Create Reference</h4>
<div class="reference-creation">
<input type="text" class="reference-name-input" id="referenceNameInput" placeholder="Enter reference name">
<input type="text" class="reference-description-input" id="referenceDescriptionInput" placeholder="Description (optional)">
<button class="btn btn--primary" id="createReferenceBtn">
<i data-lucide="target"></i>
Capture Reference
</button>
</div>
</div>
<!-- Existing References -->
<div class="workflow-section">
<h4 class="workflow-title">Existing References</h4>
<div class="existing-references" id="existingReferences">
<select class="reference-dropdown" id="referenceDropdown" disabled>
<option value="">No references available</option>
</select>
<div class="reference-actions">
<button class="btn btn--primary" id="setReferenceBtn" disabled>
<i data-lucide="check"></i>
Set Active
</button>
<button class="btn btn--secondary" id="clearReferenceBtn" disabled>
<i data-lucide="x"></i>
Clear
</button>
<button class="btn btn--danger" id="deleteReferenceBtn" disabled>
<i data-lucide="trash-2"></i>
Delete
</button>
</div>
</div>
<!-- Current Reference Info -->
<div class="current-reference-info" id="currentReferenceInfo" style="display: none;">
<div class="reference-info-card">
<h5>Current Reference</h5>
<div class="reference-details">
<span class="reference-name" id="currentReferenceName">-</span>
<span class="reference-timestamp" id="currentReferenceTimestamp">-</span>
<p class="reference-description" id="currentReferenceDescription">-</p>
</div>
</div>
</div>
</div>
</div>
</div>
<!-- Status Summary -->
<div class="settings-card">
<h3 class="settings-card-title">Status Summary</h3>