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

some fixes and improvements

Browse Source
This commit is contained in:
ayzen
2025-10-06 17:35:52 +03:00
parent c16c90cdfd
commit 6297155f71
9 changed files with 372 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

79
vna_system/core/processors/base_processor.py
View File

@ -559,6 +559,84 @@ class BaseProcessor:
if max_v is not None and value > max_v:
raise ValueError(f"{key} {value} > max {max_v}")
# --------------------------------------------------------------------- #
# History Export/Import
# --------------------------------------------------------------------- #
def export_history_data(self) -> list[dict[str, Any]]:
"""
Export sweep history in JSON-serializable format.
Returns
-------
list[dict]
Serializable history records with sweep data converted to points.
"""
with self._lock:
exported = []
for entry in self._sweep_history:
sweep_data = entry["sweep_data"]
calibrated_data = entry["calibrated_data"]
reference_data = entry.get("reference_data")
exported.append({
"sweep_number": sweep_data.sweep_number if sweep_data else None,
"timestamp": entry.get("timestamp"),
"sweep_points": sweep_data.points if sweep_data else [],
"calibrated_points": calibrated_data.points if calibrated_data else [],
"reference_points": reference_data.points if reference_data else [],
"vna_config": entry.get("vna_config", {}),
})
return exported
def import_history_data(self, history_data: list[dict[str, Any]]) -> None:
"""
Import sweep history from JSON data.
Parameters
----------
history_data : list[dict]
History records in the format exported by export_history_data.
"""
from vna_system.core.acquisition.sweep_buffer import SweepData
with self._lock:
self._sweep_history.clear()
for entry in history_data:
sweep_points = entry.get("sweep_points", [])
calibrated_points = entry.get("calibrated_points", [])
reference_points = entry.get("reference_points", [])
# Reconstruct SweepData objects
sweep_data = SweepData(
sweep_number=entry.get("sweep_number", 0),
timestamp=entry.get("timestamp", 0.0),
points=sweep_points
) if sweep_points else None
calibrated_data = SweepData(
sweep_number=entry.get("sweep_number", 0),
timestamp=entry.get("timestamp", 0.0),
points=calibrated_points
) if calibrated_points else None
reference_data = SweepData(
sweep_number=entry.get("sweep_number", 0),
timestamp=entry.get("timestamp", 0.0),
points=reference_points
) if reference_points else None
self._sweep_history.append({
"sweep_data": sweep_data,
"calibrated_data": calibrated_data,
"reference_data": reference_data,
"vna_config": entry.get("vna_config", {}),
"timestamp": entry.get("timestamp"),
})
logger.info("History imported", processor_id=self.processor_id, records=len(history_data))
# --------------------------------------------------------------------- #
# Utilities
# --------------------------------------------------------------------- #
@ -572,4 +650,5 @@ class BaseProcessor:
"config": self._config.copy(),
"history_count": len(self._sweep_history),
"max_history": self._max_history,
"sweep_history": self.export_history_data(),
}

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

@ -1,5 +1,6 @@
from __future__ import annotations
from datetime import datetime
from pathlib import Path
from typing import Any
@ -363,6 +364,8 @@ class BScanProcessor(BaseProcessor):
self._plot_history.clear()
# Process all sweeps in history with current config
last_processed = None
last_vna_config = {}
for entry in self._sweep_history:
sweep_data = entry["sweep_data"]
calibrated_data = entry["calibrated_data"]
@ -372,8 +375,9 @@ class BScanProcessor(BaseProcessor):
processed = self.process_sweep(sweep_data, calibrated_data, vna_config)
# Skip if processing failed
if "error" in processed:
continue
if "error" not in processed:
last_processed = processed
last_vna_config = vna_config
# Trim plot history if needed
if len(self._plot_history) > self._max_history:
@ -383,16 +387,22 @@ class BScanProcessor(BaseProcessor):
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"]
)
# Build result from last successful processing
if last_processed is None:
return None
return None
# Generate plotly config and wrap into ProcessedResult
plotly_conf = self.generate_plotly_config(last_processed, last_vna_config)
ui_params = self.get_ui_parameters()
return ProcessedResult(
processor_id=self.processor_id,
timestamp=datetime.now().timestamp(),
data=last_processed,
plotly_config=plotly_conf,
ui_parameters=ui_params,
metadata=self._get_metadata(),
)
# -------------------------------------------------------------------------
# Low-level helpers

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

