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

added scan averaging and consideration of the epsilon environmen

Browse Source
This commit is contained in:
sad-santa
2026-02-03 12:01:04 +03:00
parent 14bd16ca9d
commit da799aed29
3 changed files with 150 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
vna_system/core/config.py
+1 -1
View File
@@ -35,7 +35,7 @@ VNA_PID = 0x5740 # STM32 Virtual ComPort
# -----------------------------------------------------------------------------
# Simulator mode settings
# -----------------------------------------------------------------------------
USE_SIMULATOR = False # Set to True to use simulator instead of real device
USE_SIMULATOR = True # Set to True to use simulator instead of real device
SIMULATOR_SWEEP_FILE = BASE_DIR / "binary_input" / "sweep_example" / "example.json"
SIMULATOR_NOISE_LEVEL = 100 # Standard deviation of Gaussian noise to add to real and imaginary parts
vna_system/core/processors/configs/bscan_config.json
+9 -5
View File
@@ -1,17 +1,21 @@
{
"open_air": true,
"open_air": false,
"subtract_mean_ascan": false,
"axis": "abs",
"cut": 0.2,
"max": 1.5,
"cut": 0.546,
"max": 0.5,
"gain": 1.0,
"start_freq": 400.0,
"stop_freq": 5170.0,
"clear_history": false,
"sigma": 0.54,
"border_border_m": 0.3,
"if_normalize": false,
"border_border_m": 0.09,
"if_normalize": true,
"if_draw_level": false,
"detection_level": 8.0,
"apply_eps_correction": true,
"eps_r": 4.4,
"eps_boundary_m": 0.12,
"data_limit": 500,
"y_min": -50,
"y_max": 40,
vna_system/core/processors/implementations/bscan_processor.py
+140 -4
View File
@@ -49,6 +49,7 @@ class BScanProcessor(BaseProcessor):
"""Return default configuration values."""
return {
"open_air": False, # Toggle for reference usage
"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"
"cut": 0.824, # Cut parameter (meters)
@@ -62,6 +63,9 @@ class BScanProcessor(BaseProcessor):
"if_normalize" : False,
"if_draw_level" : False,
"detection_level" : 5,
"apply_eps_correction": False,
"eps_r": 4.0,
"eps_boundary_m": 0.5,
}
def get_ui_parameters(self) -> list[UIParameter]:
@@ -75,6 +79,12 @@ class BScanProcessor(BaseProcessor):
type="toggle",
value=cfg["open_air"],
),
UIParameter(
name="subtract_mean_ascan",
label="Вычесть средний A-скан",
type="toggle",
value=cfg["subtract_mean_ascan"],
),
UIParameter(
name="axis",
label="Ось",
@@ -168,6 +178,27 @@ class BScanProcessor(BaseProcessor):
value=cfg["clear_history"],
options={"action": "Очистить накопленную историю графика"},
),
# --- NEW: epsilon correction controls ---
UIParameter(
name="apply_eps_correction",
label="Учет ε ниже границы (новая)",
type="toggle",
value=cfg["apply_eps_correction"],
),
UIParameter(
name="eps_r",
label="εr ниже границы (новая)",
type="slider",
value=cfg["eps_r"],
options={"min": 1.0, "max": 30.0, "step": 0.1, "dtype": "float"},
),
UIParameter(
name="eps_boundary_m",
label="Граница среды (ε) (м) (новая)",
type="slider",
value=cfg["eps_boundary_m"],
options={"min": 0.0, "max": 2.5, "step": 0.01, "dtype": "float"},
),
]
def update_config(self, updates: dict[str, Any]) -> None:
@@ -310,8 +341,11 @@ class BScanProcessor(BaseProcessor):
all_sweep_numbers = list(range(1, len(self._plot_history) + 1))
all_timestamps = [record["timestamp"] for record in self._plot_history]
adjusted_time_domain = self._apply_mean_ascan_subtraction(all_time_domain)
latest_time_domain = adjusted_time_domain[-1] if adjusted_time_domain else analysis["time_data"].tolist()
return {
"time_domain_data": analysis["time_data"].tolist(), # Latest sweep
"time_domain_data": latest_time_domain, # Latest sweep (after mean subtraction if enabled)
"distance_data": analysis["distance"].tolist(), # Latest sweep
"frequency_range": analysis["freq_range"],
"reference_used": bool(self._config["open_air"] and reference_data is not None),
@@ -319,7 +353,7 @@ class BScanProcessor(BaseProcessor):
"points_processed": int(complex_data.size),
"plot_history_count": len(self._plot_history),
# Full history data
"all_time_domain_data": all_time_domain,
"all_time_domain_data": adjusted_time_domain,
"all_distance_data": all_distance,
"all_sweep_numbers": all_sweep_numbers,
"all_timestamps": all_timestamps,
@@ -383,9 +417,12 @@ class BScanProcessor(BaseProcessor):
z_values: list[float] = []
z_values_square = np.zeros((len(history[0]["distance_data"]),len(history)),dtype=float)
time_series = [record["time_domain_data"] for record in history]
adjusted_time_series = self._apply_mean_ascan_subtraction(time_series)
for sweep_index, item in enumerate(history, start=1):
depths = item["distance_data"]
amps = item["time_domain_data"]
amps = adjusted_time_series[sweep_index - 1] if sweep_index - 1 < len(adjusted_time_series) else item["time_domain_data"]
if self._config['if_normalize']:
depth_mask = np.array(depths) < Y_VALUE
@@ -492,6 +529,25 @@ class BScanProcessor(BaseProcessor):
}
]
if self._config.get("apply_eps_correction", False):
eps_boundary = float(self._config.get("eps_boundary_m", 0.0))
layout["shapes"] = layout.get("shapes", []) + [
{
"type": "line",
"xref": "paper",
"yref": "y",
"x0": 0,
"x1": 1,
"y0": eps_boundary,
"y1": eps_boundary,
"line": {
"width": 2,
"dash": "dot",
"color": "#00E5FF",
},
}
]
if detected_trace is not None:
return {"data": [heatmap_trace,detected_trace], "layout": layout}
return {"data": [heatmap_trace], "layout": layout}
@@ -786,6 +842,59 @@ class BScanProcessor(BaseProcessor):
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_eps_depth_correction(
self,
depth_out: NDArray[np.floating],
) -> NDArray[np.floating]:
"""
NEW FEATURE:
Compress depth axis below eps_boundary_m using epsilon.
Inputs
------
depth_out:
One-way depth axis (meters). This is what you later plot on Y.
Config keys used
---------------
apply_eps_correction: bool
If False -> do nothing.
eps_r: float
Relative permittivity below boundary. Compression factor = 1/sqrt(eps_r).
eps_boundary_m: float
Depth (m) where the medium changes.
Mapping
-------
d0 = eps_boundary_m
if d <= d0: unchanged
if d > d0: d_corr = d0 + (d - d0)/sqrt(eps_r)
"""
# 1) Check toggle
if not self._config.get("apply_eps_correction", False):
return depth_out
# 2) Read eps_r
eps_r = float(self._config.get("eps_r", 1.0))
if eps_r <= 1.0:
return depth_out
# 3) Read epsilon boundary depth
d0 = float(self._config.get("eps_boundary_m", 0.0))
# 4) Boundary at/under 0 -> compress everything
if d0 <= 0.0:
return depth_out / np.sqrt(eps_r)
# 5) Copy to avoid mutating input array
depth_corr = depth_out.astype(float, copy=True)
# 6) Apply only below boundary
mask = depth_corr > d0
depth_corr[mask] = d0 + (depth_corr[mask] - d0) / np.sqrt(eps_r)
return depth_corr
def _apply_depth_processing(
self,
depth_m: NDArray[np.floating],
@@ -824,6 +933,8 @@ class BScanProcessor(BaseProcessor):
# Convert to one-way depth relative to the cut
depth_out = (depth_win - lo) / 2.0
depth_out = self._apply_eps_depth_correction(depth_out)
# Depth-dependent gain (safe for zero depth with exponent >= 0)
with np.errstate(invalid="ignore"):
@@ -838,6 +949,31 @@ class BScanProcessor(BaseProcessor):
logger.error("Depth processing failed", error=repr(exc))
return depth_m, response
def _apply_mean_ascan_subtraction(
self,
time_domain_series: list[list[float]],
) -> list[list[float]]:
"""
Subtract mean A-scan (across sweeps) from each A-scan in the B-scan.
This operates on time-domain data after reference subtraction and IFFT.
"""
if not self._config.get("subtract_mean_ascan", False):
return time_domain_series
lengths = {len(series) for series in time_domain_series}
if len(lengths) != 1:
logger.warning(
"Mean A-scan subtraction skipped due to inconsistent lengths",
lengths=sorted(lengths),
)
return time_domain_series
data = np.asarray(time_domain_series, dtype=float)
mean_trace = np.mean(data, axis=0)
adjusted = data - mean_trace
return [row.tolist() for row in adjusted]
# -------------------------------------------------------------------------
# State export override
# -------------------------------------------------------------------------
@@ -862,7 +998,7 @@ class BScanProcessor(BaseProcessor):
all_timestamps = [record["timestamp"] for record in self._plot_history]
state["plot_data"] = {
"all_time_domain_data": all_time_domain,
"all_time_domain_data": self._apply_mean_ascan_subtraction(all_time_domain),
"all_distance_data": all_distance,
"all_sweep_numbers": all_sweep_numbers,
"all_timestamps": all_timestamps,