diff --git a/vna_system/core/processors/configs/bscan_config.json b/vna_system/core/processors/configs/bscan_config.json index 157dea2..986ec3e 100644 --- a/vna_system/core/processors/configs/bscan_config.json +++ b/vna_system/core/processors/configs/bscan_config.json @@ -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, diff --git a/vna_system/core/processors/implementations/bscan_processor.py b/vna_system/core/processors/implementations/bscan_processor.py index 80b6403..b3a7077 100644 --- a/vna_system/core/processors/implementations/bscan_processor.py +++ b/vna_system/core/processors/implementations/bscan_processor.py @@ -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: