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

some fixes

Browse Source
This commit is contained in:
Ayzen
2025-09-29 20:18:03 +03:00
parent f4e223ca96
commit de094eeca7
14 changed files with 272 additions and 809 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
vna_system/binary_input/current_input.bin
View File

@ -1 +1 @@
config_inputs/s11_start100_stop8800_points1000_bw1khz.bin
config_inputs/s21_start100_stop8800_points1000_bw1khz.bin

9
vna_system/core/config.py
View File

@ -14,7 +14,7 @@ API_PORT = 8000
# -----------------------------------------------------------------------------
# Logging settings (используются из main)
# -----------------------------------------------------------------------------
LOG_LEVEL = "INFO" # {"DEBUG","INFO","WARNING","ERROR","CRITICAL"}
LOG_LEVEL = "DEBUG" # {"DEBUG","INFO","WARNING","ERROR","CRITICAL"}
LOG_DIR = BASE_DIR / "logs" # Directory for application logs
LOG_APP_FILE = LOG_DIR / "vna_system.log" # Main application log file
@ -73,4 +73,9 @@ SERIAL_DRAIN_DELAY = 0.05
SERIAL_DRAIN_CHECK_DELAY = 0.01
SERIAL_CONNECT_DELAY = 0.01
PROCESSORS_CONFIG_DIR_PATH = "vna_system/core/processors/configs"
PROCESSORS_CONFIG_DIR_PATH = "vna_system/core/processors/configs"
# -----------------------------------------------------------------------------
# Physical constants
# -----------------------------------------------------------------------------
SPEED_OF_LIGHT_M_S = 299_700_000.0 # Speed of light in meters per second

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

