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

added s21 s11 normalisations

Browse Source
This commit is contained in:
sad-santa
2026-02-17 17:29:24 +03:00
parent 80335a9983
commit 4897209773
2 changed files with 257 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
vna_system/core/processors/configs/bscan_config.json
+12 -10
View File
@@ -1,21 +1,23 @@
{
"open_air": false,
"subtract_mean_ascan": true,
"open_air": true,
"ach_norm_enabled": false,
"s11_norm_enabled": false,
"subtract_mean_ascan": false,
"axis": "abs",
"cut": 0.546,
"max": 0.5,
"gain": 1.0,
"start_freq": 400.0,
"stop_freq": 5170.0,
"cut": 0.183,
"max": 2.0,
"gain": 1.5,
"start_freq": 1730.0,
"stop_freq": 7320.0,
"clear_history": false,
"sigma": 0.54,
"sigma": 1.38,
"border_border_m": 0.09,
"if_normalize": false,
"if_draw_level": false,
"detection_level": 8.0,
"apply_eps_correction": true,
"eps_r": 3.3,
"eps_boundary_m": 0.12,
"eps_r": 3.7,
"eps_boundary_m": 0.0,
"data_limit": 500,
"y_min": -50,
"y_max": 40,
vna_system/core/processors/implementations/bscan_processor.py
+245 -4
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
import json
from datetime import datetime
from pathlib import Path
from typing import Any
@@ -38,6 +39,12 @@ class BScanProcessor(BaseProcessor):
# Local plot history (separate from sweep history maintained by BaseProcessor)
self._plot_history: list[dict[str, Any]] = []
self._ach_norm_curve: NDArray[np.complex128] | None = None
self._ach_norm_mtime: float | None = None
self._s11_norm_curve_1: NDArray[np.complex128] | None = None
self._s11_norm_curve_2: NDArray[np.complex128] | None = None
self._s11_norm_mtime_1: float | None = None
self._s11_norm_mtime_2: float | None = None
logger.info("BScanProcessor initialized", processor_id=self.processor_id)
@@ -49,6 +56,8 @@ class BScanProcessor(BaseProcessor):
"""Return default configuration values."""
return {
"open_air": False, # Toggle for reference usage
"ach_norm_enabled": False, # Toggle ACH normalization from JSON
"s11_norm_enabled": False, # Toggle S21 normalization using two S11 files
"subtract_mean_ascan": False, # Subtract mean A-scan from each sweep in B-scan
"axis": "abs", # "real", "abs", or "phase"
# "data_limitation": None, # None, "ph_only_1", "ph_only_2"
@@ -65,7 +74,7 @@ class BScanProcessor(BaseProcessor):
"detection_level" : 5,
"apply_eps_correction": False,
"eps_r": 4.0,
"eps_boundary_m": 0.5,
"eps_boundary_m": 0.0,
}
def get_ui_parameters(self) -> list[UIParameter]:
@@ -79,6 +88,18 @@ class BScanProcessor(BaseProcessor):
type="toggle",
value=cfg["open_air"],
),
UIParameter(
name="ach_norm_enabled",
label="\u041d\u043e\u0440\u043c\u0438\u0440\u043e\u0432\u043a\u0430 \u0410\u0427\u0425 \u043f\u043e s21",
type="toggle",
value=cfg["ach_norm_enabled"],
),
UIParameter(
name="s11_norm_enabled",
label="\u041d\u043e\u0440\u043c\u0438\u0440\u043e\u0432\u043a\u0430 \u0410\u0427\u0425 \u043f\u043e s11",
type="toggle",
value=cfg["s11_norm_enabled"],
),
UIParameter(
name="subtract_mean_ascan",
label="Вычесть средний A-скан",
@@ -181,20 +202,20 @@ class BScanProcessor(BaseProcessor):
# --- NEW: epsilon correction controls ---
UIParameter(
name="apply_eps_correction",
label="Учет ε ниже границы (новая)",
label="Учет ε ниже границы",
type="toggle",
value=cfg["apply_eps_correction"],
),
UIParameter(
name="eps_r",
label="εr ниже границы (новая)",
label="ε ниже границы",
type="slider",
value=cfg["eps_r"],
options={"min": 1.0, "max": 30.0, "step": 0.1, "dtype": "float"},
),
UIParameter(
name="eps_boundary_m",
label="Граница среды (ε) (м) (новая)",
label="Граница среды (ε) (м)",
type="slider",
value=cfg["eps_boundary_m"],
options={"min": 0.0, "max": 2.5, "step": 0.01, "dtype": "float"},
@@ -312,6 +333,10 @@ class BScanProcessor(BaseProcessor):
complex_data = self._subtract_reference(complex_data, reference_complex)
logger.debug("Applied open-air reference subtraction")
# Optional ACH normalization using fixed JSON file
complex_data = self._apply_ach_normalization(complex_data)
complex_data = self._apply_s11_normalization(complex_data)
# Keep frequency controls in sync with the current VNA config
self._update_frequency_ranges(vna_config)
@@ -675,6 +700,222 @@ class BScanProcessor(BaseProcessor):
logger.error("Reference subtraction failed", error=repr(exc))
return signal # Non-fatal; continue with original signal
def _get_ach_norm_file_path(self) -> Path:
"""Return fixed ACH normalization JSON path."""
return Path(__file__).resolve().parents[3] / "references" / "normalisation.json"
def _apply_ach_normalization(
self,
signal: NDArray[np.complex128],
) -> NDArray[np.complex128]:
"""
Apply ACH normalization from JSON:
signal = signal / calibrated_points
"""
if not self._config.get("ach_norm_enabled", False):
return signal
norm_curve = self._load_ach_norm_curve()
if norm_curve is None or norm_curve.size == 0:
logger.warning("ACH normalization enabled but no valid calibrated_points found")
return signal
n = min(signal.size, norm_curve.size)
if n == 0:
return signal
denom = norm_curve[:n].copy()
eps = 1e-12
zero_mask = np.abs(denom) < eps
if np.any(zero_mask):
denom[zero_mask] = eps + 0j
logger.debug("Applied ACH normalization", points=n)
out = signal.copy()
denom_abs = np.abs(denom)
denom_abs[denom_abs < eps] = eps
out[:n] = out[:n] / denom_abs
# out[:n] = out[:n] / denom
return out
def _load_ach_norm_curve(self) -> NDArray[np.complex128] | None:
"""Load normalization curve from fixed JSON file with mtime cache."""
file_path = self._get_ach_norm_file_path()
if not file_path.exists():
logger.warning("ACH normalization file not found", file=str(file_path))
return None
try:
mtime = file_path.stat().st_mtime
if self._ach_norm_curve is not None and self._ach_norm_mtime == mtime:
return self._ach_norm_curve
payload = json.loads(file_path.read_text(encoding="utf-8"))
points = self._find_last_calibrated_points(payload)
if not points:
return None
curve = self._points_to_complex(points)
if curve is None or curve.size == 0:
return None
self._ach_norm_curve = curve
self._ach_norm_mtime = mtime
logger.info("Loaded ACH normalization curve", file=str(file_path), points=curve.size)
return curve
except Exception as exc: # noqa: BLE001
logger.error("Failed to load ACH normalization file", file=str(file_path), error=repr(exc))
return None
def _find_last_calibrated_points(self, node: Any) -> list[Any] | None:
"""Recursively find the latest non-empty calibrated_points list."""
found: list[Any] | None = None
if isinstance(node, dict):
local = node.get("calibrated_points")
if isinstance(local, list) and local:
found = local
for value in node.values():
nested = self._find_last_calibrated_points(value)
if nested:
found = nested
return found
if isinstance(node, list):
for item in node:
nested = self._find_last_calibrated_points(item)
if nested:
found = nested
return found
return None
def _points_to_complex(self, points: list[Any]) -> NDArray[np.complex128] | None:
"""Convert JSON points (real, imag) into complex ndarray."""
out: list[complex] = []
for point in points:
real: Any = None
imag: Any = None
if isinstance(point, dict):
real = point.get("real", point.get("r"))
imag = point.get("imag", point.get("i"))
elif isinstance(point, (list, tuple)) and len(point) >= 2:
real, imag = point[0], point[1]
if real is None or imag is None:
continue
try:
out.append(complex(float(real), float(imag)))
except (TypeError, ValueError):
continue
if not out:
return None
return np.asarray(out, dtype=np.complex128)
def _get_s11_norm_file_paths(self) -> tuple[Path, Path]:
"""Return fixed file paths for S11-based normalization."""
candidates = [
Path(__file__).resolve().parents[3] / "references",
Path(__file__).resolve().parents[4] / "references",
]
base_dir = next((p for p in candidates if p.exists()), candidates[0])
file_1 = base_dir / "s11_blue_bl.json"
file_2 = base_dir / "s11_blue_gr.json"
return file_1, file_2
def _apply_s11_normalization(
self,
signal: NDArray[np.complex128],
) -> NDArray[np.complex128]:
"""
Apply S11-based normalization:
s21 = s21 / sqrt((1-|s11_1|^2) * (1-|s11_2|^2))
"""
if not self._config.get("s11_norm_enabled", False):
return signal
curves = self._load_s11_norm_curves()
if curves is None:
logger.warning("S11 normalization enabled but S11 curves are unavailable")
return signal
s11_1, s11_2 = curves
n = min(signal.size, s11_1.size, s11_2.size)
if n == 0:
return signal
abs_s11_1_sq = np.abs(s11_1[:n]) ** 2
abs_s11_2_sq = np.abs(s11_2[:n]) ** 2
product = (1.0 - abs_s11_1_sq) * (1.0 - abs_s11_2_sq)
eps = 1e-12
denom = np.sqrt(np.clip(product, eps, None))
out = signal.copy()
out[:n] = out[:n] / denom
logger.debug("Applied S11 normalization", points=n)
return out
def _load_s11_norm_curves(self) -> tuple[NDArray[np.complex128], NDArray[np.complex128]] | None:
"""Load both S11 normalization curves from fixed sweep_data.json files."""
file_1, file_2 = self._get_s11_norm_file_paths()
curve_1 = self._load_sweep_points_curve(
file_path=file_1,
curve_attr="_s11_norm_curve_1",
mtime_attr="_s11_norm_mtime_1",
)
curve_2 = self._load_sweep_points_curve(
file_path=file_2,
curve_attr="_s11_norm_curve_2",
mtime_attr="_s11_norm_mtime_2",
)
if curve_1 is None or curve_2 is None:
return None
return curve_1, curve_2
def _load_sweep_points_curve(
self,
file_path: Path,
curve_attr: str,
mtime_attr: str,
) -> NDArray[np.complex128] | None:
"""Load complex curve from sweep_data.json['points'] with mtime cache."""
if not file_path.exists():
logger.warning("S11 normalization file not found", file=str(file_path))
return None
try:
mtime = file_path.stat().st_mtime
cached_curve = getattr(self, curve_attr)
cached_mtime = getattr(self, mtime_attr)
if cached_curve is not None and cached_mtime == mtime:
return cached_curve
payload = json.loads(file_path.read_text(encoding="utf-8"))
points = self._find_last_calibrated_points(payload)
if not isinstance(points, list) or not points:
logger.warning("Invalid points in S11 normalization file", file=str(file_path))
return None
curve = self._points_to_complex(points)
if curve is None or curve.size == 0:
logger.warning("Failed to parse points in S11 normalization file", file=str(file_path))
return None
setattr(self, curve_attr, curve)
setattr(self, mtime_attr, mtime)
logger.info("Loaded S11 normalization curve", file=str(file_path), points=curve.size)
return curve
except Exception as exc: # noqa: BLE001
logger.error("Failed to load S11 normalization file", file=str(file_path), error=repr(exc))
return None
def _update_frequency_ranges(self, vna_config: dict[str, Any]) -> None:
"""Clamp configured frequency sliders to VNA limits."""
if not vna_config: