surface_align added

optimization fix
2026-08-05 19:01:26 +03:00 · 2026-08-05 17:58:26 +03:00 · 2026-08-05 15:40:41 +03:00 · 2026-04-30 14:23:15 +03:00 · 2026-04-30 13:59:21 +03:00 · 2026-04-22 13:12:37 +03:00
78 changed files with 198300 additions and 50 deletions
@@ -4,4 +4,5 @@ fastapi==0.117.1
 numpy==2.3.3
 matplotlib==3.10.6
 plotly==6.3.0
+scipy==1.16.3
 # ??? kaleido==1.1.0
@@ -0,0 +1,18 @@
+{
+  "preset": {
+    "filename": "s11_start100_stop8800_points1000_bw1khz.bin",
+    "mode": "s11",
+    "start_freq": 100000000.0,
+    "stop_freq": 8800000000.0,
+    "points": 1000,
+    "bandwidth": 1000.0
+  },
+  "calibration_name": "sol-new-1m-cable_1",
+  "standards": [
+    "open",
+    "short",
+    "load"
+  ],
+  "created_timestamp": "2025-12-15T14:46:10.474190",
+  "is_complete": true
+}
@@ -0,0 +1,16 @@
+{
+  "preset": {
+    "filename": "s11_start100_stop8800_points1000_bw1khz.bin",
+    "mode": "s11",
+    "start_freq": 100000000.0,
+    "stop_freq": 8800000000.0,
+    "points": 1000,
+    "bandwidth": 1000.0
+  },
+  "calibration_name": "sol-new-1m-cable_1",
+  "standard": "load",
+  "sweep_number": 23,
+  "sweep_timestamp": 1765799144.3783984,
+  "created_timestamp": "2025-12-15T14:46:10.471566",
+  "total_points": 1000
+}
@@ -0,0 +1,16 @@
+{
+  "preset": {
+    "filename": "s11_start100_stop8800_points1000_bw1khz.bin",
+    "mode": "s11",
+    "start_freq": 100000000.0,
+    "stop_freq": 8800000000.0,
+    "points": 1000,
+    "bandwidth": 1000.0
+  },
+  "calibration_name": "sol-new-1m-cable_1",
+  "standard": "open",
+  "sweep_number": 21,
+  "sweep_timestamp": 1765799091.847419,
+  "created_timestamp": "2025-12-15T14:46:10.429933",
+  "total_points": 1000
+}
@@ -0,0 +1,16 @@
+{
+  "preset": {
+    "filename": "s11_start100_stop8800_points1000_bw1khz.bin",
+    "mode": "s11",
+    "start_freq": 100000000.0,
+    "stop_freq": 8800000000.0,
+    "points": 1000,
+    "bandwidth": 1000.0
+  },
+  "calibration_name": "sol-new-1m-cable_1",
+  "standard": "short",
+  "sweep_number": 22,
+  "sweep_timestamp": 1765799117.1214204,
+  "created_timestamp": "2025-12-15T14:46:10.455744",
+  "total_points": 1000
+}
@@ -0,0 +1,16 @@
+{
+  "preset": {
+    "filename": "s21_start100_stop8800_points1000_bw1khz.bin",
+    "mode": "s21",
+    "start_freq": 100000000.0,
+    "stop_freq": 8800000000.0,
+    "points": 1000,
+    "bandwidth": 1000.0
+  },
+  "calibration_name": "T_N2_451U_x2_noamp_20251105",
+  "standards": [
+    "through"
+  ],
+  "created_timestamp": "2025-11-05T20:50:07.441271",
+  "is_complete": true
+}
@@ -0,0 +1,16 @@
+{
+  "preset": {
+    "filename": "s21_start100_stop8800_points1000_bw1khz.bin",
+    "mode": "s21",
+    "start_freq": 100000000.0,
+    "stop_freq": 8800000000.0,
+    "points": 1000,
+    "bandwidth": 1000.0
+  },
+  "calibration_name": "T_N2_451U_x2_noamp_20251105",
+  "standard": "through",
+  "sweep_number": 14,
+  "sweep_timestamp": 1762365005.456789,
+  "created_timestamp": "2025-11-05T20:50:07.440235",
+  "total_points": 1000
+}
@@ -0,0 +1,16 @@
+{
+  "preset": {
+    "filename": "s21_start100_stop8800_points1000_bw100hz.bin",
+    "mode": "s21",
+    "start_freq": 100000000.0,
+    "stop_freq": 8800000000.0,
+    "points": 1000,
+    "bandwidth": 100.0
+  },
+  "calibration_name": "blue_ann_through_100hz",
+  "standards": [
+    "through"
+  ],
+  "created_timestamp": "2025-12-04T15:04:35.142648",
+  "is_complete": true
+}
@@ -0,0 +1,16 @@
+{
+  "preset": {
+    "filename": "s21_start100_stop8800_points1000_bw100hz.bin",
+    "mode": "s21",
+    "start_freq": 100000000.0,
+    "stop_freq": 8800000000.0,
+    "points": 1000,
+    "bandwidth": 100.0
+  },
+  "calibration_name": "blue_ann_through_100hz",
+  "standard": "through",
+  "sweep_number": 23,
+  "sweep_timestamp": 1764849859.7482991,
+  "created_timestamp": "2025-12-04T15:04:35.140647",
+  "total_points": 1000
+}
@@ -0,0 +1,16 @@
+{
+  "preset": {
+    "filename": "s21_start100_stop8800_points1000_bw100hz.bin",
+    "mode": "s21",
+    "start_freq": 100000000.0,
+    "stop_freq": 8800000000.0,
+    "points": 1000,
+    "bandwidth": 100.0
+  },
+  "calibration_name": "blue_ann_through_20hz",
+  "standards": [
+    "through"
+  ],
+  "created_timestamp": "2025-12-04T15:04:56.454793",
+  "is_complete": true
+}
@@ -0,0 +1,16 @@
+{
+  "preset": {
+    "filename": "s21_start100_stop8800_points1000_bw100hz.bin",
+    "mode": "s21",
+    "start_freq": 100000000.0,
+    "stop_freq": 8800000000.0,
+    "points": 1000,
+    "bandwidth": 100.0
+  },
+  "calibration_name": "blue_ann_through_20hz",
+  "standard": "through",
+  "sweep_number": 24,
+  "sweep_timestamp": 1764849893.2069495,
+  "created_timestamp": "2025-12-04T15:04:56.453793",
+  "total_points": 1000
+}
@@ -0,0 +1,16 @@
+{
+  "preset": {
+    "filename": "s21_start100_stop8800_points1000_bw1khz.bin",
+    "mode": "s21",
+    "start_freq": 100000000.0,
+    "stop_freq": 8800000000.0,
+    "points": 1000,
+    "bandwidth": 1000.0
+  },
+  "calibration_name": "T_N2_tair_451U_x2+rfc18-1m",
+  "standards": [
+    "through"
+  ],
+  "created_timestamp": "2025-12-19T17:03:08.354734",
+  "is_complete": true
+}
@@ -0,0 +1,16 @@
+{
+  "preset": {
+    "filename": "s21_start100_stop8800_points1000_bw1khz.bin",
+    "mode": "s21",
+    "start_freq": 100000000.0,
+    "stop_freq": 8800000000.0,
+    "points": 1000,
+    "bandwidth": 1000.0
+  },
+  "calibration_name": "T_N2_tair_451U_x2+rfc18-1m",
+  "standard": "through",
+  "sweep_number": 4,
+  "sweep_timestamp": 1766152953.8570125,
+  "created_timestamp": "2025-12-19T17:03:08.349898",
+  "total_points": 1000
+}
@@ -0,0 +1,16 @@
+{
+  "preset": {
+    "filename": "s21_start100_stop8800_points1000_bw1khz.bin",
+    "mode": "s21",
+    "start_freq": 100000000.0,
+    "stop_freq": 8800000000.0,
+    "points": 1000,
+    "bandwidth": 1000.0
+  },
+  "calibration_name": "blue_ann_through_1khz",
+  "standards": [
+    "through"
+  ],
+  "created_timestamp": "2025-12-04T15:05:28.008925",
+  "is_complete": true
+}
@@ -0,0 +1,16 @@
+{
+  "preset": {
+    "filename": "s21_start100_stop8800_points1000_bw1khz.bin",
+    "mode": "s21",
+    "start_freq": 100000000.0,
+    "stop_freq": 8800000000.0,
+    "points": 1000,
+    "bandwidth": 1000.0
+  },
+  "calibration_name": "blue_ann_through_1khz",
+  "standard": "through",
+  "sweep_number": 25,
+  "sweep_timestamp": 1764849920.0909598,
+  "created_timestamp": "2025-12-04T15:05:28.007879",
+  "total_points": 1000
+}
@@ -0,0 +1,16 @@
+{
+  "preset": {
+    "filename": "s21_start100_stop8800_points1000_bw1khz.bin",
+    "mode": "s21",
+    "start_freq": 100000000.0,
+    "stop_freq": 8800000000.0,
+    "points": 1000,
+    "bandwidth": 1000.0
+  },
+  "calibration_name": "blueann_through_1khz",
+  "standards": [
+    "through"
+  ],
+  "created_timestamp": "2025-12-04T15:03:58.727780",
+  "is_complete": true
+}
@@ -0,0 +1,16 @@
+{
+  "preset": {
+    "filename": "s21_start100_stop8800_points1000_bw1khz.bin",
+    "mode": "s21",
+    "start_freq": 100000000.0,
+    "stop_freq": 8800000000.0,
+    "points": 1000,
+    "bandwidth": 1000.0
+  },
+  "calibration_name": "blueann_through_1khz",
+  "standard": "through",
+  "sweep_number": 22,
+  "sweep_timestamp": 1764849814.6352062,
+  "created_timestamp": "2025-12-04T15:03:58.726778",
+  "total_points": 1000
+}
@@ -0,0 +1,16 @@
+{
+  "preset": {
+    "filename": "s21_start100_stop8800_points1000_bw1khz.bin",
+    "mode": "s21",
+    "start_freq": 100000000.0,
+    "stop_freq": 8800000000.0,
+    "points": 1000,
+    "bandwidth": 1000.0
+  },
+  "calibration_name": "thought-25-12-15",
+  "standards": [
+    "through"
+  ],
+  "created_timestamp": "2025-12-15T13:36:06.578246",
+  "is_complete": true
+}
@@ -0,0 +1,16 @@
+{
+  "preset": {
+    "filename": "s21_start100_stop8800_points1000_bw1khz.bin",
+    "mode": "s21",
+    "start_freq": 100000000.0,
+    "stop_freq": 8800000000.0,
+    "points": 1000,
+    "bandwidth": 1000.0
+  },
+  "calibration_name": "thought-25-12-15",
+  "standard": "through",
+  "sweep_number": 31,
+  "sweep_timestamp": 1765794952.1075795,
+  "created_timestamp": "2025-12-15T13:36:06.577088",
+  "total_points": 1000
+}
@@ -0,0 +1,16 @@
+{
+  "preset": {
+    "filename": "s21_start100_stop8800_points1000_bw1khz.bin",
+    "mode": "s21",
+    "start_freq": 100000000.0,
+    "stop_freq": 8800000000.0,
+    "points": 1000,
+    "bandwidth": 1000.0
+  },
+  "calibration_name": "T_N2_451U_x2_noamp_20251105",
+  "standards": [
+    "through"
+  ],
+  "created_timestamp": "2025-11-05T20:50:07.441271",
+  "is_complete": true
+}
@@ -0,0 +1,16 @@
+{
+  "preset": {
+    "filename": "s21_start100_stop8800_points1000_bw1khz.bin",
+    "mode": "s21",
+    "start_freq": 100000000.0,
+    "stop_freq": 8800000000.0,
+    "points": 1000,
+    "bandwidth": 1000.0
+  },
+  "calibration_name": "T_N2_451U_x2_noamp_20251105",
+  "standard": "through",
+  "sweep_number": 14,
+  "sweep_timestamp": 1762365005.456789,
+  "created_timestamp": "2025-11-05T20:50:07.440235",
+  "total_points": 1000
+}
@@ -0,0 +1,16 @@
+{
+  "preset": {
+    "filename": "s21_start400_stop6000_points1000_bw20hz.bin",
+    "mode": "s21",
+    "start_freq": 400000000.0,
+    "stop_freq": 6000000000.0,
+    "points": 1000,
+    "bandwidth": 20.0
+  },
+  "calibration_name": "blue_through_20hz",
+  "standards": [
+    "through"
+  ],
+  "created_timestamp": "2025-12-05T19:15:54.254887",
+  "is_complete": true
+}
@@ -0,0 +1,16 @@
+{
+  "preset": {
+    "filename": "s21_start400_stop6000_points1000_bw20hz.bin",
+    "mode": "s21",
+    "start_freq": 400000000.0,
+    "stop_freq": 6000000000.0,
+    "points": 1000,
+    "bandwidth": 20.0
+  },
+  "calibration_name": "blue_through_20hz",
+  "standard": "through",
+  "sweep_number": 4,
+  "sweep_timestamp": 1764949677.0854876,
+  "created_timestamp": "2025-12-05T19:15:54.252679",
+  "total_points": 1000
+}
@@ -33,7 +33,7 @@ VNA_VID = 0x0483  # STMicroelectronics
 VNA_PID = 0x5740  # STM32 Virtual ComPort

 # -----------------------------------------------------------------------------