@ -71,25 +71,27 @@ class MagnitudeProcessor(BaseProcessor):
# Magnitude in dB (clamp zero magnitude to -120 dB)
mags_db: list[float] = []
phases_deg: list[float] = []
# real_points: list[str] = []
# imag_points: list[str] = []
real_points: list[float] = []
imag_points: list[float] = []
for real, imag in points:
complex_val = complex(real, imag)
# real_points.append(str(real))
# imag_points.append(str(imag))
real_points.append(float(real))
imag_points.append(float(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,
# "real_points" : real_points,
# "imag_points" : imag_points,
"real_points": real_points,
"imag_points": imag_points,
"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)),
"autoscale": bool(self._config.get("autoscale", False)),
"show_magnitude": bool(self._config.get("show_magnitude", True)),
"show_phase": bool(self._config.get("show_phase", False)),
}
logger.debug("Magnitude sweep processed", points=n)
@ -105,7 +107,9 @@ class MagnitudeProcessor(BaseProcessor):
freqs: list[float] = processed_data["frequencies"]
mags_db: list[float] = processed_data["magnitudes_db"]
phases_deg: list[float] = processed_data["phases_deg"]
show_magnitude: bool = processed_data["show_magnitude"]
show_phase: bool = processed_data["show_phase"]
autoscale: bool = processed_data["autoscale"]
# Determine the parameter type from preset mode
parameter_type = vna_config["mode"].value.upper() # Convert "s11" -> "S11", "s21" -> "S21"
@ -113,17 +117,23 @@ class MagnitudeProcessor(BaseProcessor):
# Convert Hz to GHz for x-axis
freqs_ghz = [f / 1e9 for f in freqs]
traces = [
{
# Pleasant colors
magnitude_color = "rgb(46, 204, 113)" # Pleasant green
phase_color = "rgb(231, 76, 60)" # Pleasant red/coral
traces = []
# Add magnitude trace if enabled
if show_magnitude:
traces.append({
"x": freqs_ghz,
"y": mags_db,
"type": "scatter",
"mode": "lines",
"name": f"|{parameter_type}| Magnitude",
"line": {"color": "blue", "width": 2},
"line": {"color": magnitude_color, "width": 2},
"yaxis": "y",
}
]
})
# Add phase trace if enabled
if show_phase:
@ -133,34 +143,58 @@ class MagnitudeProcessor(BaseProcessor):
"type": "scatter",
"mode": "lines",
"name": f"∠{parameter_type} Phase",
"line": {"color": "red", "width": 2},
"yaxis": "y2",
"line": {"color": phase_color, "width": 2},
"yaxis": "y2" if show_magnitude else "y",
})
# 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],
# Configure y-axis based on what's shown
if show_magnitude:
y_axis_config = {
"title": "Magnitude (dB)",
"showgrid": True,
"side": "left",
"titlefont": {"color": magnitude_color},
"tickfont": {"color": magnitude_color},
}
# Apply autoscale or manual range
if not autoscale:
y_axis_config["range"] = [processed_data["y_min"], processed_data["y_max"]]
layout["yaxis"] = y_axis_config
# Add second y-axis for phase if both are shown
if show_phase:
if show_magnitude:
# Phase on second axis (radians converted to degrees, but displayed as -π to π)
layout["yaxis2"] = {
"title": "Phase (rad)",
"overlaying": "y",
"side": "right",
"showgrid": False,
"range": [-180, 180], # -π to π in degrees
"titlefont": {"color": phase_color},
"tickfont": {"color": phase_color},
}
else:
# Phase on primary axis if magnitude is hidden
layout["yaxis"] = {
"title": "Phase (rad)",
"showgrid": True,
"side": "left",
"range": [-180, 180], # -π to π in degrees
"titlefont": {"color": phase_color},
"tickfont": {"color": phase_color},
}
fig = {
"data": traces,
"layout": layout,
@ -175,6 +209,24 @@ class MagnitudeProcessor(BaseProcessor):
UI/validation schema for magnitude processor.
"""
return [
UIParameter(
name="show_magnitude",
label="Show Magnitude",
type="toggle",
value=self._config.get("show_magnitude", True),
),
UIParameter(
name="show_phase",
label="Show Phase",
type="toggle",
value=self._config.get("show_phase", False),
),
UIParameter(
name="autoscale",
label="Autoscale Y Axis",
type="toggle",
value=self._config.get("autoscale", False),
),
UIParameter(
name="y_min",
label="Y Axis Min (dB)",
@ -189,12 +241,6 @@ class MagnitudeProcessor(BaseProcessor):
value=self._config.get("y_max", 10),
options={"min": -20, "max": 40, "step": 5, "dtype": "int"},
),
UIParameter(
name="show_phase",
label="Show Phase",
type="toggle",
value=self._config.get("show_phase", False),
),
]
def _get_default_config(self) -> dict[str, Any]:
@ -202,6 +248,8 @@ class MagnitudeProcessor(BaseProcessor):
return {
"y_min": -80,
"y_max": 10,
"autoscale": False,
"show_magnitude": True,
"show_phase": False,
}

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

@ -156,6 +156,46 @@ class ProcessorManager:
logger.error("Recalculation error", processor_id=processor_id, error=repr(exc))
raise
def load_processor_history(self, processor_id: str, history_data: list[dict[str, Any]]) -> ProcessedResult | None:
"""
Load sweep history into a processor from JSON data and recalculate.
Parameters
----------
processor_id : str
The processor to load history into.
history_data : list[dict]
History records in the format exported by export_history_data.
Returns
-------
ProcessedResult | None
The result of recalculation after loading history.
"""
processor = self.get_processor(processor_id)
if not processor:
raise ValueError(f"Processor {processor_id} not found")
try:
processor.import_history_data(history_data)
result = processor.recalculate()
if result:
with self._lock:
callbacks = list(self._result_callbacks)
for cb in callbacks:
try:
cb(processor_id, result)
except Exception as exc: # noqa: BLE001
logger.error("Result callback failed", processor_id=processor_id, error=repr(exc))
logger.info("History loaded and recalculated", processor_id=processor_id, records=len(history_data))
return result
except Exception as exc: # noqa: BLE001
logger.error("History load error", processor_id=processor_id, error=repr(exc))
raise
# --------------------------------------------------------------------- #
# Runtime control
# --------------------------------------------------------------------- #

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

@ -100,6 +100,8 @@ class ProcessorWebSocketHandler:
await self._handle_recalculate(websocket, message)
elif mtype == "get_history":
await self._handle_get_history(websocket, message)
elif mtype == "load_history":
await self._handle_load_history(websocket, message)
else:
await self._send_error(websocket, f"Unknown message type: {mtype!r}")
except json.JSONDecodeError as json_error:
@ -170,6 +172,31 @@ class ProcessorWebSocketHandler:
logger.error("Error getting history")
await self._send_error(websocket, f"Error getting history: {exc}")
async def _handle_load_history(self, websocket: WebSocket, message: dict[str, Any]) -> None:
"""
Load sweep history from JSON data into a processor and recalculate.
"""
processor_id = message.get("processor_id")
history_data = message.get("history_data")
if not processor_id:
await self._send_error(websocket, "processor_id is required")
return
if not history_data or not isinstance(history_data, list):
await self._send_error(websocket, "history_data (list) is required")
return
try:
result = self.processor_manager.load_processor_history(processor_id, history_data)
if result:
await websocket.send_text(json.dumps(self._result_to_message(processor_id, result)))
else:
await self._send_error(websocket, f"No result from processor {processor_id} after loading history")
except Exception as exc: # noqa: BLE001
logger.error("History load failed", processor_id=processor_id, error=repr(exc))
await self._send_error(websocket, f"History load failed: {exc}")
# --------------------------------------------------------------------- #
# Outbound helpers
# --------------------------------------------------------------------- #

26
vna_system/web_ui/static/css/layout.css
View File

@ -107,13 +107,18 @@ body {
}
.header__status {
flex: 1;
display: flex;
align-items: center;
justify-content: center;
gap: var(--space-6);
}
.header__controls {
display: flex;
align-items: center;
gap: var(--space-2);
}
.header__stats {
display: flex;
gap: var(--space-4);
@ -387,9 +392,21 @@ body {
}
}
@media (max-width: 1200px) {
.header__controls .btn__text {
display: none;
}
.header__controls .btn {
min-width: auto;
padding: var(--space-2);
}
}
@media (max-width: 768px) {
.header__container {
padding: 0 var(--space-4);
flex-wrap: wrap;
}
.header__brand {
@ -402,6 +419,13 @@ body {
flex-direction: column;
}
.header__controls {
order: 3;
width: 100%;
justify-content: center;
padding-top: var(--space-2);
}
.main {
padding: var(--space-4);
}

6
vna_system/web_ui/static/js/modules/charts.js
View File

@ -319,6 +319,9 @@ export class ChartManager {
if (!chart || !latestData) return null;
// Extract sweep_history from metadata if available
const sweepHistory = latestData.metadata?.sweep_history || [];
return {
processor_info: {
processor_id: processorId,
@ -330,7 +333,8 @@ export class ChartManager {
metadata: safeClone(latestData.metadata),
timestamp: latestData.timestamp instanceof Date ? latestData.timestamp.toISOString() : latestData.timestamp,
plotly_config: safeClone(latestData.plotly_config)
}
},
sweep_history: sweepHistory
};
}

79
vna_system/web_ui/static/js/modules/charts/chart-settings.js
View File

@ -63,11 +63,22 @@ export class ChartSettingsManager {
this.updateParametersSelectively(processorId, settingsContainer, uiParameters);
} else {
// Initial render: full rebuild
const settingsHtml = uiParameters.map(param =>
const loadHistoryButton = `
<div class="chart-setting" style="border-top: 1px solid var(--border-primary); padding-top: var(--space-3); margin-bottom: var(--space-3);">
<button class="btn btn--secondary btn--sm" id="loadHistoryBtn_${processorId}" style="width: 100%;">
<span data-icon="upload"></span>
<span>Загрузить историю</span>
</button>
<input type="file" id="historyFileInput_${processorId}" accept=".json" style="display: none;">
</div>
`;
const settingsHtml = loadHistoryButton + uiParameters.map(param =>
createParameterControl(param, processorId, 'chart')
).join('');
settingsContainer.innerHTML = settingsHtml;
this.setupEvents(settingsContainer, processorId);
this.setupLoadHistoryButton(processorId);
}
// Initialize last values
@ -282,6 +293,72 @@ export class ChartSettingsManager {
}, 1000);
}
setupLoadHistoryButton(processorId) {
const loadBtn = document.getElementById(`loadHistoryBtn_${processorId}`);
const fileInput = document.getElementById(`historyFileInput_${processorId}`);
if (!loadBtn || !fileInput) return;
loadBtn.addEventListener('click', () => {
fileInput.click();
});
fileInput.addEventListener('change', async (e) => {
const file = e.target.files?.[0];
if (!file) return;
try {
const text = await file.text();
const jsonData = JSON.parse(text);
// Extract sweep_history from the saved JSON file
const sweepHistory = jsonData.sweep_history || [];
if (!sweepHistory || sweepHistory.length === 0) {
window.vnaDashboard?.notifications?.show?.({
type: 'error',
title: 'Ошибка загрузки',
message: 'Файл не содержит истории свипов'
});
return;
}
// Send load_history message via WebSocket
const websocket = window.vnaDashboard?.websocket;
if (websocket && websocket.ws && websocket.ws.readyState === WebSocket.OPEN) {
websocket.ws.send(JSON.stringify({
type: 'load_history',
processor_id: processorId,
history_data: sweepHistory
}));
window.vnaDashboard?.notifications?.show?.({
type: 'success',
title: 'История загружена',
message: `Загружено ${sweepHistory.length} записей для ${processorId}`
});
} else {
window.vnaDashboard?.notifications?.show?.({
type: 'error',
title: 'Ошибка подключения',
message: 'WebSocket не подключен'
});
}
} catch (err) {
console.error('Error loading history:', err);
window.vnaDashboard?.notifications?.show?.({
type: 'error',
title: 'Ошибка загрузки',
message: `Не удалось прочитать файл: ${err.message}`
});
}
// Reset file input
fileInput.value = '';
});
}
destroy() {
// Clear timers
Object.keys(this.settingDebounceTimers).forEach(key => {

31
vna_system/web_ui/templates/index.html
View File

@ -42,6 +42,21 @@
</div>
</div>
<div class="header__controls">
<button class="btn btn--primary btn--sm" id="startBtn" title="Запустить непрерывный сбор">
<span data-icon="play"></span>
<span class="btn__text">Запуск</span>
</button>
<button class="btn btn--secondary btn--sm" id="stopBtn" title="Остановить сбор">
<span data-icon="square"></span>
<span class="btn__text">Стоп</span>
</button>
<button class="btn btn--accent btn--sm" id="singleSweepBtn" title="Запустить одиночный свип">
<span data-icon="zap"></span>
<span class="btn__text">Одиночный</span>
</button>
</div>
<nav class="header__nav">
<button class="nav-btn nav-btn--active" data-view="dashboard">
<span data-icon="bar-chart-3"></span>
@ -63,22 +78,8 @@
<div class="controls-panel">
<div class="controls-panel__container">
<div class="controls-group">
<label class="controls-label">Управление сбором</label>
<label class="controls-label">Информация о сборе</label>
<div class="acquisition-controls">
<div class="acquisition-controls__buttons">
<button class="btn btn--primary" id="startBtn" title="Запустить непрерывный сбор">
<span data-icon="play"></span>
Запуск
</button>
<button class="btn btn--secondary" id="stopBtn" title="Остановить сбор">
<span data-icon="square"></span>
Стоп
</button>
<button class="btn btn--accent" id="singleSweepBtn" title="Запустить одиночный свип">
<span data-icon="zap"></span>
Одиночный
</button>
</div>
<div class="acquisition-summary header__summary" id="headerSummary">
<div class="header-summary__item">
<span class="header-summary__label">Пресет</span>