@ -1,9 +1,9 @@
{
"open_air": true,
"axis": "real",
"data_limitation": null,
"cut": 0.824,
"max": 1.1,
"axis": "abs",
"data_limitation": "ph_only_1",
"cut": 0.945,
"max": 1.4,
"gain": 1.2,
"start_freq": 100.0,
"stop_freq": 8800.0,

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

@ -1,10 +1,5 @@
{
"y_min": -90,
"y_min": -55,
"y_max": 20,
"smoothing_enabled": true,
"smoothing_window": 17,
"marker_enabled": true,
"marker_frequency": 300000009.0,
"grid_enabled": true,
"reset_smoothing": false
"show_phase": true
}

5
vna_system/core/processors/implementations/__init__.py
View File

@ -1,5 +0,0 @@
from .magnitude_processor import MagnitudeProcessor
from .phase_processor import PhaseProcessor
from .smith_chart_processor import SmithChartProcessor
__all__ = ['MagnitudeProcessor', 'PhaseProcessor', 'SmithChartProcessor']

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

@ -1,14 +1,15 @@
from __future__ import annotations
from pathlib import Path
from typing import Any, Final
from typing import Any
import numpy as np
from numpy.typing import NDArray
from vna_system.core.logging.logger import get_component_logger
from vna_system.core.processors.base_processor import BaseProcessor, UIParameter
from vna_system.core.processors.base_processor import BaseProcessor, UIParameter, ProcessedResult
from vna_system.core.acquisition.sweep_buffer import SweepData
from vna_system.core.config import SPEED_OF_LIGHT_M_S
logger = get_component_logger(__file__)
@ -26,18 +27,15 @@ class BScanProcessor(BaseProcessor):
- Plot history accumulation for multi-sweep heatmaps
"""
# Physical and display-related constants
SPEED_OF_LIGHT_M_S: Final[float] = 299_700_000.0 # m/s
def __init__(self, config_dir: Path) -> None:
super().__init__("bscan", config_dir)
# Increase history size for multi-sweep plotting
self.max_history = 10
self._max_history = 50
# Local plot history (separate from sweep history maintained by BaseProcessor)
self._plot_history: list[dict[str, Any]] = []
self._max_plot_history = 50
logger.info("BScanProcessor initialized", processor_id=self.processor_id)
@ -141,7 +139,9 @@ class BScanProcessor(BaseProcessor):
def _clear_plot_history(self) -> None:
"""Clear the accumulated plot history."""
with self._lock:
self._plot_history.clear()
latest = self._sweep_history[-1]
self._sweep_history.clear()
self._sweep_history.append(latest)
logger.info("Plot history cleared", processor_id=self.processor_id)
# -------------------------------------------------------------------------
@ -161,7 +161,7 @@ class BScanProcessor(BaseProcessor):
-------
dict
Keys: time_domain_data, distance_data, frequency_range, reference_used,
axis_type, data_limitation, points_processed, plot_history_count
axis_type, data_limitation, points_processed, plot_history_count
Or: {"error": "..."} on failure.
"""
try:
@ -204,8 +204,8 @@ class BScanProcessor(BaseProcessor):
with self._lock:
self._plot_history.append(plot_record)
if len(self._plot_history) > self._max_plot_history:
self._plot_history = self._plot_history[-self._max_plot_history :]
if len(self._plot_history) > self._max_history:
self._plot_history = self._plot_history[-self._max_history :]
return {
"time_domain_data": analysis["time_data"].tolist(),
@ -327,13 +327,65 @@ class BScanProcessor(BaseProcessor):
"xaxis": {"title": "Sweep Number", "side": "top"},
"yaxis": {"title": "Depth (m)", "autorange": "reversed"},
"hovermode": "closest",
"height": 600,
"width": 500,
"height": 546,
"template": "plotly_dark",
"margin": {"t": 30, "r": 50, "b": 50, "l": 50},
"autosize": True
}
return {"data": [heatmap_trace], "layout": layout}
# -------------------------------------------------------------------------
# Recalculation override
# -------------------------------------------------------------------------
def recalculate(self) -> ProcessedResult | None:
"""
Recompute all plot history using current config settings.
Unlike the base implementation that processes only the latest sweep,
this method processes all sweep history to rebuild the complete heatmap.
"""
with self._lock:
if not self._sweep_history:
logger.debug("Recalculate skipped; sweep history empty")
return None
# Clear existing plot history to rebuild from scratch
self._plot_history.clear()
# Process all sweeps in history with current config
for entry in self._sweep_history:
sweep_data = entry["sweep_data"]
calibrated_data = entry["calibrated_data"]
vna_config = entry["vna_config"]
# Use process_sweep to handle the processing logic
processed = self.process_sweep(sweep_data, calibrated_data, vna_config)
# Skip if processing failed
if "error" in processed:
continue
# Trim plot history if needed
if len(self._plot_history) > self._max_history:
self._plot_history = self._plot_history[-self._max_history:]
logger.info("Recalculated B-scan with all history",
plot_records=len(self._plot_history),
sweep_records=len(self._sweep_history))
# Return the result based on the last sweep processed
if self._sweep_history:
latest = self._sweep_history[-1]
return self._process_data(
latest["sweep_data"],
latest["calibrated_data"],
latest["vna_config"]
)
return None
# -------------------------------------------------------------------------
# Low-level helpers
# -------------------------------------------------------------------------
@ -502,7 +554,7 @@ class BScanProcessor(BaseProcessor):
y = np.fft.ifft(H)
# Convert time (s) to one-way distance (m): d = c * t
depth_m = t_sec * self.SPEED_OF_LIGHT_M_S
depth_m = t_sec * SPEED_OF_LIGHT_M_S
if axis == "abs":
y_fin = np.abs(y)

241
vna_system/core/processors/implementations/magnitude_processor.py
View File

@ -16,19 +16,16 @@ class MagnitudeProcessor(BaseProcessor):
--------
1) Derive frequency axis from VNA config (start/stop, N points).
2) Compute |S| in dB per point (20*log10(|complex|), clamped for |complex|==0).
3) Optionally smooth using moving-average (odd window).
4) Provide Plotly configuration including an optional marker.
3) Generate Plotly configuration with parameter type in legend.
Notes
-----
- `calibrated_data` is expected to be a `SweepData` with `.points: list[tuple[float,float]]`.
- Marker frequency is validated by BaseProcessor via `UIParameter(options=...)`.
- Parameter type (S11, S21, etc.) is extracted from VNA config for proper labeling.
"""
def __init__(self, config_dir: Path) -> None:
super().__init__("magnitude", config_dir)
# Internal state that can be reset via a UI "button"
self._smoothing_history: list[float] = []
# ------------------------------------------------------------------ #
# Core processing
@ -61,8 +58,8 @@ class MagnitudeProcessor(BaseProcessor):
return {"error": "Empty calibrated sweep"}
# Frequency axis from VNA config (defaults if not provided)
start_freq = float(vna_config.get("start_frequency", 100e6))
stop_freq = float(vna_config.get("stop_frequency", 8.8e9))
start_freq = float(vna_config.get("start_freq", 100e6))
stop_freq = float(vna_config.get("stop_freq", 8.8e9))
if n == 1:
freqs = [start_freq]
@ -72,84 +69,94 @@ class MagnitudeProcessor(BaseProcessor):
# Magnitude in dB (clamp zero magnitude to -120 dB)
mags_db: list[float] = []
for real, imag in points:
mag = abs(complex(real, imag))
mags_db.append(20.0 * np.log10(mag) if mag > 0.0 else -120.0)
phases_deg: list[float] = []
# Optional smoothing
if self._config.get("smoothing_enabled", False):
window = int(self._config.get("smoothing_window", 5))
mags_db = self._apply_moving_average(mags_db, window)
for real, imag in points:
complex_val = complex(real, imag)
mag = abs(complex_val)
mags_db.append(20.0 * np.log10(mag) if mag > 0.0 else -120.0)
phases_deg.append(np.degrees(np.angle(complex_val)))
result = {
"frequencies": freqs,
"magnitudes_db": mags_db,
"phases_deg": phases_deg,
"y_min": float(self._config.get("y_min", -80)),
"y_max": float(self._config.get("y_max", 10)),
"marker_enabled": bool(self._config.get("marker_enabled", True)),
"marker_frequency": float(
self._config.get("marker_frequency", freqs[len(freqs) // 2] if freqs else 1e9)
),
"grid_enabled": bool(self._config.get("grid_enabled", True)),
"show_phase": bool(self._config.get("show_phase", False)),
}
logger.debug("Magnitude sweep processed", points=n)
return result
def generate_plotly_config(self, processed_data: dict[str, Any], vna_config: dict[str, Any]) -> dict[str, Any]:
"""
Build a Plotly figure config for the magnitude trace and optional marker.
Build a Plotly figure config for the magnitude trace and optional phase.
"""
if "error" in processed_data:
return {"error": processed_data["error"]}
freqs: list[float] = processed_data["frequencies"]
mags_db: list[float] = processed_data["magnitudes_db"]
grid_enabled: bool = processed_data["grid_enabled"]
phases_deg: list[float] = processed_data["phases_deg"]
show_phase: bool = processed_data["show_phase"]
# Marker resolution
marker_freq: float = processed_data["marker_frequency"]
if freqs:
idx = min(range(len(freqs)), key=lambda i: abs(freqs[i] - marker_freq))
marker_mag = mags_db[idx]
marker_x = freqs[idx] / 1e9
marker_trace = {
"x": [marker_x],
"y": [marker_mag],
"type": "scatter",
"mode": "markers",
"name": f"Marker: {freqs[idx]/1e9:.3f} GHz, {marker_mag:.2f} dB",
"marker": {"color": "red", "size": 8, "symbol": "circle"},
}
else:
idx = 0
marker_trace = None
# Determine the parameter type from preset mode
parameter_type = vna_config["mode"].value.upper() # Convert "s11" -> "S11", "s21" -> "S21"
# Convert Hz to GHz for x-axis
freqs_ghz = [f / 1e9 for f in freqs]
traces = [
{
"x": [f / 1e9 for f in freqs], # Hz -> GHz
"x": freqs_ghz,
"y": mags_db,
"type": "scatter",
"mode": "lines",
"name": "Magnitude",
"name": f"|{parameter_type}| Magnitude",
"line": {"color": "blue", "width": 2},
"yaxis": "y",
}
]
if processed_data["marker_enabled"] and marker_trace:
traces.append(marker_trace)
# Add phase trace if enabled
if show_phase:
traces.append({
"x": freqs_ghz,
"y": phases_deg,
"type": "scatter",
"mode": "lines",
"name": f"∠{parameter_type} Phase",
"line": {"color": "red", "width": 2},
"yaxis": "y2",
})
# Layout configuration
layout = {
"title": f"{parameter_type} Response",
"xaxis": {"title": "Frequency (GHz)", "showgrid": True},
"yaxis": {
"title": "Magnitude (dB)",
"range": [processed_data["y_min"], processed_data["y_max"]],
"showgrid": True,
"side": "left",
},
"hovermode": "x unified",
"showlegend": True,
}
# Add second y-axis for phase if enabled
if show_phase:
layout["yaxis2"] = {
"title": "Phase (°)",
"overlaying": "y",
"side": "right",
"showgrid": False,
"range": [-180, 180],
}
fig = {
"data": traces,
"layout": {
"title": "Magnitude Response",
"xaxis": {"title": "Frequency (GHz)", "showgrid": grid_enabled},
"yaxis": {
"title": "Magnitude (dB)",
"range": [processed_data["y_min"], processed_data["y_max"]],
"showgrid": grid_enabled,
},
"hovermode": "x unified",
"showlegend": True,
},
"layout": layout,
}
return fig
@ -158,13 +165,7 @@ class MagnitudeProcessor(BaseProcessor):
# ------------------------------------------------------------------ #
def get_ui_parameters(self) -> list[UIParameter]:
"""
UI/validation schema.
Conforms to BaseProcessor rules:
- slider: requires dtype + min/max/step alignment checks
- toggle: bool only
- input: numeric only, with {"type": "int"|"float", "min"?, "max"?}
- button: {"action": "..."}; value ignored by validation
UI/validation schema for magnitude processor.
"""
return [
UIParameter(
@ -172,56 +173,20 @@ class MagnitudeProcessor(BaseProcessor):
label="Y Axis Min (dB)",
type="slider",
value=self._config.get("y_min", -80),
options={"min": -120, "max": 0, "step": 5, "dtype": "int"},
options={"min": -80, "max": 20, "step": 5, "dtype": "int"},
),
UIParameter(
name="y_max",
label="Y Axis Max (dB)",
type="slider",
value=self._config.get("y_max", 10),
options={"min": -20, "max": 20, "step": 5, "dtype": "int"},
options={"min": -20, "max": 40, "step": 5, "dtype": "int"},
),
UIParameter(
name="smoothing_enabled",
label="Enable Smoothing",
name="show_phase",
label="Show Phase",
type="toggle",
value=self._config.get("smoothing_enabled", False),
options={},
),
UIParameter(
name="smoothing_window",
label="Smoothing Window Size",
type="slider",
value=self._config.get("smoothing_window", 5),
options={"min": 3, "max": 21, "step": 2, "dtype": "int"},
),
UIParameter(
name="marker_enabled",
label="Show Marker",
type="toggle",
value=self._config.get("marker_enabled", True),
options={},
),
UIParameter(
name="marker_frequency",
label="Marker Frequency (Hz)",
type="input",
value=self._config.get("marker_frequency", 1e9),
options={"type": "float", "min": 100e6, "max": 8.8e9},
),
UIParameter(
name="grid_enabled",
label="Show Grid",
type="toggle",
value=self._config.get("grid_enabled", True),
options={},
),
UIParameter(
name="reset_smoothing",
label="Reset Smoothing",
type="button",
value=False, # buttons carry no state; ignored by validator
options={"action": "Reset the smoothing filter state"},
value=self._config.get("show_phase", False),
),
]
@ -230,64 +195,34 @@ class MagnitudeProcessor(BaseProcessor):
return {
"y_min": -80,
"y_max": 10,
"smoothing_enabled": False,
"smoothing_window": 5,
"marker_enabled": True,
"marker_frequency": 1e9,
"grid_enabled": True,
"show_phase": False,
}
# ------------------------------------------------------------------ #
# Config updates & actions
# ------------------------------------------------------------------ #
def update_config(self, updates: dict[str, Any]) -> None:
"""
Apply config updates; handle UI buttons out-of-band.
Any key that corresponds to a button triggers an action when True and
is *not* persisted in the config. Other keys are forwarded to the
BaseProcessor (with type conversion + validation).
Update configuration with validation to ensure y_max > y_min.
"""
ui_params = {param.name: param for param in self.get_ui_parameters()}
button_actions: dict[str, bool] = {}
config_updates: dict[str, Any] = {}
# Apply base update first
super().update_config(updates)
for key, value in updates.items():
schema = ui_params.get(key)
if schema and schema.type == "button":
if value:
button_actions[key] = True
else:
config_updates[key] = value
# Validate y_max > y_min after update
y_min = float(self._config.get("y_min", -80))
y_max = float(self._config.get("y_max", 10))
if config_updates:
super().update_config(config_updates)
if y_max <= y_min:
# Adjust y_max to be at least 5 dB above y_min
new_y_max = y_min + 5
# Clamp to maximum allowed value
max_allowed = 40
if new_y_max > max_allowed:
new_y_max = max_allowed
# If we hit the ceiling, adjust y_min instead
if new_y_max <= y_min:
self._config["y_min"] = new_y_max - 5
# Execute button actions
if button_actions.get("reset_smoothing"):
self._smoothing_history.clear()
logger.info("Smoothing state reset via UI button")
self._config["y_max"] = new_y_max
self.save_config()
logger.info("Adjusted y_max to maintain y_max > y_min",
y_min=self._config["y_min"],
y_max=self._config["y_max"])
# ------------------------------------------------------------------ #
# Smoothing
# ------------------------------------------------------------------ #
@staticmethod
def _apply_moving_average(data: list[float], window_size: int) -> list[float]:
"""
Centered moving average with clamped edges.
Requirements
------------
- window_size must be odd and >= 3 (enforced by UI schema).
"""
n = len(data)
if n == 0 or window_size <= 1 or window_size >= n:
return data
half = window_size // 2
out: list[float] = []
for i in range(n):
lo = max(0, i - half)
hi = min(n, i + half + 1)
out.append(sum(data[lo:hi]) / (hi - lo))
return out

307
vna_system/core/processors/implementations/phase_processor.py
View File

@ -1,307 +0,0 @@
import numpy as np
from pathlib import Path
from typing import Any
from vna_system.core.logging.logger import get_component_logger
from vna_system.core.processors.base_processor import BaseProcessor, UIParameter
logger = get_component_logger(__file__)
class PhaseProcessor(BaseProcessor):
"""
Compute and visualize phase (degrees) over frequency from calibrated sweep data.
Pipeline
--------
1) Derive frequency axis from VNA config (start/stop, N points).
2) Compute phase in degrees from complex samples.
3) Optional simple unwrapping (+/-180° jumps) and offset.
4) Optional moving-average smoothing.
5) Provide Plotly configuration including an optional marker and reference line.
"""
def __init__(self, config_dir: Path) -> None:
super().__init__("phase", config_dir)
# ------------------------------------------------------------------ #
# Core processing
# ------------------------------------------------------------------ #
def process_sweep(self, sweep_data: Any, calibrated_data: Any, vna_config: dict[str, Any]) -> dict[str, Any]:
"""
Produce phase trace (degrees) and ancillary info from a calibrated sweep.
Returns
-------
dict[str, Any]
{
'frequencies': list[float],
'phases_deg': list[float],
'y_min': float,
'y_max': float,
'marker_enabled': bool,
'marker_frequency': float,
'grid_enabled': bool,
'reference_line_enabled': bool,
'reference_phase': float
}
"""
if not calibrated_data or not hasattr(calibrated_data, "points"):
logger.warning("No calibrated data available for phase processing")
return {"error": "No calibrated data available"}
points: list[tuple[float, float]] = calibrated_data.points
n = len(points)
if n == 0:
logger.warning("Calibrated sweep contains zero points")
return {"error": "Empty calibrated sweep"}
# Frequency axis from VNA config (defaults if not provided)
start_freq = float(vna_config.get("start_frequency", 100e6))
stop_freq = float(vna_config.get("stop_frequency", 8.8e9))
if n == 1:
freqs = [start_freq]
else:
step = (stop_freq - start_freq) / (n - 1)
freqs = [start_freq + i * step for i in range(n)]
# Phase in degrees
phases_deg: list[float] = []
unwrap = bool(self._config.get("unwrap_phase", True))
for i, (real, imag) in enumerate(points):
z = complex(real, imag)
deg = float(np.degrees(np.angle(z)))
if unwrap and phases_deg:
diff = deg - phases_deg[-1]
if diff > 180.0:
deg -= 360.0
elif diff < -180.0:
deg += 360.0
phases_deg.append(deg)
# Offset
phase_offset = float(self._config.get("phase_offset", 0.0))
if phase_offset:
phases_deg = [p + phase_offset for p in phases_deg]
# Optional smoothing
if self._config.get("smoothing_enabled", False):
window = int(self._config.get("smoothing_window", 5))
phases_deg = self._apply_moving_average(phases_deg, window)
result = {
"frequencies": freqs,
"phases_deg": phases_deg,
"y_min": float(self._config.get("y_min", -180)),
"y_max": float(self._config.get("y_max", 180)),
"marker_enabled": bool(self._config.get("marker_enabled", True)),
"marker_frequency": float(
self._config.get("marker_frequency", freqs[len(freqs) // 2] if freqs else 1e9)
),
"grid_enabled": bool(self._config.get("grid_enabled", True)),
"reference_line_enabled": bool(self._config.get("reference_line_enabled", False)),
"reference_phase": float(self._config.get("reference_phase", 0)),
}
logger.debug("Phase sweep processed", points=n)
return result
def generate_plotly_config(self, processed_data: dict[str, Any], vna_config: dict[str, Any]) -> dict[str, Any]:
"""
Build a Plotly figure config for the phase trace, marker, and optional reference line.
"""
if "error" in processed_data:
return {"error": processed_data["error"]}
freqs: list[float] = processed_data["frequencies"]
phases: list[float] = processed_data["phases_deg"]
grid_enabled: bool = processed_data["grid_enabled"]
# Marker
marker_freq: float = processed_data["marker_frequency"]
if freqs:
idx = min(range(len(freqs)), key=lambda i: abs(freqs[i] - marker_freq))
marker_y = phases[idx]
marker_trace = {
"x": [freqs[idx] / 1e9],
"y": [marker_y],
"type": "scatter",
"mode": "markers",
"name": f"Marker: {freqs[idx]/1e9:.3f} GHz, {marker_y:.1f}°",
"marker": {"color": "red", "size": 8, "symbol": "circle"},
}
else:
marker_trace = None
traces = [
{
"x": [f / 1e9 for f in freqs], # Hz -> GHz
"y": phases,
"type": "scatter",
"mode": "lines",
"name": "Phase",
"line": {"color": "green", "width": 2},
}
]
if processed_data["marker_enabled"] and marker_trace:
traces.append(marker_trace)
# Reference line
if processed_data["reference_line_enabled"] and freqs:
ref = float(processed_data["reference_phase"])
traces.append(
{
"x": [freqs[0] / 1e9, freqs[-1] / 1e9],
"y": [ref, ref],
"type": "scatter",
"mode": "lines",
"name": f"Reference: {ref:.1f}°",
"line": {"color": "gray", "width": 1, "dash": "dash"},
}
)
fig = {
"data": traces,
"layout": {
"title": "Phase Response",
"xaxis": {"title": "Frequency (GHz)", "showgrid": grid_enabled},
"yaxis": {
"title": "Phase (degrees)",
"range": [processed_data["y_min"], processed_data["y_max"]],
"showgrid": grid_enabled,
},
"hovermode": "x unified",
"showlegend": True,
},
}
return fig
# ------------------------------------------------------------------ #
# UI schema
# ------------------------------------------------------------------ #
def get_ui_parameters(self) -> list[UIParameter]:
"""
UI/validation schema (compatible with BaseProcessor's validators).
"""
return [
UIParameter(
name="y_min",
label="Y Axis Min (degrees)",
type="slider",
value=self._config.get("y_min", -180),
options={"min": -360, "max": 0, "step": 15, "dtype": "int"},
),
UIParameter(
name="y_max",
label="Y Axis Max (degrees)",
type="slider",
value=self._config.get("y_max", 180),
options={"min": 0, "max": 360, "step": 15, "dtype": "int"},
),
UIParameter(
name="unwrap_phase",
label="Unwrap Phase",
type="toggle",
value=self._config.get("unwrap_phase", True),
options={},
),
UIParameter(
name="phase_offset",
label="Phase Offset (degrees)",
type="slider",
value=self._config.get("phase_offset", 0),
options={"min": -180, "max": 180, "step": 5, "dtype": "int"},
),
UIParameter(
name="smoothing_enabled",
label="Enable Smoothing",
type="toggle",
value=self._config.get("smoothing_enabled", False),
options={},
),
UIParameter(
name="smoothing_window",
label="Smoothing Window Size",
type="slider",
value=self._config.get("smoothing_window", 5),
options={"min": 3, "max": 21, "step": 2, "dtype": "int"},
),
UIParameter(
name="marker_enabled",
label="Show Marker",
type="toggle",
value=self._config.get("marker_enabled", True),
options={},
),
UIParameter(
name="marker_frequency",
label="Marker Frequency (Hz)",
type="input",
value=self._config.get("marker_frequency", 1e9),
options={"type": "float", "min": 100e6, "max": 8.8e9},
),
UIParameter(
name="reference_line_enabled",
label="Show Reference Line",
type="toggle",
value=self._config.get("reference_line_enabled", False),
options={},
),
UIParameter(
name="reference_phase",
label="Reference Phase (degrees)",
type="slider",
value=self._config.get("reference_phase", 0),
options={"min": -180, "max": 180, "step": 15, "dtype": "int"},
),
UIParameter(
name="grid_enabled",
label="Show Grid",
type="toggle",
value=self._config.get("grid_enabled", True),
options={},
),
]
def _get_default_config(self) -> dict[str, Any]:
"""Defaults aligned with the UI schema."""
return {
"y_min": -180,
"y_max": 180,
"unwrap_phase": True,
"phase_offset": 0,
"smoothing_enabled": False,
"smoothing_window": 5,
"marker_enabled": True,
"marker_frequency": 1e9,
"reference_line_enabled": False,
"reference_phase": 0,
"grid_enabled": True,
}
# ------------------------------------------------------------------ #
# Smoothing
# ------------------------------------------------------------------ #
@staticmethod
def _apply_moving_average(data: list[float], window_size: int) -> list[float]:
"""
Centered moving average with clamped edges.
Requirements
------------
- `window_size` must be odd and >= 3 (enforced by UI schema).
"""
n = len(data)
if n == 0 or window_size <= 1 or window_size >= n:
return data
half = window_size // 2
out: list[float] = []
for i in range(n):
lo = max(0, i - half)
hi = min(n, i + half + 1)
out.append(sum(data[lo:hi]) / (hi - lo))
return out

303
vna_system/core/processors/implementations/smith_chart_processor.py
View File

@ -1,303 +0,0 @@
import numpy as np
from typing import Dict, Any, List
from pathlib import Path
from ..base_processor import BaseProcessor, UIParameter
class SmithChartProcessor(BaseProcessor):
def __init__(self, config_dir: Path):
super().__init__("smith_chart", config_dir)
def process_sweep(self, sweep_data: Any, calibrated_data: Any, vna_config: Dict[str, Any]) -> Dict[str, Any]:
if not calibrated_data or not hasattr(calibrated_data, 'points'):
return {'error': 'No calibrated data available'}
frequencies = []
real_parts = []
imag_parts = []
reflection_coeffs = []
for i, (real, imag) in enumerate(calibrated_data.points):
complex_val = complex(real, imag)
# Calculate frequency
start_freq = vna_config.get('start_frequency', 100e6)
stop_freq = vna_config.get('stop_frequency', 8.8e9)
total_points = len(calibrated_data.points)
frequency = start_freq + (stop_freq - start_freq) * i / (total_points - 1)
frequencies.append(frequency)
real_parts.append(real)
imag_parts.append(imag)
reflection_coeffs.append(complex_val)
# Convert to impedance if requested
impedance_mode = self._config.get('impedance_mode', False)
z0 = self._config.get('reference_impedance', 50)
if impedance_mode:
impedances = [(z0 * (1 + gamma) / (1 - gamma)) for gamma in reflection_coeffs]
plot_real = [z.real for z in impedances]
plot_imag = [z.imag for z in impedances]
else:
plot_real = real_parts
plot_imag = imag_parts
return {
'frequencies': frequencies,
'real_parts': plot_real,
'imag_parts': plot_imag,
'impedance_mode': impedance_mode,
'reference_impedance': z0,
'marker_enabled': self._config.get('marker_enabled', True),
'marker_frequency': self._config.get('marker_frequency', frequencies[len(frequencies)//2] if frequencies else 1e9),
'grid_circles': self._config.get('grid_circles', True),
'grid_radials': self._config.get('grid_radials', True),
'trace_color_mode': self._config.get('trace_color_mode', 'frequency')
}
def generate_plotly_config(self, processed_data: Dict[str, Any], vna_config: Dict[str, Any]) -> Dict[str, Any]:
if 'error' in processed_data:
return {'error': processed_data['error']}
real_parts = processed_data['real_parts']
imag_parts = processed_data['imag_parts']
frequencies = processed_data['frequencies']
# Find marker point
marker_freq = processed_data['marker_frequency']
marker_idx = min(range(len(frequencies)), key=lambda i: abs(frequencies[i] - marker_freq))
# Create main trace
trace_config = {
'x': real_parts,
'y': imag_parts,
'type': 'scatter',
'mode': 'lines+markers',
'marker': {'size': 3},
'line': {'width': 2}
}
if processed_data['trace_color_mode'] == 'frequency':
# Color by frequency
trace_config['marker']['color'] = [f / 1e9 for f in frequencies]
trace_config['marker']['colorscale'] = 'Viridis'
trace_config['marker']['showscale'] = True
trace_config['marker']['colorbar'] = {'title': 'Frequency (GHz)'}
trace_config['name'] = 'S11 (colored by frequency)'
else:
trace_config['marker']['color'] = 'blue'
trace_config['name'] = 'S11'
traces = [trace_config]
# Add marker if enabled
if processed_data['marker_enabled']:
marker_real = real_parts[marker_idx]
marker_imag = imag_parts[marker_idx]
if processed_data['impedance_mode']:
marker_label = f'Z = {marker_real:.1f} + j{marker_imag:.1f} Ω'
else:
marker_label = f'Γ = {marker_real:.3f} + j{marker_imag:.3f}'
traces.append({
'x': [marker_real],
'y': [marker_imag],
'type': 'scatter',
'mode': 'markers',
'name': f'Marker: {frequencies[marker_idx]/1e9:.3f} GHz<br>{marker_label}',
'marker': {'color': 'red', 'size': 10, 'symbol': 'diamond'}
})
# Add Smith chart grid if not in impedance mode
if not processed_data['impedance_mode']:
if processed_data['grid_circles']:
traces.extend(self._generate_smith_circles())
if processed_data['grid_radials']:
traces.extend(self._generate_smith_radials())
# Layout configuration
if processed_data['impedance_mode']:
layout = {
'title': f'Impedance Plot (Z₀ = {processed_data["reference_impedance"]} Ω)',
'xaxis': {'title': 'Resistance (Ω)', 'showgrid': True, 'zeroline': True},
'yaxis': {'title': 'Reactance (Ω)', 'showgrid': True, 'zeroline': True, 'scaleanchor': 'x'},
'hovermode': 'closest'
}
else:
layout = {
'title': 'Smith Chart',
'xaxis': {
'title': 'Real Part',
'range': [-1.1, 1.1],
'showgrid': False,
'zeroline': False,
'showticklabels': True
},
'yaxis': {
'title': 'Imaginary Part',
'range': [-1.1, 1.1],
'showgrid': False,
'zeroline': False,
'scaleanchor': 'x',
'scaleratio': 1
},
'hovermode': 'closest'
}
layout['showlegend'] = True
return {
'data': traces,
'layout': layout
}
def _generate_smith_circles(self) -> List[Dict[str, Any]]:
circles = []
# Constant resistance circles
for r in [0, 0.2, 0.5, 1, 2, 5]:
if r == 0:
# Unit circle
theta = np.linspace(0, 2*np.pi, 100)
x = np.cos(theta)
y = np.sin(theta)
else:
center_x = r / (r + 1)
radius = 1 / (r + 1)
theta = np.linspace(0, 2*np.pi, 100)
x = center_x + radius * np.cos(theta)
y = radius * np.sin(theta)
circles.append({
'x': x.tolist(),
'y': y.tolist(),
'type': 'scatter',
'mode': 'lines',
'line': {'color': 'lightgray', 'width': 1},
'showlegend': False,
'hoverinfo': 'skip'
})
# Constant reactance circles
for x in [-5, -2, -1, -0.5, -0.2, 0.2, 0.5, 1, 2, 5]:
if abs(x) < 1e-10:
continue
center_y = 1/x
radius = abs(1/x)
theta = np.linspace(-np.pi, np.pi, 100)
circle_x = 1 + radius * np.cos(theta)
circle_y = center_y + radius * np.sin(theta)
# Clip to unit circle
valid_points = circle_x**2 + circle_y**2 <= 1.01
circle_x = circle_x[valid_points]
circle_y = circle_y[valid_points]
if len(circle_x) > 0:
circles.append({
'x': circle_x.tolist(),
'y': circle_y.tolist(),
'type': 'scatter',
'mode': 'lines',
'line': {'color': 'lightblue', 'width': 1},
'showlegend': False,
'hoverinfo': 'skip'
})
return circles
def _generate_smith_radials(self) -> List[Dict[str, Any]]:
radials = []
# Radial lines for constant phase
for angle_deg in range(0, 360, 30):
angle_rad = np.radians(angle_deg)
x = [0, np.cos(angle_rad)]
y = [0, np.sin(angle_rad)]
radials.append({
'x': x,
'y': y,
'type': 'scatter',
'mode': 'lines',
'line': {'color': 'lightgray', 'width': 1, 'dash': 'dot'},
'showlegend': False,
'hoverinfo': 'skip'
})
return radials
def get_ui_parameters(self) -> List[UIParameter]:
return [
UIParameter(
name='impedance_mode',
label='Impedance Plot Mode',
type='toggle',
value=self._config.get('impedance_mode', False)
),
UIParameter(
name='reference_impedance',
label='Reference Impedance (Ω)',
type='select',
value=self._config.get('reference_impedance', 50),
options={'choices': [25, 50, 75, 100, 600]}
),
UIParameter(
name='marker_enabled',
label='Show Marker',
type='toggle',
value=self._config.get('marker_enabled', True)
),
UIParameter(
name='marker_frequency',
label='Marker Frequency (Hz)',
type='input',
value=self._config.get('marker_frequency', 1e9),
options={'type': 'number', 'min': 100e6, 'max': 8.8e9}
),
UIParameter(
name='grid_circles',
label='Show Impedance Circles',
type='toggle',
value=self._config.get('grid_circles', True)
),
UIParameter(
name='grid_radials',
label='Show Phase Radials',
type='toggle',
value=self._config.get('grid_radials', True)
),
UIParameter(
name='trace_color_mode',
label='Trace Color Mode',
type='select',
value=self._config.get('trace_color_mode', 'frequency'),
options={'choices': ['frequency', 'solid']}
)
]
def _get_default_config(self) -> Dict[str, Any]:
return {
'impedance_mode': False,
'reference_impedance': 50,
'marker_enabled': True,
'marker_frequency': 1e9,
'grid_circles': True,
'grid_radials': True,
'trace_color_mode': 'frequency'
}
def _validate_config(self):
required_keys = ['impedance_mode', 'reference_impedance', 'marker_enabled',
'marker_frequency', 'grid_circles', 'grid_radials', 'trace_color_mode']
for key in required_keys:
if key not in self._config:
raise ValueError(f"Missing required config key: {key}")
if self._config['reference_impedance'] <= 0:
raise ValueError("reference_impedance must be positive")
if self._config['trace_color_mode'] not in ['frequency', 'solid']:
raise ValueError("trace_color_mode must be 'frequency' or 'solid'")

4
vna_system/core/processors/manager.py
View File

@ -267,11 +267,11 @@ class ProcessorManager:
we log an error and keep going with whatever is available.
"""
try:
from .implementations import MagnitudeProcessor, PhaseProcessor, SmithChartProcessor
from .implementations.magnitude_processor import MagnitudeProcessor
from .implementations.bscan_processor import BScanProcessor
self.register_processor(MagnitudeProcessor(self.config_dir))
self.register_processor(PhaseProcessor(self.config_dir))
# self.register_processor(PhaseProcessor(self.config_dir))
self.register_processor(BScanProcessor(self.config_dir))
# self.register_processor(SmithChartProcessor(self.config_dir))

41
vna_system/core/processors/websocket_handler.py
View File

@ -84,6 +84,14 @@ class ProcessorWebSocketHandler:
# --------------------------------------------------------------------- #
async def handle_message(self, websocket: WebSocket, data: str) -> None:
"""Parse and route an inbound client message."""
# Handle ping/pong messages first (they are not JSON)
if data == "ping":
await websocket.send_text("pong")
return
elif data == "pong":
# Just acknowledge, no response needed
return
try:
message = json.loads(data)
mtype = message.get("type")
@ -94,8 +102,15 @@ class ProcessorWebSocketHandler:
await self._handle_get_history(websocket, message)
else:
await self._send_error(websocket, f"Unknown message type: {mtype!r}")
except json.JSONDecodeError:
await self._send_error(websocket, "Invalid JSON format")
except json.JSONDecodeError as json_error:
logger.error("JSON decode error", raw_data=data[:200], error=str(json_error))
await self._send_error(
websocket,
"Invalid JSON format",
details=f"JSON parse error: {json_error}",
source="websocket_handler",
raw_data=data[:200] if len(data) <= 200 else f"{data[:200]}..."
)
except Exception as exc: # noqa: BLE001
logger.error("Error handling websocket message")
await self._send_error(websocket, f"Internal error: {exc}")
@ -170,7 +185,16 @@ class ProcessorWebSocketHandler:
"metadata": result.metadata,
}
async def _send_error(self, websocket: WebSocket, message: str) -> None:
async def _send_error(
self,
websocket: WebSocket,
message: str,
*,
details: str | None = None,
source: str | None = None,
code: str | None = None,
raw_data: str | None = None
) -> None:
"""Send a standardized error payload to a single client."""
try:
payload = {
@ -178,6 +202,17 @@ class ProcessorWebSocketHandler:
"message": message,
"timestamp": datetime.now().timestamp(),
}
# Add optional fields if provided
if details:
payload["details"] = details
if source:
payload["source"] = source
if code:
payload["code"] = code
if raw_data:
payload["raw_data"] = raw_data
await websocket.send_text(json.dumps(payload))
except Exception as exc: # noqa: BLE001
logger.error("Error sending error message", error=repr(exc))