-# Simulator mode settings
+# Simulator mode settings..
 # -----------------------------------------------------------------------------
 USE_SIMULATOR = True  # Set to True to use simulator instead of real device
 SIMULATOR_SWEEP_FILE = BASE_DIR / "binary_input" / "sweep_example" / "example.json"
@@ -0,0 +1,25 @@
+"""Signal-processing algorithms shared by the VNA processors."""
+
+from vna_system.core.processing.surface_align import (
+    AlignConfig,
+    AlignResult,
+    apply_alignment,
+    check_geometry,
+    estimate_surface,
+)
+from vna_system.core.processing.surface_alignment import (
+    AlignmentState,
+    AlignOptions,
+    SurfaceAlignmentEngine,
+)
+
+__all__ = [
+    "AlignConfig",
+    "AlignResult",
+    "AlignOptions",
+    "AlignmentState",
+    "SurfaceAlignmentEngine",
+    "apply_alignment",
+    "check_geometry",
+    "estimate_surface",
+]
@@ -0,0 +1,316 @@
+#!/usr/bin/env python3
+"""Ground-surface alignment for a multi-channel air-coupled SFCW GPR.
+
+Vendored verbatim from radar_system/python_app/processing/surface_align.py.
+Keep it dependency-free (numpy only) and keep the public contract stable --
+`AlignConfig`, `estimate_surface`, `apply_alignment`, `check_geometry` -- so the
+two copies stay interchangeable.  The application-side glue lives in
+`surface_alignment.py`; put integration logic there, not here.
+
+The estimator does not track each channel separately.  It solves for the single
+physical quantity that explains every channel at once -- the antenna height h --
+using the known bistatic geometry:
+
+    path_c(h) = sqrt(L_c^2 + (2h)^2)        L_c = transmitter-receiver baseline
+
+A wrong pick in one channel does not lie on the other channels' hyperbolae, so
+the joint score suppresses it instead of following it.
+
+Two smoothing modes:
+  * "offline"  zero-phase low-pass  (replay: whole profile available)
+  * "live"     constant-velocity Kalman (only the past is available)
+
+Background suppression is the operator's choice: a recorded reference, the
+global median trace, or a sliding-window median.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Literal
+
+import numpy as np
+
+C_LIGHT = 299_792_458.0
+
+
+# --------------------------------------------------------------------------- #
+#  configuration
+# --------------------------------------------------------------------------- #
+@dataclass
+class AlignConfig:
+    band_hz: tuple[float, float] = (0.3e9, 5.0e9)
+    pad_factor: int = 16                 # zero-padding -> finer peak grid
+    height_range_m: tuple[float, float] = (0.15, 0.90)
+    height_step_m: float = 0.001
+    track_halfwidth_m: float = 0.09      # search window around the prediction
+    # first-break: take the first score peak reaching this fraction of the best
+    first_break_frac: float = 0.45
+    min_confidence: float = 2.0          # peak / median of the score curve
+    background: Literal["reference", "global_median", "sliding_median"] = "global_median"
+    sliding_window: int = 101            # traces, for background="sliding_median"
+    smoothing: Literal["offline", "live"] = "offline"
+    lowpass_hz: float = 1.5              # offline mode
+    process_noise_m: float = 0.02        # live mode: how fast h may change
+    measurement_noise_m: float = 0.005
+    mad_floor_m: float = 0.002
+    outlier_sigmas: float = 4.0
+
+
+@dataclass
+class AlignResult:
+    height_m: np.ndarray                 # per cycle, smoothed
+    height_raw_m: np.ndarray             # per cycle, before smoothing
+    confidence: np.ndarray
+    accepted: np.ndarray                 # bool, survived outlier rejection
+    baselines_m: dict                    # combo -> L
+    quality: dict = field(default_factory=dict)
+
+
+# --------------------------------------------------------------------------- #
+#  helpers
+# --------------------------------------------------------------------------- #
+def _to_time_domain(spectra: np.ndarray, freq: np.ndarray, pad: int):
+    """Windowed IFFT -> analytic traces on a zero-padded time grid."""
+    n = freq.size
+    N = n * pad
+    df = float(freq[1] - freq[0])
+    z = np.fft.ifft(spectra * np.hanning(n), n=N, axis=-1) * N
+    t = np.arange(N) / (N * df)
+    return z, t
+
+
+def _suppress_background(z: np.ndarray, cfg: AlignConfig,
+                         reference: np.ndarray | None) -> np.ndarray:
+    """Remove whatever is stationary, by the operator's chosen means."""
+    if cfg.background == "reference":
+        if reference is None:
+            raise ValueError("background='reference' requires a reference trace")
+        return z - reference[None, :]
+    if cfg.background == "global_median":
+        bg = np.median(z.real, axis=0) + 1j * np.median(z.imag, axis=0)
+        return z - bg[None, :]
+    if cfg.background == "sliding_median":
+        w = max(3, int(cfg.sliding_window) | 1)
+        half = w // 2
+        out = np.empty_like(z)
+        for i in range(z.shape[0]):
+            lo, hi = max(0, i - half), min(z.shape[0], i + half + 1)
+            bg = (np.median(z[lo:hi].real, axis=0)
+                  + 1j * np.median(z[lo:hi].imag, axis=0))
+            out[i] = z[i] - bg
+        return out
+    raise ValueError(f"unknown background mode {cfg.background!r}")
+
+
+def estimate_baselines(z_by_combo: dict, t: np.ndarray,
+                       search_m: tuple[float, float] = (0.10, 1.20)) -> dict:
+    """Baseline L from the stationary echo: it sits at one-way delay L/c.
+
+    Returns {combo: L}.  Verify with `check_geometry` before trusting it -- if
+    the stationary echo is a frame reflection rather than the direct path, the
+    per-channel heights will disagree.
+    """
+    out = {}
+    for combo, z in z_by_combo.items():
+        fixed = np.median(z.real, axis=0) + 1j * np.median(z.imag, axis=0)
+        d = C_LIGHT * t                      # one-way path length
+        m = (d >= search_m[0]) & (d <= search_m[1])
+        idx = np.where(m)[0]
+        k = idx[int(np.argmax(np.abs(fixed[idx])))]
+        out[combo] = float(d[k])
+    return out
+
+
+def _parabolic(y0: float, y1: float, y2: float) -> float:
+    den = y0 - 2.0 * y1 + y2
+    if den == 0.0:
+        return 0.0
+    return float(np.clip(0.5 * (y0 - y2) / den, -1.0, 1.0))
+
+
+# --------------------------------------------------------------------------- #
+#  the joint estimator
+# --------------------------------------------------------------------------- #
+def _score_curve(env_by_combo: dict, baselines: dict, t: np.ndarray,
+                 heights: np.ndarray, cycle: int) -> np.ndarray:
+    """Sum the channels' envelopes along the bistatic hyperbola for each h."""
+    dt = t[1] - t[0]
+    score = np.zeros_like(heights)
+    for combo, env in env_by_combo.items():
+        L = baselines[combo]
+        tau = np.sqrt(L * L + 4.0 * heights * heights) / C_LIGHT
+        idx = tau / dt
+        i0 = np.floor(idx).astype(int)
+        frac = idx - i0
+        ok = (i0 >= 0) & (i0 + 1 < env.shape[1])
+        row = env[cycle]
+        vals = np.zeros_like(heights)
+        vals[ok] = row[i0[ok]] * (1 - frac[ok]) + row[i0[ok] + 1] * frac[ok]
+        # normalise per channel so a loud channel cannot dominate the vote
+        scale = np.median(row) + 1e-30
+        score += vals / scale
+    return score
+
+
+def _pick_from_score(score: np.ndarray, heights: np.ndarray,
+                     lo: int, hi: int, cfg: AlignConfig) -> tuple[float, float]:
+    """First significant peak inside [lo, hi) -- nothing subsurface can be
+    shallower than the surface, so the *first* arrival is the right target."""
+    seg = score[lo:hi]
+    if seg.size < 3:
+        return float(heights[(lo + hi) // 2]), 0.0
+    thr = cfg.first_break_frac * seg.max()
+    k = None
+    for i in range(1, seg.size - 1):
+        if seg[i] >= thr and seg[i] >= seg[i - 1] and seg[i] >= seg[i + 1]:
+            k = i
+            break
+    if k is None:
+        k = int(np.argmax(seg))
+    delta = _parabolic(seg[k - 1], seg[k], seg[k + 1]) if 0 < k < seg.size - 1 else 0.0
+    step = heights[1] - heights[0]
+    h = float(heights[lo + k] + delta * step)
+    conf = float(seg[k] / (np.median(score) + 1e-30))
+    return h, conf
+
+
+def _kalman(meas: np.ndarray, conf: np.ndarray, dt: float,
+            cfg: AlignConfig) -> np.ndarray:
+    """Constant-velocity Kalman.  Causal: the estimate for cycle i uses only
+    cycles <= i, so it is usable live, and unlike a causal low-pass it does not
+    lag (the velocity state extrapolates)."""
+    x = np.array([meas[0], 0.0])
+    P = np.diag([1e-2, 1e-2])
+    F = np.array([[1.0, dt], [0.0, 1.0]])
+    q = cfg.process_noise_m ** 2
+    Q = q * np.array([[dt ** 3 / 3.0, dt ** 2 / 2.0], [dt ** 2 / 2.0, dt]])
+    H = np.array([[1.0, 0.0]])
+    out = np.empty_like(meas)
+    for i, (zm, cf) in enumerate(zip(meas, conf)):
+        x = F @ x
+        P = F @ P @ F.T + Q
+        if cf >= cfg.min_confidence:
+            r = cfg.measurement_noise_m ** 2 * max(1.0, cfg.min_confidence / cf)
+            S = (H @ P @ H.T).item() + r
+            K = (P @ H.T / S).ravel()
+            x = x + K * (zm - (H @ x).item())
+            P = (np.eye(2) - np.outer(K, H)) @ P
+        out[i] = x[0]
+    return out
+
+
+def _zero_phase_lowpass(y: np.ndarray, fs: float, cut: float) -> np.ndarray:
+    sp = np.fft.rfft(y - y.mean())
+    fr = np.fft.rfftfreq(y.size, 1.0 / fs)
+    sp[fr > cut] = 0.0
+    return np.fft.irfft(sp, n=y.size) + y.mean()
+
+
+def estimate_surface(spectra_by_combo: dict, freq: np.ndarray,
+                     times_s: np.ndarray, cfg: AlignConfig,
+                     baselines: dict | None = None,
+                     references: dict | None = None) -> AlignResult:
+    """Estimate the antenna height for every cycle, jointly over all channels."""
+    z_by_combo, env_by_combo = {}, {}
+    t = None
+    for combo, S in spectra_by_combo.items():
+        z, t = _to_time_domain(S, freq, cfg.pad_factor)
+        z_by_combo[combo] = z
+    if baselines is None:
+        baselines = estimate_baselines(z_by_combo, t)
+    for combo, z in z_by_combo.items():
+        ref = None if references is None else references.get(combo)
+        env_by_combo[combo] = np.abs(_suppress_background(z, cfg, ref))
+
+    heights = np.arange(cfg.height_range_m[0], cfg.height_range_m[1],
+                        cfg.height_step_m)
+    n_cycles = next(iter(env_by_combo.values())).shape[0]
+    raw = np.zeros(n_cycles)
+    conf = np.zeros(n_cycles)
+    half = int(round(cfg.track_halfwidth_m / cfg.height_step_m))
+
+    prev = None
+    for i in range(n_cycles):
+        score = _score_curve(env_by_combo, baselines, t, heights, i)
+        if prev is None:
+            lo, hi = 0, heights.size
+        else:
+            c = int(round((prev - heights[0]) / cfg.height_step_m))
+            lo, hi = max(0, c - half), min(heights.size, c + half + 1)
+        h, cf = _pick_from_score(score, heights, lo, hi, cfg)
+        raw[i], conf[i] = h, cf
+        if cf >= cfg.min_confidence:
+            prev = h
+        elif prev is None:
+            prev = h
+
+    # robust outlier rejection against a running median
+    med = np.array([np.median(raw[max(0, i - 4):i + 5]) for i in range(n_cycles)])
+    resid = raw - med
+    mad = max(1.4826 * float(np.median(np.abs(resid - np.median(resid)))),
+              cfg.mad_floor_m)
+    accepted = (np.abs(resid) < cfg.outlier_sigmas * mad) & (conf >= cfg.min_confidence)
+    if accepted.sum() < 3:
+        accepted = np.ones(n_cycles, dtype=bool)
+
+    filled = np.interp(times_s, times_s[accepted], raw[accepted])
+    dt = float(np.median(np.diff(times_s)))
+    if cfg.smoothing == "offline":
+        height = _zero_phase_lowpass(filled, 1.0 / dt, cfg.lowpass_hz)
+    else:
+        height = _kalman(filled, conf, dt, cfg)
+
+    quality = {
+        "rejected_fraction": float(1.0 - accepted.mean()),
+        "confidence_median": float(np.median(conf)),
+        "mad_m": float(mad),
+        "height_p2p_m": float(height.max() - height.min()),
+        "trace_rate_hz": 1.0 / dt,
+    }
+    return AlignResult(height, raw, conf, accepted, baselines, quality)
+
+
+def check_geometry(spectra_by_combo: dict, freq: np.ndarray, cfg: AlignConfig,
+                   baselines: dict, height_m: np.ndarray) -> dict:
+    """Per-channel height implied by each channel's own picked delay.
+
+    All channels see the same platform, so these must agree.  Disagreement means
+    a channel is tracking something that is not the surface -- exactly the
+    failure a single-channel tracker cannot detect.
+    """
+    out = {}
+    for combo, S in spectra_by_combo.items():
+        z, t = _to_time_domain(S, freq, cfg.pad_factor)
+        env = np.abs(_suppress_background(z, cfg, None))
+        L = baselines[combo]
+        dt = t[1] - t[0]
+        hs = []
+        for i in range(env.shape[0]):
+            tau_pred = np.sqrt(L * L + 4 * height_m[i] ** 2) / C_LIGHT
+            k0 = int(round(tau_pred / dt))
+            w = int(round(0.15 / C_LIGHT / dt))     # +-15 cm of path
+            lo, hi = max(0, k0 - w), min(env.shape[1], k0 + w + 1)
+            k = lo + int(np.argmax(env[i, lo:hi]))
+            p = (C_LIGHT * t[k]) ** 2 - L * L
+            hs.append(np.sqrt(max(p, 0.0)) / 2.0)
+        out[combo] = np.array(hs)
+    return out
+
+
+def apply_alignment(spectra_by_combo: dict, freq: np.ndarray,
+                    baselines: dict, height_m: np.ndarray,
+                    height_ref_m: float | None = None) -> dict:
+    """Shift every channel so the surface lands at a common reference height.
+
+    The shift is a linear phase ramp, which is an *exact* fractional delay: the
+    magnitude of exp(j2*pi*f*tau) is one, so nothing is filtered or interpolated.
+    """
+    if height_ref_m is None:
+        height_ref_m = float(np.median(height_m))
+    out = {}
+    for combo, S in spectra_by_combo.items():
+        L = baselines[combo]
+        tau = np.sqrt(L * L + 4.0 * height_m ** 2) / C_LIGHT
+        tau_ref = np.sqrt(L * L + 4.0 * height_ref_m ** 2) / C_LIGHT
+        out[combo] = S * np.exp(1j * 2 * np.pi * np.outer(tau - tau_ref, freq))
+    return out
@@ -0,0 +1,479 @@
+"""Glue between the sweep-history model of the processors and `surface_align`.
+
+`surface_align` works on a whole profile at once: it wants every cycle's
+spectrum, a uniform frequency grid and the cycle timestamps. A live VNA hands us
+one sweep at a time, and the B-scan can be rebuilt from scratch at any moment
+when the operator changes a setting. This module reconciles the two:
+
+* `plan()` decides whether the cached estimate is still usable, whether it can be
+  extended with a trailing window (the live case), or whether everything has to
+  be recomputed (config change, history reload).
+* `update()` runs the estimator for the chosen mode and keeps one height per
+  history index, so a processor can ask "what was the height for cycle i?".
+* `shift()` applies the alignment to a single spectrum over the *full* measured
+  band, even though the height was estimated on a narrower band.
+
+The estimator itself is untouched -- all physics lives in `surface_align`.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field, replace
+from typing import Any, Literal
+
+import numpy as np
+from numpy.typing import NDArray
+
+from vna_system.core.logging.logger import get_component_logger
+from vna_system.core.processing.surface_align import (
+    AlignConfig,
+    apply_alignment,
+    check_geometry,
+    estimate_surface,
+)
+# The estimator's own windowed IFFT. Reused rather than reimplemented so a
+# recorded reference lands on exactly the time grid the estimator works on.
+from vna_system.core.processing.surface_align import _to_time_domain
+
+logger = get_component_logger(__file__)
+
+PlanMode = Literal["reuse", "extend", "full"]
+
+# A single VNA measures one TX/RX pair per sweep, so there is exactly one
+# channel. The estimator is dict-based, so adding channels later is a matter of
+# passing more keys.
+DEFAULT_CHANNEL = "vna"
+
+
+@dataclass(frozen=True)
+class AlignOptions:
+    """Everything the caller can tune, in SI units.
+
+    Frozen so it can be compared as a whole: any change invalidates the cached
+    estimate, which is exactly the desired behaviour.
+    """
+    band_hz: tuple[float, float] = (0.3e9, 5.0e9)
+    height_range_m: tuple[float, float] = (0.25, 0.90)
+    height_step_m: float = 0.001
+    background: str = "reference"
+    smoothing: str = "live"
+    pad_factor: int = 8
+    baseline_m: float | None = None      # None -> from the stationary echo
+    height_ref_m: float | None = None    # None or 0 -> median of the first estimate
+    geometry_check: bool = True
+
+
+@dataclass
+class AlignmentState:
+    """One height per history index, plus the numbers worth watching."""
+    heights_m: NDArray[np.floating]
+    raw_heights_m: NDArray[np.floating]
+    confidence: NDArray[np.floating]
+    baseline_m: float
+    height_ref_m: float
+    quality: dict[str, float] = field(default_factory=dict)
+    geometry_residual_m: float | None = None
+    geometry_spread_m: float | None = None
+    # True once heights were appended from a trailing window rather than
+    # estimated over the whole profile in one pass.
+    incremental: bool = False
+
+    @property
+    def covered(self) -> int:
+        """Number of cycles the state has a height for."""
+        return int(self.heights_m.size)
+
+
+class SurfaceAlignmentEngine:
+    """Cached, incrementally extendable surface-height estimate."""
+
+    # Median-based background needs several traces before it means anything.
+    MIN_CYCLES = 8
+    # With a recorded reference the background is known up front, but the
+    # smoothers still need a trace interval, i.e. more than one cycle.
+    MIN_REFERENCE_CYCLES = 3
+    # Trailing window used to extend the estimate live.
+    WINDOW_CYCLES = 128
+    # More new cycles than this and a full recompute is cheaper/safer.
+    MAX_EXTEND = 32
+    # Below this many frequency points the band is too narrow to pick a peak.
+    MIN_BAND_POINTS = 16
+
+    def __init__(self, channel: str = DEFAULT_CHANNEL) -> None:
+        self.channel = channel
+        self._state: AlignmentState | None = None
+        self._key: tuple[Any, ...] | None = None
+        # The reference fallback is decided per estimate, i.e. once per sweep
+        # while streaming; warn about it once instead of on every sweep.
+        self._warned_reference_missing = False
+
+    # ------------------------------------------------------------------ #
+    # Cache control
+    # ------------------------------------------------------------------ #
+    @property
+    def state(self) -> AlignmentState | None:
+        return self._state
+
+    def invalidate(self) -> None:
+        """Forget the estimate; the next `plan()` asks for a full recompute."""
+        self._state = None
+        self._key = None
+        self._warned_reference_missing = False
+
+    def refresh_if_incremental(self, smoothing: str) -> None:
+        """Drop an incrementally built estimate before a full rebuild.
+
+        Live sweeps can only be smoothed causally, so an estimate grown one
+        window at a time is not what "offline" smoothing promises: that mode
+        low-passes the whole profile at once. A rebuild is the moment to deliver
+        it. Cheap modes are left alone -- the Kalman is causal by design, so
+        re-running it over everything would change little.
+        """
+        if smoothing == "offline" and self._state is not None and self._state.incremental:
+            logger.debug("Dropping incremental surface estimate for an offline rebuild")
+            self.invalidate()
+
+    def drop_leading(self, count: int) -> None:
+        """Drop heights for the `count` oldest cycles (history was trimmed)."""
+        if count <= 0 or self._state is None:
+            return
+        state = self._state
+        if count >= state.covered:
+            self.invalidate()
+            return
+        self._state = replace(
+            state,
+            heights_m=state.heights_m[count:],
+            raw_heights_m=state.raw_heights_m[count:],
+            confidence=state.confidence[count:],
+        )
+
+    # ------------------------------------------------------------------ #
+    # Planning
+    # ------------------------------------------------------------------ #
+    @staticmethod
+    def freq_key(freq_hz: NDArray[np.floating]) -> tuple[float, float, int]:
+        """Identity of a frequency grid: a different grid invalidates the cache."""
+        return float(freq_hz[0]), float(freq_hz[-1]), int(freq_hz.size)
+
+    def plan(
+        self,
+        n_cycles: int,
+        options: AlignOptions,
+        freq_key: tuple[float, float, int],
+    ) -> tuple[PlanMode, int]:
+        """Decide what has to be computed for cycles `[0, n_cycles)`.
+
+        Returns
+        -------
+        (mode, start_index)
+            `start_index` is the first cycle whose spectrum the caller must
+            supply to `update()`.
+        """
+        key = (options, freq_key, self.channel)
+        if key != self._key or self._state is None:
+            return "full", 0
+
+        covered = self._state.covered
+        if covered == n_cycles:
+            return "reuse", 0
+        if covered < n_cycles <= covered + self.MAX_EXTEND:
+            return "extend", max(0, n_cycles - self.WINDOW_CYCLES)
+        return "full", 0
+
+    # ------------------------------------------------------------------ #
+    # Estimation
+    # ------------------------------------------------------------------ #
+    def update(
+        self,
+        mode: PlanMode,
+        start_index: int,
+        spectra: NDArray[np.complexfloating],
+        freq_hz: NDArray[np.floating],
+        times_s: NDArray[np.floating],
+        options: AlignOptions,
+        reference: NDArray[np.complexfloating] | None = None,
+    ) -> AlignmentState | None:
+        """Run the estimator for `mode` and return the updated state.
+
+        `spectra` covers cycles `[start_index, n_cycles)` over the full measured
+        band; `freq_hz` is that band's grid (uniform, Hz). Returns None when the
+        data cannot support an estimate yet, in which case the caller should
+        leave the sweeps unshifted.
+        """
+        if mode == "reuse" and self._state is not None:
+            return self._state
+
+        band = (freq_hz >= options.band_hz[0]) & (freq_hz <= options.band_hz[1])
+        if int(band.sum()) < self.MIN_BAND_POINTS:
+            logger.warning(
+                "Surface alignment band too narrow; skipping",
+                points=int(band.sum()),
+                band_hz=options.band_hz,
+            )
+            return None
+
+        fb = np.asarray(freq_hz, dtype=float)[band]
+        window = np.ascontiguousarray(np.asarray(spectra, dtype=np.complex128)[:, band])
+        times = self._sanitize_times(np.asarray(times_s, dtype=float))
+        n_window = window.shape[0]
+        n_cycles = start_index + n_window
+
+        cfg, refs = self._build_config(options, fb, times, reference, band, n_cycles)
+        if cfg is None:
+            return None
+
+        if mode == "extend" and self._state is not None:
+            return self._extend(cfg, refs, fb, window, times, n_cycles)
+
+        state = self._full(cfg, refs, fb, window, times, options)
+        if state is not None:
+            self._key = (options, self.freq_key(freq_hz), self.channel)
+            self._state = state
+        return state
+
+    def _build_config(
+        self,
+        options: AlignOptions,
+        fb: NDArray[np.floating],
+        times: NDArray[np.floating],
+        reference: NDArray[np.complexfloating] | None,
+        band: NDArray[np.bool_],
+        n_cycles: int,
+    ) -> tuple[AlignConfig | None, dict[str, NDArray[np.complexfloating]] | None]:
+        """Translate `AlignOptions` into an `AlignConfig` plus reference traces."""
+        background = options.background
+        refs: dict[str, NDArray[np.complexfloating]] | None = None
+
+        if background == "reference":
+            if reference is None:
+                if not self._warned_reference_missing:
+                    logger.warning(
+                        "Alignment background='reference' but no matching reference sweep; "
+                        "falling back to the global median"
+                    )
+                    self._warned_reference_missing = True
+                background = "global_median"
+            else:
+                ref_band = np.asarray(reference, dtype=np.complex128)[band]
+                z_ref, _ = _to_time_domain(ref_band[None, :], fb, options.pad_factor)
+                refs = {self.channel: z_ref[0]}
+
+        # A median background is built across cycles, so it needs a few of them.
+        min_cycles = self.MIN_REFERENCE_CYCLES if refs is not None else self.MIN_CYCLES
+        if n_cycles < min_cycles:
+            logger.debug("Not enough cycles for surface alignment yet", cycles=n_cycles, needed=min_cycles)
+            return None, None
+
+        dt = self._median_dt(times)
+        trace_rate_hz = 1.0 / dt
+        # The estimator's 1.5 Hz default assumes a fast towed profile. A VNA
+        # sweep takes about a second, so tie the cutoff to the actual trace rate
+        # instead -- otherwise it sits above Nyquist and smooths nothing.
+        lowpass_hz = float(min(1.5, max(0.05, 0.25 * trace_rate_hz)))
+
+        cfg = AlignConfig(
+            band_hz=options.band_hz,
+            pad_factor=int(options.pad_factor),
+            height_range_m=options.height_range_m,
+            height_step_m=float(options.height_step_m),
+            background=background,          # type: ignore[arg-type]
+            smoothing=options.smoothing,    # type: ignore[arg-type]
+            lowpass_hz=lowpass_hz,
+        )
+        return cfg, refs
+
+    def _full(
+        self,
+        cfg: AlignConfig,
+        refs: dict[str, NDArray[np.complexfloating]] | None,
+        fb: NDArray[np.floating],
+        spectra: NDArray[np.complexfloating],
+        times: NDArray[np.floating],
+        options: AlignOptions,
+    ) -> AlignmentState | None:
+        """Estimate every cycle from scratch."""
+        baselines = None if options.baseline_m is None else {self.channel: float(options.baseline_m)}
+
+        try:
+            res = estimate_surface(
+                {self.channel: spectra}, fb, times, cfg,
+                baselines=baselines, references=refs,
+            )
+        except Exception as exc:  # noqa: BLE001
+            logger.error("Surface estimation failed", error=repr(exc))
+            return None
+
+        baseline_m = float(res.baselines_m[self.channel])
+        height_ref = (
+            float(options.height_ref_m)
+            if options.height_ref_m
+            else float(np.median(res.height_m))
+        )
+
+        residual: float | None = None
+        spread: float | None = None
+        if options.geometry_check:
+            residual, spread = self._geometry_check(
+                cfg, fb, spectra, res.baselines_m, res.height_m,
+            )
+
+        logger.info(
+            "Surface alignment estimated",
+            cycles=int(res.height_m.size),
+            baseline_cm=round(baseline_m * 100.0, 2),
+            height_cm=round(float(np.median(res.height_m)) * 100.0, 2),
+            height_ref_cm=round(height_ref * 100.0, 2),
+            rejected_fraction=round(res.quality.get("rejected_fraction", 0.0), 3),
+            confidence_median=round(res.quality.get("confidence_median", 0.0), 2),
+            geometry_residual_mm=None if residual is None else round(residual * 1000.0, 1),
+        )
+
+        return AlignmentState(
+            heights_m=np.asarray(res.height_m, dtype=float),
+            raw_heights_m=np.asarray(res.height_raw_m, dtype=float),
+            confidence=np.asarray(res.confidence, dtype=float),
+            baseline_m=baseline_m,
+            height_ref_m=height_ref,
+            quality=dict(res.quality),
+            geometry_residual_m=residual,
+            geometry_spread_m=spread,
+        )
+
+    def _extend(
+        self,
+        cfg: AlignConfig,
+        refs: dict[str, NDArray[np.complexfloating]] | None,
+        fb: NDArray[np.floating],
+        window: NDArray[np.complexfloating],
+        times: NDArray[np.floating],
+        n_cycles: int,
+    ) -> AlignmentState | None:
+        """Estimate only the new cycles from a trailing window.
+
+        The baseline and the reference height stay frozen -- they define the
+        common target every already-drawn column was shifted to. Smoothing is
+        forced to the causal Kalman: inside a trailing window only the past is
+        available, and a zero-phase low-pass would distort exactly the newest
+        samples we are after.
+        """
+        state = self._state
+        assert state is not None
+        new_count = n_cycles - state.covered
+
+        try:
+            res = estimate_surface(
+                {self.channel: window}, fb, times, replace(cfg, smoothing="live"),
+                baselines={self.channel: state.baseline_m}, references=refs,
+            )
+        except Exception as exc:  # noqa: BLE001
+            logger.error("Surface estimation failed for trailing window", error=repr(exc))
+            return None
+
+        updated = replace(
+            state,
+            heights_m=np.concatenate([state.heights_m, res.height_m[-new_count:]]),
+            raw_heights_m=np.concatenate([state.raw_heights_m, res.height_raw_m[-new_count:]]),
+            confidence=np.concatenate([state.confidence, res.confidence[-new_count:]]),
+            quality=dict(res.quality),
+            incremental=True,
+        )
+        self._state = updated
+        logger.debug(
+            "Surface alignment extended",
+            new_cycles=new_count,
+            covered=updated.covered,
+            height_cm=round(float(updated.heights_m[-1]) * 100.0, 2),
+        )
+        return updated
+
+    def _geometry_check(
+        self,
+        cfg: AlignConfig,
+        fb: NDArray[np.floating],
+        spectra: NDArray[np.complexfloating],
+        baselines: dict[str, float],
+        height_m: NDArray[np.floating],
+    ) -> tuple[float | None, float | None]:
+        """Re-derive the height from each channel's own pick and compare.
+
+        With several channels the spread of the per-channel means is the metric
+        that catches a confident mistracking. With a single channel there is no
+        spread to look at, so the useful number is the residual between the
+        joint height and what that channel alone implies: it grows when the
+        tracked peak is not where the geometry says the surface should be.
+        """
+        # check_geometry subtracts a background itself and is not handed the
+        # reference trace, so a "reference" config would make it raise.
+        probe_cfg = cfg if cfg.background != "reference" else replace(cfg, background="global_median")
+        try:
+            per_channel = check_geometry({self.channel: spectra}, fb, probe_cfg, baselines, height_m)
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("Geometry check failed", error=repr(exc))
+            return None, None
+
+        means = [float(np.mean(h)) for h in per_channel.values()]
+        spread = float(np.max(means) - np.min(means)) if len(means) > 1 else 0.0
+        implied = per_channel[self.channel]
+        residual = float(np.sqrt(np.mean((implied - height_m) ** 2)))
+        return residual, spread
+
+    # ------------------------------------------------------------------ #
+    # Application
+    # ------------------------------------------------------------------ #
+    def shift(
+        self,
+        spectrum: NDArray[np.complexfloating],
+        freq_hz: NDArray[np.floating],
+        state: AlignmentState,
+        index: int,
+    ) -> NDArray[np.complexfloating]:
+        """Move cycle `index` so its surface lands at the reference height.
+
+        Applied over the full measured band: the shift is a linear phase ramp
+        derived from geometry, so it is an exact fractional delay and does not
+        care which sub-band the height came from.
+        """
+        if index < 0 or index >= state.covered:
+            return spectrum
+
+        aligned = apply_alignment(
+            {self.channel: np.asarray(spectrum, dtype=np.complex128)[None, :]},
+            np.asarray(freq_hz, dtype=float),
+            {self.channel: state.baseline_m},
+            np.asarray([state.heights_m[index]], dtype=float),
+            height_ref_m=state.height_ref_m,
+        )
+        return aligned[self.channel][0]
+
+    def surface_path_m(self, state: AlignmentState) -> float:
+        """Two-way path length of the aligned surface echo, in metres.
+
+        This is where the surface sits after alignment: `sqrt(L^2 + (2h)^2)`.
+        """
+        return float(np.hypot(state.baseline_m, 2.0 * state.height_ref_m))
+
+    # ------------------------------------------------------------------ #
+    # Helpers
+    # ------------------------------------------------------------------ #
+    @staticmethod
+    def _median_dt(times_s: NDArray[np.floating]) -> float:
+        if times_s.size < 2:
+            return 1.0
+        dt = float(np.median(np.diff(times_s)))
+        return dt if dt > 0.0 else 1.0
+
+    @staticmethod
+    def _sanitize_times(times_s: NDArray[np.floating]) -> NDArray[np.floating]:
+        """Return a strictly increasing time axis.
+
+        The estimator interpolates over rejected cycles and derives the trace
+        rate from these values, both of which need monotonic times. Imported
+        histories can carry missing or duplicated timestamps, so fall back to a
+        unit-spaced index in that case.
+        """
+        t = np.asarray(times_s, dtype=float)
+        if t.size == 0:
+            return t
+        if not np.all(np.isfinite(t)) or (t.size > 1 and np.any(np.diff(t) <= 0.0)):
+            return np.arange(t.size, dtype=float)
+        return t - t[0]
@@ -894,7 +894,13 @@ class BaseProcessor:
                    "timestamp": entry.get("timestamp"),
                })

-            logger.info("History imported", processor_id=self.processor_id, records=len(history_data))
+            self._trim_history()
+            logger.info(
+                "History imported",
+                processor_id=self.processor_id,
+                records=len(history_data),
+                kept=len(self._sweep_history),
+            )

    @staticmethod
    def _points_to_list(points: Any) -> list[tuple[float, float]]:
@@ -1,17 +1,37 @@
 {
-  "open_air": true,
-  "axis": "abs",
-  "cut": 0.2,
-  "max": 1.5,
-  "gain": 1.0,
-  "start_freq": 400.0,
-  "stop_freq": 3000.0,
+  "open_air": false,
+  "ach_norm_enabled": false,
+  "s11_norm_enabled": false,
+  "subtract_mean_ascan": false,
+  "hardcoded_notch_enabled": false,
+  "axis": "imag",
+  "cut": 0.0,
+  "max": 3.0,
+  "gain": 1.5,
+  "start_freq": 1730.0,
+  "stop_freq": 6000.0,
  "clear_history": false,
-  "sigma": 2.0,
-  "border_border_m": 0.3,
-  "if_normalize": true,
-  "if_draw_level": true,
+  "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.7,
+  "eps_boundary_m": 0.0,
+  "surface_align": true,
+  "align_start_freq": 300.0,
+  "align_stop_freq": 6010.0,
+  "align_height_min": 0.25,
+  "align_height_max": 0.9,
+  "align_background": "reference",
+  "align_smoothing": "offline",
+  "align_baseline_mode": "manual",
+  "align_baseline_m": 0.36,
+  "align_ref_m": 0.0,
+  "align_pad_factor": 8,
+  "align_geometry_check": false,
+  "align_show_line": false,
  "data_limit": 500,
  "y_min": -50,
  "y_max": 40,
@@ -38,6 +38,16 @@ class ProcessorWebSocketHandler:

        self.active_connections: set[WebSocket] = set()

+        # Starlette's WebSocket.send() is not safe to call concurrently from
+        # multiple tasks on the same connection (it corrupts internal ASGI
+        # protocol state, raising AssertionError and killing the connection).
+        # Broadcasts triggered by unrelated processor results (scheduled from
+        # worker threads) can otherwise race with a direct reply to a client
+        # command, especially once a payload is large enough to take a while
+        # to transmit (e.g. a big B-scan history). Serialize all sends per
+        # connection with a lock to eliminate that race.
+        self._send_locks: dict[WebSocket, asyncio.Lock] = {}
+
        # Main FastAPI/uvicorn event loop handle (set on first connection).
        self._loop: asyncio.AbstractEventLoop | None = None

@@ -61,6 +71,7 @@ class ProcessorWebSocketHandler:
            
        await websocket.accept()
        self.active_connections.add(websocket)
+        self._send_locks[websocket] = asyncio.Lock()
        logger.info("WebSocket connected", total_connections=len(self.active_connections))

        try:
@@ -77,8 +88,22 @@ class ProcessorWebSocketHandler:
        """Remove a connection and log the updated count."""
        if websocket in self.active_connections:
            self.active_connections.remove(websocket)
+            self._send_locks.pop(websocket, None)
            logger.info("WebSocket disconnected", total_connections=len(self.active_connections))

+    async def _send(self, websocket: WebSocket, message_str: str) -> None:
+        """
+        Send text on a connection, serialized against any other send on it.
+
+        All outbound traffic must go through here; see `_send_locks`.
+        """
+        lock = self._send_locks.get(websocket)
+        if lock is None:
+            # Connection already torn down.
+            return
+        async with lock:
+            await websocket.send_text(message_str)
+
    # --------------------------------------------------------------------- #
    # Inbound messages
    # --------------------------------------------------------------------- #
@@ -86,7 +111,7 @@ class ProcessorWebSocketHandler:
        """Parse and route an inbound client message."""
        # Handle ping/pong messages first (they are not JSON)
        if data == "ping":
-            await websocket.send_text("pong")
+            await self._send(websocket, "pong")
            return
        elif data == "pong":
            # Just acknowledge, no response needed
@@ -138,10 +163,16 @@ class ProcessorWebSocketHandler:
            return

        try:
-            result = self.processor_manager.recalculate_processor(processor_id, config_updates)
-            if result:
-                await websocket.send_text(json.dumps(self._result_to_message(processor_id, result)))
-            else:
+            # Note: recalculate_processor() already broadcasts the result to every
+            # connected client (including this one) via the manager's result
+            # callbacks, so we must NOT also send it here directly — two
+            # concurrent writes to the same connection (the broadcast task and
+            # this handler) can race on the ASGI transport and corrupt/close it,
+            # especially with large payloads that take a while to transmit.
+            result = await asyncio.to_thread(
+                self.processor_manager.recalculate_processor, processor_id, config_updates
+            )
+            if not result:
                await self._send_error(websocket, f"Нет результата от процессора {processor_id}")
        except Exception as exc:  # noqa: BLE001
            logger.error("Recalculation failed")
@@ -173,7 +204,7 @@ class ProcessorWebSocketHandler:
                    for r in history
                ],
            }
-            await websocket.send_text(json.dumps(response))
+            await self._send(websocket, json.dumps(response))
        except Exception as exc:  # noqa: BLE001
            logger.error("Error getting history")
            await self._send_error(websocket, f"Ошибка получения истории: {exc}")
@@ -197,10 +228,14 @@ class ProcessorWebSocketHandler:
            return

        try:
-            result = self.processor_manager.load_processor_history(processor_id, history_data, config)
-            if result:
-                await websocket.send_text(json.dumps(self._result_to_message(processor_id, result)))
-            else:
+            # See note in _handle_recalculate: load_processor_history() already
+            # broadcasts the result to all clients via result callbacks, so no
+            # explicit send here (avoids racing two concurrent writes on the
+            # same connection, which reliably crashes it for large histories).
+            result = await asyncio.to_thread(
+                self.processor_manager.load_processor_history, processor_id, history_data, config
+            )
+            if not result:
                await self._send_error(websocket, f"Нет результата от процессора {processor_id} после загрузки истории")
        except Exception as exc:  # noqa: BLE001
            logger.error("History load failed", processor_id=processor_id, error=repr(exc))
@@ -224,10 +259,12 @@ class ProcessorWebSocketHandler:
            return

        try:
-            result = self.processor_manager.append_processor_history(processor_id, history_data)
-            if result:
-                await websocket.send_text(json.dumps(self._result_to_message(processor_id, result)))
-            else:
+            # See note in _handle_recalculate: append_processor_history() already
+            # broadcasts the result to all clients via result callbacks.
+            result = await asyncio.to_thread(
+                self.processor_manager.append_processor_history, processor_id, history_data
+            )
+            if not result:
                await self._send_error(websocket, f"Нет результата от процессора {processor_id} после дополнения истории")
        except Exception as exc:  # noqa: BLE001
            logger.error("History append failed", processor_id=processor_id, error=repr(exc))
@@ -247,8 +284,8 @@ class ProcessorWebSocketHandler:
            return

        try:
-            response = self.processor_manager.build_processor_state(processor_id)
-            await websocket.send_text(json.dumps(response))
+            response = await asyncio.to_thread(self.processor_manager.build_processor_state, processor_id)
+            await self._send(websocket, json.dumps(response))
        except Exception as exc:  # noqa: BLE001
            logger.error("Error getting processor state", processor_id=processor_id, error=repr(exc))
            await self._send_error(websocket, f"Ошибка получения состояния процессора: {exc}")
@@ -293,7 +330,7 @@ class ProcessorWebSocketHandler:
                return

            # Recalculate and send updated result to all clients
-            result = processor.recalculate()
+            result = await asyncio.to_thread(processor.recalculate)
            if result:
                # Broadcast to all connected clients
                message_str = json.dumps(self._result_to_message(processor_id, result))
@@ -349,7 +386,7 @@ class ProcessorWebSocketHandler:
            if raw_data:
                payload["raw_data"] = raw_data

-            await websocket.send_text(json.dumps(payload))
+            await self._send(websocket, json.dumps(payload))
        except Exception as exc:  # noqa: BLE001
            logger.error("Error sending error message", error=repr(exc))

@@ -410,7 +447,7 @@ class ProcessorWebSocketHandler:
        disconnected: list[WebSocket] = []
        for websocket in list(self.active_connections):  # snapshot
            try:
-                await websocket.send_text(message_str)
+                await self._send(websocket, message_str)
            except Exception as exc:  # noqa: BLE001
                logger.error("Broadcast to client failed; marking for disconnect", error=repr(exc))
                disconnected.append(websocket)
@@ -363,7 +363,7 @@ class VNASettingsManager:

        return summary

-    def _wait_for_new_sweep(self, sweep_buffer, current_sweep_number: int, timeout_seconds: float = 5.0) -> SweepData | None:
+    def _wait_for_new_sweep(self, sweep_buffer, current_sweep_number: int, timeout_seconds: float = 25.0) -> SweepData | None:
        """
        Wait for a new sweep to appear in the buffer with a higher sweep number.

@@ -99,6 +99,15 @@ def main() -> None:
        port=port,
        log_level="info",
        reload=False,
+        # Default 16MB is too small for uploading large B-scan history files
+        # (hundreds of sweeps x 1000 points each easily exceeds it); the
+        # connection is silently reset with no application-level error.
+        ws_max_size=256 * 1024 * 1024,
+        # The client drives its own ping/pong (see web_ui websocket.js), and the
+        # library's keepalive task races with our large result frames on the same
+        # transport, tripping an assertion inside websockets' drain handling.
+        ws_ping_interval=None,
+        ws_ping_timeout=None,
    )


@@ -0,0 +1,12 @@
+{
+  "name": "blue_antenna_100hz",
+  "timestamp": "2025-12-04T15:17:04.559000",
+  "preset_filename": "s21_start100_stop8800_points1000_bw100hz.bin",
+  "description": "",
+  "metadata": {
+    "calibration": {
+      "calibration_name": "blue_ann_through_100hz",
+      "preset_filename": "s21_start100_stop8800_points1000_bw100hz.bin"
+    }
+  }
+}
@@ -0,0 +1,12 @@
+{
+  "name": "openair",
+  "timestamp": "2025-12-03T18:01:54.377387",
+  "preset_filename": "s21_start100_stop8800_points1000_bw100hz.bin",
+  "description": "",
+  "metadata": {
+    "calibration": {
+      "calibration_name": "T_N2_451U_x2_noamp_20251105",
+      "preset_filename": "s21_start100_stop8800_points1000_bw100hz.bin"
+    }
+  }
+}
@@ -0,0 +1,12 @@
+{
+  "name": "11-21-b",
+  "timestamp": "2025-11-21T14:35:28.246450",
+  "preset_filename": "s21_start100_stop8800_points1000_bw1khz.bin",
+  "description": "",
+  "metadata": {
+    "calibration": {
+      "calibration_name": "T_N2_noamp_451U_x2",
+      "preset_filename": "s21_start100_stop8800_points1000_bw1khz.bin"
+    }
+  }
+}
@@ -0,0 +1,12 @@
+{
+  "name": "2025-11-12_b",
+  "timestamp": "2025-11-12T18:10:13.091024",
+  "preset_filename": "s21_start100_stop8800_points1000_bw1khz.bin",
+  "description": "",
+  "metadata": {
+    "calibration": {
+      "calibration_name": "T_N2_noamp_451U_x2",
+      "preset_filename": "s21_start100_stop8800_points1000_bw1khz.bin"
+    }
+  }
+}
@@ -0,0 +1,12 @@
+{
+  "name": "2026-01-30_floor",
+  "timestamp": "2026-01-30T16:56:51.990616",
+  "preset_filename": "s21_start100_stop8800_points1000_bw1khz.bin",
+  "description": "",
+  "metadata": {
+    "calibration": {
+      "calibration_name": "T_N2_451U_x2_noamp_20251105",
+      "preset_filename": "s21_start100_stop8800_points1000_bw1khz.bin"
+    }
+  }
+}
@@ -0,0 +1,12 @@
+{
+  "name": "25-11-13",
+  "timestamp": "2025-11-13T16:30:36.164757",
+  "preset_filename": "s21_start100_stop8800_points1000_bw1khz.bin",
+  "description": "",
+  "metadata": {
+    "calibration": {
+      "calibration_name": "T_N2_noamp_451U_x2",
+      "preset_filename": "s21_start100_stop8800_points1000_bw1khz.bin"
+    }
+  }
+}
@@ -0,0 +1,12 @@
+{
+  "name": "25-11-14",
+  "timestamp": "2025-11-14T12:41:59.918674",
+  "preset_filename": "s21_start100_stop8800_points1000_bw1khz.bin",
+  "description": "",
+  "metadata": {
+    "calibration": {
+      "calibration_name": "T_N2_noamp_451U_x2",
+      "preset_filename": "s21_start100_stop8800_points1000_bw1khz.bin"
+    }
+  }
+}
@@ -0,0 +1,12 @@
+{
+  "name": "blue_antenna_1khz",
+  "timestamp": "2025-12-04T15:16:19.336172",
+  "preset_filename": "s21_start100_stop8800_points1000_bw1khz.bin",
+  "description": "",
+  "metadata": {
+    "calibration": {
+      "calibration_name": "blue_ann_through_1khz",
+      "preset_filename": "s21_start100_stop8800_points1000_bw1khz.bin"
+    }
+  }
+}
@@ -0,0 +1,12 @@
+{
+  "name": "blue_n2",
+  "timestamp": "2026-01-29T18:36:21.755535",
+  "preset_filename": "s21_start100_stop8800_points1000_bw1khz.bin",
+  "description": "",
+  "metadata": {
+    "calibration": {
+      "calibration_name": "T_N2_451U_x2_noamp_20251105",
+      "preset_filename": "s21_start100_stop8800_points1000_bw1khz.bin"
+    }
+  }
+}
@@ -0,0 +1,12 @@
+{
+  "name": "green_ann_1khz",
+  "timestamp": "2025-12-05T20:43:59.532520",
+  "preset_filename": "s21_start100_stop8800_points1000_bw1khz.bin",
+  "description": "",
+  "metadata": {
+    "calibration": {
+      "calibration_name": "blue_ann_through_1khz",
+      "preset_filename": "s21_start100_stop8800_points1000_bw1khz.bin"
+    }
+  }
+}
@@ -0,0 +1,12 @@
+{
+  "name": "green_open_air_after-fix_wide-config",
+  "timestamp": "2025-12-15T13:43:49.637105",
+  "preset_filename": "s21_start100_stop8800_points1000_bw1khz.bin",
+  "description": "",
+  "metadata": {
+    "calibration": {
+      "calibration_name": "thought-25-12-15",
+      "preset_filename": "s21_start100_stop8800_points1000_bw1khz.bin"
+    }
+  }
+}
@@ -0,0 +1,12 @@
+{
+  "name": "long_cable_green_ann",
+  "timestamp": "2025-12-19T17:09:30.725124",
+  "preset_filename": "s21_start100_stop8800_points1000_bw1khz.bin",
+  "description": "",
+  "metadata": {
+    "calibration": {
+      "calibration_name": "T_N2_tair_451U_x2+rfc18-1m",
+      "preset_filename": "s21_start100_stop8800_points1000_bw1khz.bin"
+    }
+  }
+}
@@ -0,0 +1,12 @@
+{
+  "name": "open_air_green_213",
+  "timestamp": "2025-12-12T17:55:58.564470",
+  "preset_filename": "s21_start100_stop8800_points1000_bw1khz.bin",
+  "description": "",
+  "metadata": {
+    "calibration": {
+      "calibration_name": "blue_ann_through_1khz",
+      "preset_filename": "s21_start100_stop8800_points1000_bw1khz.bin"
+    }
+  }
+}
@@ -0,0 +1,12 @@
+{
+  "name": "blue_antenna_20hz",
+  "timestamp": "2025-12-05T19:24:34.272799",
+  "preset_filename": "s21_start400_stop6000_points1000_bw20hz.bin",
+  "description": "",
+  "metadata": {
+    "calibration": {
+      "calibration_name": "blue_through_20hz",
+      "preset_filename": "s21_start400_stop6000_points1000_bw20hz.bin"
+    }
+  }
+}
@@ -0,0 +1,12 @@
+{
+  "name": "opair",
+  "timestamp": "2025-12-03T18:29:32.724722",
+  "preset_filename": "s21_start400_stop6000_points1000_bw20hz.bin",
+  "description": "",
+  "metadata": {
+    "calibration": {
+      "calibration_name": "T_N2_451U_x2_noamp_20251105",
+      "preset_filename": "s21_start400_stop6000_points1000_bw20hz.bin"
+    }
+  }
+}
Author	SHA1	Message	Date
BogatskiyG	39b2ff99d5	surface_align added	2026-08-05 19:01:26 +03:00
BogatskiyG	06836fa244	optimization fix	2026-08-05 17:58:26 +03:00
BogatskiyG	8a419302d5	optimization fix	2026-08-05 15:40:41 +03:00
sad-santa	63138d8d9c	added notch toggle	2026-04-30 14:23:15 +03:00
sad-santa	6c1a1151bc	hardcoded filters	2026-04-30 13:59:21 +03:00
sad-santa	4da9e9711a	del one dot	2026-04-22 13:12:37 +03:00
morlo	7192577e3a	add ...	2026-02-17 18:22:06 +03:00
sad-santa	4897209773	added s21 s11 normalisations	2026-02-17 17:29:24 +03:00
sad-santa	80335a9983	checkpoint before ach normalization feature	2026-02-12 13:33:44 +03:00
sad-santa	da799aed29	added scan averaging and consideration of the epsilon environmen	2026-02-03 12:01:04 +03:00
mipt-user	14bd16ca9d	new presets added	2025-12-03 18:41:36 +03:00
mipt-user	3124b7141a	Merge branch 'master' of 10.55.228.171:/home/mipt-user/projects/vna_system	2025-11-26 13:52:11 +03:00
sad-santa	c1f5d6580e	added requirement	2025-11-21 19:54:05 +03:00