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

Added normalization and detection.

Browse Source
This commit is contained in:
ivngrmk
2025-11-19 17:47:31 +03:00
parent 333ec5d196
commit 2b2e323fbf
3 changed files with 203 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
vna_system/core/processors/configs/bscan_config.json
View File

@ -1,12 +1,16 @@
{
"open_air": true,
"axis": "abs",
"cut": 0.267,
"cut": 0.23,
"max": 1.5,
"gain": 1.0,
"start_freq": 470.0,
"stop_freq": 8800.0,
"clear_history": false,
"sigma": 4.28,
"sigma": 3.42,
"border_border_m": 0.26,
"if_normalize": false,
"if_draw_level": false,
"detection_level": 3.0,
"data_limit": 500
}

2
vna_system/core/processors/configs/magnitude_config.json
View File

@ -3,6 +3,6 @@
"y_max": 40,
"autoscale": true,
"show_magnitude": true,
"show_phase": true,
"show_phase": false,
"open_air": false
}

310
vna_system/core/processors/implementations/bscan_processor.py
View File

@ -58,6 +58,10 @@ class BScanProcessor(BaseProcessor):
"stop_freq": 8800.0, # Stop frequency (MHz)
"clear_history": False, # UI button; not persisted
"sigma" : 0.01,
"border_border_m" : 0.5,
"if_normalize" : False,
"if_draw_level" : False,
"detection_level" : 5,
}
def get_ui_parameters(self) -> list[UIParameter]:
@ -120,6 +124,43 @@ class BScanProcessor(BaseProcessor):
value=cfg["stop_freq"],
options={"min": 100.0, "max": 8800.0, "step": 10.0, "dtype": "float"},
),
# Modern features
UIParameter(
name="sigma",
label="Степень сглаживания в abs режиме",
type="slider",
value=0.01,
options={"min": 0.01, "max": 5.0, "step": 0.01, "dtype": "float"},
),
UIParameter(
name="if_normalize",
label="Нормировка",
type="toggle",
value=False,
),
UIParameter(
name="border_border_m",
label="Глубина границы",
type="slider",
value=0.5,
options={"min": 0.0, "max": 2.5, "step": 0.01, "dtype": "float"},
),
UIParameter(
name="if_draw_level",
label="Детекция",
type="toggle",
value=False,
),
UIParameter(
name="detection_level",
label="Порог детекции,%",
type="slider",
value=5,
options={"min": 0.0, "max": 100.0, "step": 0.1, "dtype": "float"},
),
# Big botton
UIParameter(
name="clear_history",
label="Очистить историю",
@ -127,13 +168,6 @@ class BScanProcessor(BaseProcessor):
value=False,
options={"action": "Очистить накопленную историю графика"},
),
UIParameter(
name="sigma",
label="Степень сглаживания в abs режиме",
type="slider",
value=self._config.get("y_max", 10),
options={"min": 0.01, "max": 5.0, "step": 0.01, "dtype": "float"},
),
]
def update_config(self, updates: dict[str, Any]) -> None:
@ -300,112 +334,167 @@ class BScanProcessor(BaseProcessor):
# -------------------------------------------------------------------------
def generate_plotly_config(
self,
processed_data: dict[str, Any],
vna_config: dict[str, Any], # noqa: ARG002 - reserved for future layout tweaks
) -> dict[str, Any]:
"""
Produce a Plotly-compatible heatmap configuration from accumulated sweeps.
"""
if "error" in processed_data:
return {
"data": [],
"layout": {
"title": "B-Scan анализ - Ошибка",
"annotations": [
{
"text": f"Ошибка: {processed_data['error']}",
"x": 0.5,
"y": 0.5,
"xref": "paper",
"yref": "paper",
"showarrow": False,
}
],
"template": "plotly_dark",
},
self,
processed_data: dict[str, Any],
vna_config: dict[str, Any], # noqa: ARG002 - reserved for future layout tweaks
) -> dict[str, Any]:
"""
Produce a Plotly-compatible heatmap configuration from accumulated sweeps.
"""
if "error" in processed_data:
return {
"data": [],
"layout": {
"title": "B-Scan анализ - Ошибка",
"annotations": [
{
"text": f"Ошибка: {processed_data['error']}",
"x": 0.5,
"y": 0.5,
"xref": "paper",
"yref": "paper",
"showarrow": False,
}
],
"template": "plotly_dark",
},
}
with self._lock:
history = list(self._plot_history)
if not history:
return {
"data": [],
"layout": {
"title": "B-Scan анализ - Нет данных",
"xaxis": {"title": "Номер развертки"},
"yaxis": {"title": "Глубина (м)"},
"template": "plotly_dark",
},
}
Y_VALUE = self._config["border_border_m"]
# Build scatter-like heatmap (irregular grid) from history
x_coords: list[int] = []
y_coords: list[float] = []
z_values: list[float] = []
z_values_square = np.zeros((len(history[0]["distance_data"]),len(history)),dtype=float)
for sweep_index, item in enumerate(history, start=1):
depths = item["distance_data"]
amps = item["time_domain_data"]
if self._config['if_normalize']:
depth_mask = np.array(depths) < Y_VALUE
normalized_ampls = np.array(amps) / np.max(np.array(amps)[depth_mask])
else:
normalized_ampls = np.array(amps)
z_values_square[:,sweep_index-1] = normalized_ampls
for d, a in zip(depths, normalized_ampls, strict=False):
x_coords.append(sweep_index)
y_coords.append(d)
z_values.append(float(a))
if self._config["if_draw_level"]:
detection_level_abs = np.percentile(z_values, 100 - self._config["detection_level"])
detected_points_mask = (np.array(z_values) > detection_level_abs) & (np.array(y_coords) > Y_VALUE)
detected_points_x = list([float(x) for x in np.array(x_coords)[detected_points_mask]])
detected_points_y = list([float(y) for y in np.array(y_coords)[detected_points_mask]])
detected_trace = {
"type": "scatter",
"mode": "markers",
"x": detected_points_x,
"y": detected_points_y,
"marker": {
"size": 8,
"color": "red",
"symbol": "circle",
},
"name": "Обнаруженные точки",
}
else:
detected_trace = None
# Colorscale selection
if self._config["axis"] == "abs":
colorscale = "Viridis"
heatmap_kwargs: dict[str, Any] = {}
else:
colorscale = "RdBu"
heatmap_kwargs = {"zmid": 0}
heatmap_trace = {
"type": "heatmap",
"x": x_coords,
"y": y_coords,
"z": z_values,
"colorscale": colorscale,
"colorbar": {"title": "Амплитуда"},
"hovertemplate": (
"Развертка: %{x}<br>"
"Глубина: %{y:.3f} м<br>"
"Амплитуда: %{z:.3f}<br>"
"<extra></extra>"
),
**heatmap_kwargs,
}
with self._lock:
history = list(self._plot_history)
freq_start, freq_stop = processed_data.get("frequency_range", [0.0, 0.0])
config_info = (
f"Частота: {freq_start/1e6:.1f}-{freq_stop/1e6:.1f} МГц | "
f"Усиление: {self._config['gain']:.1f} | "
f"Отсечка: {self._config['cut']:.3f} м | "
f"Макс глубина: {self._config['max']:.1f} м | "
f"Ось: {self._config['axis']} | "
f"Разверток: {len(history)}"
)
if not history:
return {
"data": [],
"layout": {
"title": "B-Scan анализ - Нет данных",
"xaxis": {"title": "Номер развертки"},
"yaxis": {"title": "Глубина (м)"},
"template": "plotly_dark",
},
if processed_data.get("reference_used", False):
config_info += " | Открытый воздух: ВКЛ"
# if self._config["data_limitation"]:
# config_info += f" | Limit: {self._config['data_limitation']}"
layout = {
"title": f"B-Scan тепловая карта - {config_info}",
"xaxis": {"title": "Номер развертки", "side": "bottom"},
"yaxis": {"title": "Глубина (м)", "autorange": "reversed"},
"hovermode": "closest",
"height": 546,
"template": "plotly_dark",
"margin": {"t": 40, "r": 50, "b": 110, "l": 50},
"autosize": True,
}
# Build scatter-like heatmap (irregular grid) from history
x_coords: list[int] = []
y_coords: list[float] = []
z_values: list[float] = []
print(self._config['if_normalize'], self._config['if_draw_level'])
if self._config['if_normalize'] or self._config['if_draw_level']:
layout["shapes"] = layout.get("shapes", []) + [
{
"type": "line",
# по X — координаты "бумаги" (0–1 по всей ширине графика)
"xref": "paper",
# по Y — координаты данных (в метрах глубины)
"yref": "y",
"x0": 0,
"x1": 1,
"y0": Y_VALUE,
"y1": Y_VALUE,
"line": {
"width": 2,
"dash": "dash",
"color": "white",
},
}
]
for sweep_index, item in enumerate(history, start=1):
depths = item["distance_data"]
amps = item["time_domain_data"]
for d, a in zip(depths, amps, strict=False):
x_coords.append(sweep_index)
y_coords.append(d)
z_values.append(a)
# Colorscale selection
if self._config["axis"] == "abs":
colorscale = "Viridis"
heatmap_kwargs: dict[str, Any] = {}
else:
colorscale = "RdBu"
heatmap_kwargs = {"zmid": 0}
heatmap_trace = {
"type": "heatmap",
"x": x_coords,
"y": y_coords,
"z": z_values,
"colorscale": colorscale,
"colorbar": {"title": "Амплитуда"},
"hovertemplate": (
"Развертка: %{x}<br>"
"Глубина: %{y:.3f} м<br>"
"Амплитуда: %{z:.3f}<br>"
"<extra></extra>"
),
**heatmap_kwargs,
}
freq_start, freq_stop = processed_data.get("frequency_range", [0.0, 0.0])
config_info = (
f"Частота: {freq_start/1e6:.1f}-{freq_stop/1e6:.1f} МГц | "
f"Усиление: {self._config['gain']:.1f} | "
f"Отсечка: {self._config['cut']:.3f} м | "
f"Макс глубина: {self._config['max']:.1f} м | "
f"Ось: {self._config['axis']} | "
f"Разверток: {len(history)}"
)
if processed_data.get("reference_used", False):
config_info += " | Открытый воздух: ВКЛ"
# if self._config["data_limitation"]:
# config_info += f" | Limit: {self._config['data_limitation']}"
layout = {
"title": f"B-Scan тепловая карта - {config_info}",
"xaxis": {"title": "Номер развертки", "side": "bottom"},
"yaxis": {"title": "Глубина (м)", "autorange": "reversed"},
"hovermode": "closest",
"height": 546,
"template": "plotly_dark",
"margin": {"t": 40, "r": 50, "b": 110, "l": 50},
"autosize": True
}
return {"data": [heatmap_trace], "layout": layout}
if detected_trace is not None:
return {"data": [heatmap_trace,detected_trace], "layout": layout}
return {"data": [heatmap_trace], "layout": layout}
# -------------------------------------------------------------------------
# Recalculation override
@ -568,13 +657,8 @@ class BScanProcessor(BaseProcessor):
# Determine sigma for smoothing
if vna_config:
# calc_type = vna_config.get("axis", "abs")
# sigma = float(vna_config.get("sigma", 0.01))
calc_type = self._config["axis"]
sigma = self._config["sigma"]
pass
else:
calc_type = self._config["axis"]
sigma = self._config["sigma"]
@ -594,10 +678,8 @@ class BScanProcessor(BaseProcessor):
depth_out, time_out = self._apply_depth_processing(depth_m, time_response)
if calc_type == 'abs':
print('Filtering with sigma',sigma)
filtered_time_out = gaussian_filter1d(time_out, sigma=sigma)
else:
print("Not filtering")
filtered_time_out = time_out
return {