33
vna_system/web_ui/static/css/charts.css
View File

@ -57,23 +57,38 @@
.chart-card:fullscreen .chart-card__header {
flex-shrink: 0;
padding: var(--space-3) var(--space-4);
border-bottom: 1px solid var(--border-primary);
}
.chart-card:fullscreen .chart-card__content {
flex: 1;
display: flex;
flex-direction: column;
flex-direction: row !important; /* Горизонтальное расположение вместо колонки */
padding: var(--space-2);
min-height: 0;
height: calc(100vh - 80px); /* Вычитаем высоту header и meta */
gap: var(--space-4);
}
.chart-card:fullscreen .chart-card__plot {
flex: 1 !important;
width: 100% !important;
width: calc(100% - 270px) !important; /* Учитываем ширину настроек + gap */
height: 100% !important;
min-height: 0 !important;
}
.chart-card:fullscreen .chart-card__settings {
width: 250px !important;
flex-shrink: 0 !important;
height: 100% !important; /* Занимает всю доступную высоту */
max-height: none !important; /* Убираем ограничение высоты */
overflow-y: auto !important;
background-color: var(--bg-secondary);
border: 1px solid var(--border-primary);
border-radius: var(--radius-lg);
padding: var(--space-3);
}
.chart-card:fullscreen .chart-card__plot .js-plotly-plot {
width: 100% !important;
height: 100% !important;
@ -84,9 +99,23 @@
height: 100% !important;
}
/* Plotly specific fullscreen adjustments */
.chart-card:fullscreen .chart-card__plot .plotly {
width: 100% !important;
height: 100% !important;
}
/* Force Plotly to use full available height in fullscreen */
.chart-card:fullscreen .chart-card__plot .js-plotly-plot,
.chart-card:fullscreen .chart-card__plot .js-plotly-plot > div {
height: calc(100vh - 120px) !important; /* Вычитаем высоту header + padding + meta */
min-height: calc(100vh - 120px) !important;
}
.chart-card:fullscreen .chart-card__meta {
flex-shrink: 0;
padding: var(--space-2) var(--space-4);
border-top: 1px solid var(--border-primary);
}
/* Chart loading state */

28
vna_system/web_ui/static/css/components.css
View File

@ -276,14 +276,14 @@
display: flex;
position: relative;
padding: var(--space-4);
min-height: 450px;
min-height: 585px; /* Увеличено на 30%: 450 * 1.3 = 585 */
gap: var(--space-4);
}
.chart-card__plot {
flex: 1;
height: 420px !important;
min-height: 400px;
height: 546px !important; /* Увеличено на 30%: 420 * 1.3 = 546 */
min-height: 520px; /* Увеличено на 30%: 400 * 1.3 = 520 */
position: relative;
min-width: 0; /* Allows flex item to shrink */
}
@ -296,7 +296,7 @@
border-radius: var(--radius-lg);
padding: var(--space-3);
overflow-y: auto;
max-height: 420px;
max-height: 546px; /* Увеличено на 30%: 420 * 1.3 = 546 */
}
.chart-card__plot .js-plotly-plot {
@ -312,7 +312,7 @@
/* Fix Plotly toolbar positioning with project design system */
.chart-card__plot .modebar {
position: absolute !important;
top: var(--space-3) !important;
top: 4px !important; /* Максимально высоко - минимальный отступ */
right: var(--space-3) !important;
z-index: 1000 !important;
background: var(--bg-surface) !important;
@ -381,6 +381,24 @@
transform: translateY(-1px) !important;
}
/* Plotly title positioning to avoid overlapping with modebar */
.chart-card__plot .gtitle {
margin-top: -10px !important; /* Поднимаем заголовок на уровень кнопок управления */
margin-bottom: 25px !important; /* Увеличенный отступ от подписей осей */
}
/* Plotly chart area adjustment for better spacing */
.chart-card__plot .plotly .main-svg {
padding-top: 40px !important; /* Уменьшенный отступ сверху */
padding-right: 8px !important;
}
/* Positioning for axis titles to avoid overlap with main title */
.chart-card__plot .xtitle,
.chart-card__plot .ytitle {
margin-top: 15px !important; /* Увеличенный отступ для подписей осей */
}
.chart-card__meta {
display: flex;
justify-content: space-between;

13
vna_system/web_ui/static/js/modules/websocket.js
View File

@ -165,9 +165,18 @@ export class WebSocketManager {
this.emit('processor_history', payload);
break;
case 'error':
console.error('🔴 Server error:', payload.message);
console.error('🔴 Server error:', payload);
console.error('🔴 Error details:', {
message: payload.message,
code: payload.code,
details: payload.details,
timestamp: payload.timestamp,
source: payload.source
});
this.notifications?.show?.({
type: 'error', title: 'Server Error', message: payload.message
type: 'error',
title: 'Server Error',
message: `${payload.message}${payload.details ? ` - ${payload.details}` : ''}${payload.source ? ` (${payload.source})` : ''}`
});
break;
default: