some refactoring and socket server added
This commit is contained in:
Ayzen
@@ -0,0 +1,11 @@
|
||||
"""Plot-rendering mixins split by rendering mode."""
|
||||
|
||||
from python_app.gui.controllers.app_window_plot.bscan_plot_mixin import AppWindowBscanPlotMixin
|
||||
from python_app.gui.controllers.app_window_plot.gpr_plot_mixin import AppWindowGprPlotMixin
|
||||
from python_app.gui.controllers.app_window_plot.trace_plot_mixin import AppWindowTracePlotMixin
|
||||
|
||||
__all__ = [
|
||||
"AppWindowBscanPlotMixin",
|
||||
"AppWindowGprPlotMixin",
|
||||
"AppWindowTracePlotMixin",
|
||||
]
|
||||
@@ -0,0 +1,342 @@
|
||||
"""B-scan rendering and cache helpers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import deque
|
||||
|
||||
from PyQt6.QtCore import QRectF
|
||||
import numpy as np
|
||||
import pyqtgraph as pg
|
||||
|
||||
from python_app.models.dataset_model import ResultCollection
|
||||
from python_app.orchestration.live_processing_config import ProcessingLiveConfig
|
||||
|
||||
|
||||
def _result_tail(
|
||||
*,
|
||||
result_history: list[ResultCollection],
|
||||
history_limit: int,
|
||||
floor_collection_id: int,
|
||||
) -> list[ResultCollection]:
|
||||
"""Return filtered and de-duplicated result-history tail for B-scan usage."""
|
||||
filtered = [
|
||||
collection
|
||||
for collection in result_history[-history_limit:]
|
||||
if int(collection.collection_id) > int(floor_collection_id)
|
||||
]
|
||||
unique_reversed_tail: list[ResultCollection] = []
|
||||
seen_keys: set[tuple[int, int]] = set()
|
||||
for collection in reversed(filtered):
|
||||
key = (int(collection.collection_id), int(collection.monotonic_ns))
|
||||
if key in seen_keys:
|
||||
continue
|
||||
seen_keys.add(key)
|
||||
unique_reversed_tail.append(collection)
|
||||
|
||||
unique_reversed_tail.reverse()
|
||||
return unique_reversed_tail
|
||||
|
||||
|
||||
def build_bscan_signature(
|
||||
live_config: ProcessingLiveConfig,
|
||||
result_history: list[ResultCollection],
|
||||
history_limit: int,
|
||||
floor_collection_id: int,
|
||||
) -> tuple[object, ...]:
|
||||
"""Build deterministic signature used to detect B-scan cache invalidation."""
|
||||
result_tail = _result_tail(
|
||||
result_history=result_history,
|
||||
history_limit=history_limit,
|
||||
floor_collection_id=floor_collection_id,
|
||||
)
|
||||
return (
|
||||
str(live_config.bscan_axis),
|
||||
str(live_config.bscan_channel),
|
||||
float(live_config.bscan_cut_m),
|
||||
float(live_config.bscan_max_depth_m),
|
||||
float(live_config.bscan_gain),
|
||||
float(live_config.bscan_start_freq_mhz),
|
||||
float(live_config.bscan_stop_freq_mhz),
|
||||
int(floor_collection_id),
|
||||
tuple((int(collection.collection_id), int(collection.monotonic_ns), len(collection.blocks)) for collection in result_tail),
|
||||
)
|
||||
|
||||
|
||||
def rebuild_bscan_history_from_results(
|
||||
result_history: list[ResultCollection],
|
||||
history_limit: int,
|
||||
floor_collection_id: int,
|
||||
) -> tuple[dict[tuple[int, int], deque[np.ndarray]], dict[tuple[int, int], np.ndarray]]:
|
||||
"""Rebuild B-scan history and depth axes from processed result payloads."""
|
||||
history_by_combo: dict[tuple[int, int], deque[np.ndarray]] = {}
|
||||
depth_axis_by_combo: dict[tuple[int, int], np.ndarray] = {}
|
||||
|
||||
result_tail = _result_tail(
|
||||
result_history=result_history,
|
||||
history_limit=history_limit,
|
||||
floor_collection_id=floor_collection_id,
|
||||
)
|
||||
|
||||
for collection in result_tail:
|
||||
for block in collection.blocks:
|
||||
key = (block.combo.input_pos, block.combo.output_pos)
|
||||
for payload in block.payloads:
|
||||
if payload.kind != 1 or payload.processing_name != "bscan":
|
||||
continue
|
||||
if payload.frequency_hz.size == 0 or payload.trace.size == 0:
|
||||
continue
|
||||
if payload.frequency_hz.size != payload.trace.size:
|
||||
continue
|
||||
|
||||
depth_axis = np.asarray(payload.frequency_hz, dtype=np.float32)
|
||||
amplitudes = np.asarray(np.real(payload.trace), dtype=np.float32)
|
||||
if depth_axis.size == 0 or amplitudes.size == 0:
|
||||
continue
|
||||
|
||||
history = history_by_combo.get(key)
|
||||
stored_axis = depth_axis_by_combo.get(key)
|
||||
if (
|
||||
history is None
|
||||
or stored_axis is None
|
||||
or stored_axis.shape != depth_axis.shape
|
||||
or not np.allclose(stored_axis, depth_axis, rtol=1e-4, atol=1e-6)
|
||||
):
|
||||
history = deque(maxlen=history_limit)
|
||||
history_by_combo[key] = history
|
||||
depth_axis_by_combo[key] = depth_axis.copy()
|
||||
|
||||
history.append(amplitudes.copy())
|
||||
|
||||
return history_by_combo, depth_axis_by_combo
|
||||
|
||||
|
||||
def pick_bscan_display_key(
|
||||
history_by_combo: dict[tuple[int, int], deque[np.ndarray]],
|
||||
) -> tuple[int, int] | None:
|
||||
"""Choose combo key to display when multiple histories are present."""
|
||||
if not history_by_combo:
|
||||
return None
|
||||
return next(iter(history_by_combo.keys()))
|
||||
|
||||
|
||||
def bscan_lookup_table(axis_mode: str) -> np.ndarray:
|
||||
"""Build B-scan colormap table for selected axis mode."""
|
||||
if axis_mode == "abs":
|
||||
return build_lut(["#440154", "#31688e", "#35b779", "#fde725"])
|
||||
return build_lut(["#2166ac", "#67a9cf", "#f7f7f7", "#ef8a62", "#b2182b"])
|
||||
|
||||
|
||||
def build_lut(stops: list[str], *, size: int = 256) -> np.ndarray:
|
||||
"""Interpolate hex color stops into 8-bit RGB LUT array."""
|
||||
stop_positions = np.linspace(0.0, 1.0, num=len(stops), dtype=np.float32)
|
||||
sample_positions = np.linspace(0.0, 1.0, num=size, dtype=np.float32)
|
||||
stop_colors = np.asarray([pg.mkColor(value).getRgb()[:3] for value in stops], dtype=np.float32)
|
||||
|
||||
lut = np.empty((size, 3), dtype=np.uint8)
|
||||
for channel in range(3):
|
||||
lut[:, channel] = np.interp(sample_positions, stop_positions, stop_colors[:, channel]).astype(np.uint8)
|
||||
return lut
|
||||
|
||||
|
||||
def bscan_levels(sweeps: np.ndarray, axis_mode: str) -> tuple[float, float]:
|
||||
"""Compute image levels for B-scan data based on axis mode."""
|
||||
min_value = float(np.min(sweeps))
|
||||
max_value = float(np.max(sweeps))
|
||||
if axis_mode == "abs":
|
||||
if max_value <= min_value:
|
||||
return min_value, min_value + 1e-6
|
||||
return min_value, max_value
|
||||
|
||||
max_abs = max(abs(min_value), abs(max_value), 1e-6)
|
||||
return -max_abs, max_abs
|
||||
|
||||
|
||||
class AppWindowBscanPlotMixin:
|
||||
"""Renders B-scan heatmaps and maintains B-scan history caches."""
|
||||
|
||||
def _draw_bscan_heatmap(self, _collection: ResultCollection) -> bool:
|
||||
"""Draw B-scan image rebuilt from processed result history."""
|
||||
self._disable_phase_axis()
|
||||
self._sync_bscan_history_from_results()
|
||||
return self._draw_bscan_heatmap_from_history()
|
||||
|
||||
def _configure_bscan_plot_axes(self) -> None:
|
||||
"""Apply persistent B-scan plot axis labels and base view settings."""
|
||||
plot = self._bscan_plot
|
||||
plot_item = plot.getPlotItem()
|
||||
plot_item.showAxis("left", show=True)
|
||||
plot_item.showAxis("bottom", show=True)
|
||||
plot.setLabel("bottom", "Sweep #")
|
||||
plot.setLabel("left", "Range", units="m")
|
||||
view_box = plot.getViewBox()
|
||||
view_box.invertY(False)
|
||||
view_box.enableAutoRange(x=False, y=False)
|
||||
|
||||
def _draw_bscan_heatmap_from_history(self) -> bool:
|
||||
"""Render B-scan heatmap from currently cached history arrays."""
|
||||
display_key = self._pick_bscan_display_key()
|
||||
if display_key is None:
|
||||
return False
|
||||
|
||||
history = self._bscan_history_by_combo.get(display_key)
|
||||
depth_axis = self._bscan_depth_axis_by_combo.get(display_key)
|
||||
if not history or depth_axis is None:
|
||||
return False
|
||||
|
||||
sweeps = np.vstack(history).astype(np.float32, copy=False)
|
||||
if sweeps.size == 0:
|
||||
return False
|
||||
|
||||
depth_min = float(np.min(depth_axis))
|
||||
depth_max = float(np.max(depth_axis))
|
||||
depth_span = max(depth_max - depth_min, 1e-6)
|
||||
sweep_count = sweeps.shape[0]
|
||||
sweep_width = float(max(sweep_count, 1))
|
||||
x_min = 0.5
|
||||
x_max = x_min + sweep_width
|
||||
|
||||
image_item = pg.ImageItem(axisOrder="row-major")
|
||||
image_item.setImage(sweeps.T, autoLevels=False)
|
||||
image_item.setRect(QRectF(x_min, depth_min, sweep_width, depth_span))
|
||||
|
||||
axis_mode = self._bscan_axis.currentText()
|
||||
image_item.setLookupTable(self._bscan_lookup_table(axis_mode))
|
||||
image_item.setLevels(self._bscan_levels(sweeps, axis_mode))
|
||||
|
||||
self._bscan_plot.clear()
|
||||
self._configure_bscan_plot_axes()
|
||||
self._bscan_plot.addItem(image_item)
|
||||
self._bscan_plot.setXRange(x_min, x_max, padding=0.02)
|
||||
self._bscan_plot.setYRange(depth_min, depth_max, padding=0.02)
|
||||
bscan_channel = "S21"
|
||||
self._bscan_plot.setTitle(
|
||||
f"B-scan {bscan_channel} in{display_key[0]}/out{display_key[1]} | sweeps={sweep_count}"
|
||||
)
|
||||
return True
|
||||
|
||||
def _sync_bscan_history_from_results(self) -> None:
|
||||
"""Rebuild B-scan history cache when live params or inputs changed."""
|
||||
self._advance_bscan_floor_to_cpp_window()
|
||||
signature = self._bscan_signature()
|
||||
if signature == self._bscan_render_signature:
|
||||
return
|
||||
self._rebuild_bscan_history_from_results()
|
||||
self._bscan_render_signature = signature
|
||||
|
||||
def _bscan_signature(self) -> tuple[object, ...]:
|
||||
"""Build state signature for B-scan history cache invalidation."""
|
||||
live_config = self._live_processing_config()
|
||||
result_history = list(self._result_history)
|
||||
return build_bscan_signature(
|
||||
live_config=live_config,
|
||||
result_history=result_history,
|
||||
history_limit=self._bscan_history_limit,
|
||||
floor_collection_id=self._bscan_history_floor_collection_id,
|
||||
)
|
||||
|
||||
def _rebuild_bscan_history_from_results(self) -> None:
|
||||
"""Recompute B-scan history cache from results history buffer."""
|
||||
result_history = list(self._result_history)
|
||||
history_by_combo, depth_axis_by_combo = rebuild_bscan_history_from_results(
|
||||
result_history=result_history,
|
||||
history_limit=self._bscan_history_limit,
|
||||
floor_collection_id=self._bscan_history_floor_collection_id,
|
||||
)
|
||||
self._bscan_history_by_combo = history_by_combo
|
||||
self._bscan_depth_axis_by_combo = depth_axis_by_combo
|
||||
|
||||
def _pick_bscan_display_key(self) -> tuple[int, int] | None:
|
||||
"""Choose combo history key to render."""
|
||||
display_key = pick_bscan_display_key(self._bscan_history_by_combo)
|
||||
available_keys = sorted(self._bscan_history_by_combo.keys())
|
||||
if display_key is not None and len(available_keys) > 1:
|
||||
combo_signature = ",".join(f"{input_pos}:{output_pos}" for input_pos, output_pos in available_keys)
|
||||
details = "\n".join(
|
||||
f"- in{input_pos}/out{output_pos}"
|
||||
for input_pos, output_pos in available_keys
|
||||
)
|
||||
self._log(
|
||||
f"B-scan auto-selected combo in{display_key[0]}/out{display_key[1]} because multiple combos are available.",
|
||||
once_key=f"bscan_auto_display_{combo_signature}",
|
||||
)
|
||||
self._log_warning(
|
||||
"B-scan has multiple combo histories but the UI currently renders only one at a time.",
|
||||
details=details,
|
||||
once_key=f"bscan_multi_combo_warning_{combo_signature}",
|
||||
)
|
||||
return display_key
|
||||
|
||||
def _bscan_lookup_table(self, axis_mode: str) -> np.ndarray:
|
||||
"""Return lookup table for current B-scan axis mode."""
|
||||
return bscan_lookup_table(axis_mode)
|
||||
|
||||
@staticmethod
|
||||
def _build_lut(stops: list[str], *, size: int = 256) -> np.ndarray:
|
||||
"""Backward-compatible wrapper around LUT builder."""
|
||||
return build_lut(stops, size=size)
|
||||
|
||||
@staticmethod
|
||||
def _bscan_levels(sweeps: np.ndarray, axis_mode: str) -> tuple[float, float]:
|
||||
"""Return display levels for B-scan image."""
|
||||
return bscan_levels(sweeps, axis_mode)
|
||||
|
||||
def _clear_bscan_plot_history(self) -> None:
|
||||
"""Drop cached B-scan history and invalidate cache signature."""
|
||||
self._bscan_history_by_combo.clear()
|
||||
self._bscan_depth_axis_by_combo.clear()
|
||||
self._bscan_render_signature = None
|
||||
|
||||
def _advance_bscan_floor_to_cpp_window(self) -> None:
|
||||
"""Clamp B-scan source history to C++ available replay window."""
|
||||
if not self._result_history:
|
||||
return
|
||||
|
||||
cpp_window_limit = min(
|
||||
int(self._defaults_config.rings.preprocessed.capacity),
|
||||
int(self._defaults_config.rings.results.capacity),
|
||||
)
|
||||
cpp_window_limit = max(1, cpp_window_limit)
|
||||
latest_collection_id = int(self._result_history[-1].collection_id)
|
||||
current_floor = int(self._bscan_history_floor_collection_id)
|
||||
|
||||
# Collection ids restart from 1 on new C++ run; release floor only while
|
||||
# acquisition is running, so manual "remove last" behavior in stopped mode
|
||||
# remains deterministic.
|
||||
if latest_collection_id < current_floor and self._supervisor.is_running():
|
||||
self._bscan_history_floor_collection_id = 0
|
||||
current_floor = 0
|
||||
|
||||
floor_candidate = max(0, latest_collection_id - cpp_window_limit)
|
||||
if floor_candidate > current_floor:
|
||||
self._bscan_history_floor_collection_id = floor_candidate
|
||||
|
||||
def _ensure_phase_view_box(self) -> pg.ViewBox:
|
||||
"""Create or return secondary right-axis ViewBox for phase curves."""
|
||||
plot_item = self._bscan_plot.getPlotItem()
|
||||
phase_view_box = self._phase_viewbox
|
||||
if phase_view_box is None:
|
||||
phase_view_box = pg.ViewBox()
|
||||
self._phase_viewbox = phase_view_box
|
||||
plot_item.scene().addItem(phase_view_box)
|
||||
plot_item.getAxis("right").linkToView(phase_view_box)
|
||||
phase_view_box.setXLink(plot_item.vb)
|
||||
plot_item.vb.sigResized.connect(self._update_phase_view_box_geometry)
|
||||
self._update_phase_view_box_geometry()
|
||||
return phase_view_box
|
||||
|
||||
def _update_phase_view_box_geometry(self) -> None:
|
||||
"""Keep right-axis ViewBox geometry in sync with main plot ViewBox."""
|
||||
phase_view_box = self._phase_viewbox
|
||||
if phase_view_box is None:
|
||||
return
|
||||
plot_item = self._bscan_plot.getPlotItem()
|
||||
phase_view_box.setGeometry(plot_item.vb.sceneBoundingRect())
|
||||
phase_view_box.linkedViewChanged(plot_item.vb, phase_view_box.XAxis)
|
||||
|
||||
def _clear_phase_overlay(self) -> None:
|
||||
"""Remove all phase curves from secondary ViewBox."""
|
||||
self._trace_phase_plot.clear()
|
||||
|
||||
def _disable_phase_axis(self) -> None:
|
||||
"""Hide right axis and clear phase overlay when phase is not rendered."""
|
||||
self._clear_phase_overlay()
|
||||
@@ -0,0 +1,503 @@
|
||||
"""GPR heatmap and object-only rendering helpers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from PyQt6.QtCore import QRectF
|
||||
import numpy as np
|
||||
import pyqtgraph as pg
|
||||
|
||||
from python_app.models.dataset_model import ResultCollection
|
||||
from python_app.orchestration.gpr_locator import (
|
||||
collection_payload_by_name as gpr_collection_payload_by_name,
|
||||
collection_payloads_by_prefix as gpr_collection_payloads_by_prefix,
|
||||
gpr_object_rows as extract_gpr_object_rows,
|
||||
)
|
||||
|
||||
|
||||
class AppWindowGprPlotMixin:
|
||||
"""Renders GPR accumulator heatmaps and detected object overlays."""
|
||||
|
||||
def _clear_gpr_plot(self) -> None:
|
||||
"""Clear latest GPR plot surface."""
|
||||
if not hasattr(self, "_gpr_plot"):
|
||||
return
|
||||
self._configure_gpr_plot_axes()
|
||||
self._clear_gpr_point_labels()
|
||||
self._clear_gpr_region_labels()
|
||||
self._clear_gpr_region_masks()
|
||||
if self._gpr_image_item is not None:
|
||||
self._gpr_image_item.hide()
|
||||
if self._gpr_tx_item is not None:
|
||||
self._gpr_tx_item.setData(x=[], y=[])
|
||||
self._gpr_tx_item.hide()
|
||||
if self._gpr_rx_item is not None:
|
||||
self._gpr_rx_item.setData(x=[], y=[])
|
||||
self._gpr_rx_item.hide()
|
||||
if self._gpr_points_item is not None:
|
||||
self._gpr_points_item.setData(x=[], y=[])
|
||||
self._gpr_points_item.hide()
|
||||
if self._gpr_region_centers_item is not None:
|
||||
self._gpr_region_centers_item.setData(x=[], y=[])
|
||||
self._gpr_region_centers_item.hide()
|
||||
self._gpr_plot.setTitle(f"GPR {self._gpr_config_mode.currentText()}")
|
||||
|
||||
def _configure_gpr_plot_axes(self) -> None:
|
||||
"""Apply persistent GPR plot axis labels and base view settings."""
|
||||
plot = self._gpr_plot
|
||||
plot_item = plot.getPlotItem()
|
||||
plot_item.showAxis("left", show=True)
|
||||
plot_item.showAxis("bottom", show=True)
|
||||
plot_item.setClipToView(True)
|
||||
plot.setLabel("bottom", "X", units="m")
|
||||
plot.setLabel("left", "Range", units="m")
|
||||
view_box = plot.getViewBox()
|
||||
view_box.invertY(False)
|
||||
view_box.enableAutoRange(x=False, y=False)
|
||||
|
||||
def _ensure_gpr_plot_items(self) -> None:
|
||||
"""Create persistent GPR plot items once and reuse them on redraw."""
|
||||
if self._gpr_image_item is not None:
|
||||
return
|
||||
|
||||
plot = self._gpr_plot
|
||||
self._configure_gpr_plot_axes()
|
||||
|
||||
if self._gpr_lookup_table is None:
|
||||
self._gpr_lookup_table = self._build_lut(["#081c15", "#1b4332", "#ffd166", "#f94144"])
|
||||
|
||||
self._gpr_image_item = pg.ImageItem(axisOrder="row-major")
|
||||
self._gpr_image_item.setZValue(0)
|
||||
self._gpr_image_item.hide()
|
||||
plot.addItem(self._gpr_image_item)
|
||||
|
||||
self._gpr_tx_item = pg.ScatterPlotItem()
|
||||
self._gpr_tx_item.setZValue(20)
|
||||
self._gpr_tx_item.hide()
|
||||
plot.addItem(self._gpr_tx_item)
|
||||
|
||||
self._gpr_rx_item = pg.ScatterPlotItem()
|
||||
self._gpr_rx_item.setZValue(20)
|
||||
self._gpr_rx_item.hide()
|
||||
plot.addItem(self._gpr_rx_item)
|
||||
|
||||
self._gpr_points_item = pg.ScatterPlotItem()
|
||||
self._gpr_points_item.setZValue(30)
|
||||
self._gpr_points_item.hide()
|
||||
plot.addItem(self._gpr_points_item)
|
||||
|
||||
self._gpr_region_centers_item = pg.ScatterPlotItem()
|
||||
self._gpr_region_centers_item.setZValue(30)
|
||||
self._gpr_region_centers_item.hide()
|
||||
plot.addItem(self._gpr_region_centers_item)
|
||||
|
||||
def _clear_gpr_point_labels(self) -> None:
|
||||
"""Remove dynamic point-score labels from GPR plot."""
|
||||
for item in self._gpr_point_labels:
|
||||
try:
|
||||
self._gpr_plot.removeItem(item)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
self._log_warning(
|
||||
f"Failed to remove GPR point label: {type(exc).__name__}: {exc}",
|
||||
once_key="gpr_remove_point_label_failed",
|
||||
)
|
||||
self._gpr_point_labels.clear()
|
||||
|
||||
def _clear_gpr_region_labels(self) -> None:
|
||||
"""Remove dynamic region labels from GPR plot."""
|
||||
for item in self._gpr_region_center_labels:
|
||||
try:
|
||||
self._gpr_plot.removeItem(item)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
self._log_warning(
|
||||
f"Failed to remove GPR region label: {type(exc).__name__}: {exc}",
|
||||
once_key="gpr_remove_region_label_failed",
|
||||
)
|
||||
self._gpr_region_center_labels.clear()
|
||||
|
||||
def _clear_gpr_region_masks(self) -> None:
|
||||
"""Remove dynamic region contour carriers from GPR plot."""
|
||||
for item in self._gpr_region_mask_items:
|
||||
try:
|
||||
self._gpr_plot.removeItem(item)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
self._log_warning(
|
||||
f"Failed to remove GPR region mask: {type(exc).__name__}: {exc}",
|
||||
once_key="gpr_remove_region_mask_failed",
|
||||
)
|
||||
self._gpr_region_mask_items.clear()
|
||||
self._gpr_region_contours.clear()
|
||||
|
||||
@staticmethod
|
||||
def _collection_payload_by_name(collection: ResultCollection, name: str, kind: int | None = None):
|
||||
"""Return first collection payload matching name and optional kind."""
|
||||
return gpr_collection_payload_by_name(collection, name, kind)
|
||||
|
||||
@staticmethod
|
||||
def _collection_payloads_by_prefix(collection: ResultCollection, prefix: str, kind: int | None = None):
|
||||
"""Return collection payloads matching processing-name prefix."""
|
||||
return gpr_collection_payloads_by_prefix(collection, prefix, kind)
|
||||
|
||||
def _selected_gpr_geometry(self) -> tuple[np.ndarray, np.ndarray]:
|
||||
"""Resolve selected Tx/Rx geometry arrays for current GPR selection."""
|
||||
requested_inputs = tuple(self._parse_csv_int_list(self._gpr_input_positions_input.text()))
|
||||
requested_outputs = tuple(self._parse_csv_int_list(self._gpr_output_positions_input.text()))
|
||||
signature = (
|
||||
self._gpr_tx_geometry_input.toPlainText(),
|
||||
self._gpr_rx_geometry_input.toPlainText(),
|
||||
requested_inputs,
|
||||
requested_outputs,
|
||||
)
|
||||
if signature == self._gpr_geometry_signature and self._gpr_selected_geometry is not None:
|
||||
return self._gpr_selected_geometry
|
||||
|
||||
tx_entries = self._parse_gpr_tx_geometry_text(signature[0])
|
||||
rx_entries = self._parse_gpr_rx_geometry_text(signature[1])
|
||||
requested_input_set = set(requested_inputs)
|
||||
requested_output_set = set(requested_outputs)
|
||||
|
||||
rx_entries = sorted(rx_entries, key=lambda entry: int(entry.input_pos))
|
||||
tx_entries = sorted(tx_entries, key=lambda entry: int(entry.output_pos))
|
||||
if requested_input_set:
|
||||
rx_entries = [entry for entry in rx_entries if int(entry.input_pos) in requested_input_set]
|
||||
if requested_output_set:
|
||||
tx_entries = [entry for entry in tx_entries if int(entry.output_pos) in requested_output_set]
|
||||
|
||||
x_tx = np.asarray([float(entry.x_m) for entry in tx_entries], dtype=np.float32)
|
||||
x_rx = np.asarray([float(entry.x_m) for entry in rx_entries], dtype=np.float32)
|
||||
self._gpr_geometry_signature = signature
|
||||
self._gpr_selected_geometry = (x_tx, x_rx)
|
||||
return self._gpr_selected_geometry
|
||||
|
||||
def _draw_gpr_map(self, collection: ResultCollection) -> bool:
|
||||
"""Draw latest collection-level GPR plot according to current render mode."""
|
||||
if self._gpr_render_mode.currentText() == "objects_only":
|
||||
return self._draw_gpr_objects_only(collection)
|
||||
return self._draw_gpr_heatmap(collection)
|
||||
|
||||
def _draw_gpr_heatmap(self, collection: ResultCollection) -> bool:
|
||||
"""Draw latest collection-level GPR accumulator and annotations."""
|
||||
accumulator_payload = self._collection_payload_by_name(collection, "gpr_accumulator", kind=3)
|
||||
if accumulator_payload is None:
|
||||
self._clear_gpr_plot()
|
||||
return False
|
||||
|
||||
image = np.asarray(accumulator_payload.image, dtype=np.float32)
|
||||
x_axis = np.asarray(accumulator_payload.image_x_axis, dtype=np.float32)
|
||||
y_axis = np.asarray(accumulator_payload.image_y_axis, dtype=np.float32)
|
||||
if image.ndim != 2 or image.size == 0 or x_axis.size == 0 or y_axis.size == 0:
|
||||
self._clear_gpr_plot()
|
||||
return False
|
||||
|
||||
x_min = float(x_axis[0])
|
||||
x_max = float(x_axis[-1])
|
||||
y_min = float(y_axis[0])
|
||||
y_max = float(y_axis[-1])
|
||||
rect = QRectF(x_min, y_min, max(x_max - x_min, 1e-6), max(y_max - y_min, 1e-6))
|
||||
|
||||
plot = self._gpr_plot
|
||||
plot.setUpdatesEnabled(False)
|
||||
try:
|
||||
self._ensure_gpr_plot_items()
|
||||
plot.getViewBox().invertY(False)
|
||||
self._clear_gpr_point_labels()
|
||||
self._clear_gpr_region_labels()
|
||||
self._clear_gpr_region_masks()
|
||||
|
||||
self._gpr_image_item.setImage(image, autoLevels=False)
|
||||
self._gpr_image_item.setRect(rect)
|
||||
self._gpr_image_item.setLookupTable(self._gpr_lookup_table)
|
||||
self._gpr_image_item.setLevels((float(np.min(image)), float(np.max(image) + 1e-6)))
|
||||
self._gpr_image_item.show()
|
||||
|
||||
plot.setXRange(x_min, x_max, padding=0.02)
|
||||
plot.setYRange(self._gpr_display_y_min(y_min, y_max), y_max, padding=0.02)
|
||||
|
||||
self._draw_gpr_geometry_markers()
|
||||
|
||||
points_payload = self._collection_payload_by_name(collection, "gpr_points", kind=4)
|
||||
if points_payload is not None and np.asarray(points_payload.table).size > 0:
|
||||
points = np.asarray(points_payload.table, dtype=np.float32)
|
||||
self._gpr_points_item.setData(
|
||||
x=points[:, 0],
|
||||
y=points[:, 1],
|
||||
symbol="d",
|
||||
size=11,
|
||||
brush=pg.mkBrush("#ffffff"),
|
||||
pen=pg.mkPen("#111111", width=1.1),
|
||||
)
|
||||
self._gpr_points_item.show()
|
||||
for x_value, y_value, score in points:
|
||||
label = pg.TextItem(text=f"{float(score):.0f}", color="#ffffff", anchor=(0.0, 1.0))
|
||||
label.setZValue(40)
|
||||
label.setPos(float(x_value), float(y_value))
|
||||
plot.addItem(label)
|
||||
self._gpr_point_labels.append(label)
|
||||
else:
|
||||
self._gpr_points_item.setData(x=[], y=[])
|
||||
self._gpr_points_item.hide()
|
||||
|
||||
region_centers_payload = self._collection_payload_by_name(collection, "gpr_region_centers", kind=4)
|
||||
if region_centers_payload is not None and np.asarray(region_centers_payload.table).size > 0:
|
||||
centers = np.asarray(region_centers_payload.table, dtype=np.float32)
|
||||
self._gpr_region_centers_item.setData(
|
||||
x=centers[:, 0],
|
||||
y=centers[:, 1],
|
||||
symbol="o",
|
||||
size=10,
|
||||
brush=pg.mkBrush("#80ed99"),
|
||||
pen=pg.mkPen("#081c15", width=1.1),
|
||||
)
|
||||
self._gpr_region_centers_item.show()
|
||||
for row in centers:
|
||||
label = pg.TextItem(text=f"{float(row[2]):.0f}", color="#d8f3dc", anchor=(0.0, 1.0))
|
||||
label.setZValue(40)
|
||||
label.setPos(float(row[0]), float(row[1]))
|
||||
plot.addItem(label)
|
||||
self._gpr_region_center_labels.append(label)
|
||||
else:
|
||||
self._gpr_region_centers_item.setData(x=[], y=[])
|
||||
self._gpr_region_centers_item.hide()
|
||||
|
||||
for payload in self._collection_payloads_by_prefix(collection, "gpr_region_mask_", kind=3):
|
||||
mask = np.asarray(payload.image, dtype=np.float32)
|
||||
if mask.ndim != 2 or mask.size == 0:
|
||||
continue
|
||||
mask_image = pg.ImageItem(axisOrder="row-major")
|
||||
mask_image.setZValue(5)
|
||||
mask_image.setImage(mask, autoLevels=False)
|
||||
mask_image.setRect(rect)
|
||||
mask_image.setOpacity(0.0)
|
||||
plot.addItem(mask_image)
|
||||
contour = pg.IsocurveItem(data=mask, level=0.5, pen=pg.mkPen("#4cc9f0", width=1.3))
|
||||
contour.setParentItem(mask_image)
|
||||
self._gpr_region_mask_items.append(mask_image)
|
||||
self._gpr_region_contours.append(contour)
|
||||
|
||||
plot.setTitle(f"GPR {self._gpr_config_mode.currentText()}")
|
||||
finally:
|
||||
plot.setUpdatesEnabled(True)
|
||||
return True
|
||||
|
||||
@staticmethod
|
||||
def _normalized_display_range(start: float, stop: float, *, minimum_span: float = 0.1) -> tuple[float, float]:
|
||||
"""Return ordered display bounds with a non-zero span."""
|
||||
lower = min(float(start), float(stop))
|
||||
upper = max(float(start), float(stop))
|
||||
if upper - lower >= minimum_span:
|
||||
return lower, upper
|
||||
center = 0.5 * (lower + upper)
|
||||
half_span = 0.5 * minimum_span
|
||||
return center - half_span, center + half_span
|
||||
|
||||
def _gpr_visible_object_bounds(self) -> tuple[float, float, float, float]:
|
||||
"""Return normalized object-only visible X/Z bounds from GUI controls."""
|
||||
x_min, x_max = self._normalized_display_range(
|
||||
float(self._gpr_visible_x_min_m.value()),
|
||||
float(self._gpr_visible_x_max_m.value()),
|
||||
minimum_span=0.1,
|
||||
)
|
||||
z_min, z_max = self._normalized_display_range(
|
||||
float(self._gpr_visible_z_min_m.value()),
|
||||
float(self._gpr_visible_z_max_m.value()),
|
||||
minimum_span=0.1,
|
||||
)
|
||||
return x_min, x_max, z_min, z_max
|
||||
|
||||
@staticmethod
|
||||
def _gpr_display_y_min(z_min: float, z_max: float) -> float:
|
||||
"""Return lower GPR display bound with a small negative margin for antenna markers."""
|
||||
lower = min(0.0, float(z_min))
|
||||
span = max(float(z_max) - float(z_min), 1e-6)
|
||||
marker_margin = max(span * 0.03, 0.06)
|
||||
return lower - marker_margin
|
||||
|
||||
def _draw_gpr_geometry_markers(self) -> None:
|
||||
"""Render selected Tx/Rx geometry markers on current GPR plot."""
|
||||
x_tx, x_rx = self._selected_gpr_geometry()
|
||||
if x_tx.size > 0:
|
||||
self._gpr_tx_item.setData(
|
||||
x=x_tx,
|
||||
y=np.zeros_like(x_tx),
|
||||
symbol="t",
|
||||
size=13,
|
||||
brush=pg.mkBrush("#ff595e"),
|
||||
pen=pg.mkPen("#ffca3a", width=1.0),
|
||||
)
|
||||
self._gpr_tx_item.show()
|
||||
else:
|
||||
self._gpr_tx_item.setData(x=[], y=[])
|
||||
self._gpr_tx_item.hide()
|
||||
|
||||
if x_rx.size > 0:
|
||||
self._gpr_rx_item.setData(
|
||||
x=x_rx,
|
||||
y=np.zeros_like(x_rx),
|
||||
symbol="t1",
|
||||
size=13,
|
||||
brush=pg.mkBrush("#4cc9f0"),
|
||||
pen=pg.mkPen("#e0fbfc", width=1.0),
|
||||
)
|
||||
self._gpr_rx_item.show()
|
||||
else:
|
||||
self._gpr_rx_item.setData(x=[], y=[])
|
||||
self._gpr_rx_item.hide()
|
||||
|
||||
@staticmethod
|
||||
def _format_gpr_object_label(x_m: float, z_m: float, pair_count: float) -> str:
|
||||
"""Format object-only annotation text with pair count and coordinates."""
|
||||
return f"{int(round(pair_count))} | x={x_m:.1f} | z={z_m:.1f}"
|
||||
|
||||
@staticmethod
|
||||
def _expanded_scene_rect(rect: QRectF, *, padding_px: float = 4.0) -> QRectF:
|
||||
"""Return scene rect padded to keep labels visually separated."""
|
||||
return rect.adjusted(-padding_px, -padding_px, padding_px, padding_px)
|
||||
|
||||
@staticmethod
|
||||
def _scene_rect_intersects_any(rect: QRectF, occupied_rects: list[QRectF]) -> bool:
|
||||
"""Return whether candidate label rect intersects any already placed label."""
|
||||
return any(rect.intersects(occupied_rect) for occupied_rect in occupied_rects)
|
||||
|
||||
@staticmethod
|
||||
def _gpr_object_label_candidates(
|
||||
x_m: float,
|
||||
z_m: float,
|
||||
*,
|
||||
x_span: float,
|
||||
z_span: float,
|
||||
) -> list[tuple[float, float, tuple[float, float]]]:
|
||||
"""Return candidate label placements around one object."""
|
||||
x_offset = max(x_span * 0.015, 0.02)
|
||||
z_offset = max(z_span * 0.02, 0.02)
|
||||
return [
|
||||
(x_m + x_offset, z_m - z_offset, (0.0, 1.0)),
|
||||
(x_m + x_offset, z_m + z_offset, (0.0, 0.0)),
|
||||
(x_m - x_offset, z_m - z_offset, (1.0, 1.0)),
|
||||
(x_m - x_offset, z_m + z_offset, (1.0, 0.0)),
|
||||
(x_m, z_m - (z_offset * 1.35), (0.5, 1.0)),
|
||||
(x_m, z_m + (z_offset * 1.35), (0.5, 0.0)),
|
||||
(x_m + (x_offset * 2.2), z_m - (z_offset * 1.5), (0.0, 1.0)),
|
||||
(x_m + (x_offset * 2.2), z_m + (z_offset * 1.5), (0.0, 0.0)),
|
||||
(x_m - (x_offset * 2.2), z_m - (z_offset * 1.5), (1.0, 1.0)),
|
||||
(x_m - (x_offset * 2.2), z_m + (z_offset * 1.5), (1.0, 0.0)),
|
||||
]
|
||||
|
||||
def _place_gpr_object_label(
|
||||
self,
|
||||
*,
|
||||
label: pg.TextItem,
|
||||
x_m: float,
|
||||
z_m: float,
|
||||
x_span: float,
|
||||
z_span: float,
|
||||
occupied_scene_rects: list[QRectF],
|
||||
) -> None:
|
||||
"""Place one object label using the first non-overlapping candidate position."""
|
||||
last_rect: QRectF | None = None
|
||||
for label_x, label_z, anchor in self._gpr_object_label_candidates(
|
||||
x_m,
|
||||
z_m,
|
||||
x_span=x_span,
|
||||
z_span=z_span,
|
||||
):
|
||||
label.setAnchor(anchor)
|
||||
label.setPos(label_x, label_z)
|
||||
candidate_rect = self._expanded_scene_rect(label.sceneBoundingRect())
|
||||
last_rect = candidate_rect
|
||||
if not self._scene_rect_intersects_any(candidate_rect, occupied_scene_rects):
|
||||
occupied_scene_rects.append(candidate_rect)
|
||||
return
|
||||
|
||||
if last_rect is not None:
|
||||
occupied_scene_rects.append(last_rect)
|
||||
|
||||
def _gpr_object_rows(self, collection: ResultCollection) -> np.ndarray:
|
||||
"""Return object rows as `[x_m, z_m, pair_count]` from current GPR result payload."""
|
||||
return extract_gpr_object_rows(collection)
|
||||
|
||||
def _filtered_gpr_object_rows(self, collection: ResultCollection) -> np.ndarray:
|
||||
"""Return object rows filtered by minimum pair count and visible X/Z bounds."""
|
||||
rows = self._gpr_object_rows(collection)
|
||||
if rows.size == 0:
|
||||
return rows
|
||||
|
||||
x_min, x_max, z_min, z_max = self._gpr_visible_object_bounds()
|
||||
min_pair_count = float(self._gpr_min_visible_pair_count.value())
|
||||
finite_mask = np.all(np.isfinite(rows[:, :3]), axis=1)
|
||||
visible_mask = (
|
||||
finite_mask
|
||||
& (rows[:, 2] >= min_pair_count)
|
||||
& (rows[:, 0] >= x_min)
|
||||
& (rows[:, 0] <= x_max)
|
||||
& (rows[:, 1] >= z_min)
|
||||
& (rows[:, 1] <= z_max)
|
||||
)
|
||||
return rows[visible_mask]
|
||||
|
||||
def _draw_gpr_objects_only(self, collection: ResultCollection) -> bool:
|
||||
"""Draw only detected GPR objects inside configured X/Z bounds."""
|
||||
accumulator_payload = self._collection_payload_by_name(collection, "gpr_accumulator", kind=3)
|
||||
object_rows = self._filtered_gpr_object_rows(collection)
|
||||
if accumulator_payload is None and object_rows.size == 0:
|
||||
self._clear_gpr_plot()
|
||||
return False
|
||||
|
||||
x_min, x_max, z_min, z_max = self._gpr_visible_object_bounds()
|
||||
plot = self._gpr_plot
|
||||
plot.setUpdatesEnabled(False)
|
||||
try:
|
||||
self._ensure_gpr_plot_items()
|
||||
plot.getViewBox().invertY(False)
|
||||
self._clear_gpr_point_labels()
|
||||
self._clear_gpr_region_labels()
|
||||
self._clear_gpr_region_masks()
|
||||
|
||||
self._gpr_image_item.hide()
|
||||
self._draw_gpr_geometry_markers()
|
||||
self._gpr_region_centers_item.setData(x=[], y=[])
|
||||
self._gpr_region_centers_item.hide()
|
||||
|
||||
if object_rows.size > 0:
|
||||
self._gpr_points_item.setData(
|
||||
x=object_rows[:, 0],
|
||||
y=object_rows[:, 1],
|
||||
symbol="o",
|
||||
size=18,
|
||||
brush=pg.mkBrush("#ff4d4f"),
|
||||
pen=pg.mkPen("#ff4d4f", width=1.6),
|
||||
)
|
||||
self._gpr_points_item.show()
|
||||
|
||||
occupied_scene_rects: list[QRectF] = []
|
||||
x_span = x_max - x_min
|
||||
z_span = z_max - z_min
|
||||
for x_value, z_value, pair_count in object_rows:
|
||||
label = pg.TextItem(
|
||||
text=self._format_gpr_object_label(
|
||||
float(x_value),
|
||||
float(z_value),
|
||||
float(pair_count),
|
||||
),
|
||||
color="#ffd6d9",
|
||||
anchor=(0.0, 1.0),
|
||||
)
|
||||
label.setZValue(40)
|
||||
plot.addItem(label)
|
||||
self._place_gpr_object_label(
|
||||
label=label,
|
||||
x_m=float(x_value),
|
||||
z_m=float(z_value),
|
||||
x_span=x_span,
|
||||
z_span=z_span,
|
||||
occupied_scene_rects=occupied_scene_rects,
|
||||
)
|
||||
self._gpr_point_labels.append(label)
|
||||
else:
|
||||
self._gpr_points_item.setData(x=[], y=[])
|
||||
self._gpr_points_item.hide()
|
||||
|
||||
plot.setXRange(x_min, x_max, padding=0.0)
|
||||
plot.setYRange(self._gpr_display_y_min(z_min, z_max), z_max, padding=0.0)
|
||||
plot.setTitle(f"GPR {self._gpr_config_mode.currentText()} Objects Only")
|
||||
finally:
|
||||
plot.setUpdatesEnabled(True)
|
||||
return True
|
||||
@@ -0,0 +1,388 @@
|
||||
"""Trace plot rendering helpers for pass-through and single-trace views."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from PyQt6.QtCore import Qt
|
||||
import numpy as np
|
||||
import pyqtgraph as pg
|
||||
|
||||
from python_app.models.dataset_model import ResultCollection, TraceData
|
||||
|
||||
|
||||
class AppWindowTracePlotMixin:
|
||||
"""Renders pass-through trace plots on stacked magnitude/phase widgets."""
|
||||
|
||||
def _show_magnitude_curves(self) -> bool:
|
||||
"""Return whether magnitude curves should be rendered."""
|
||||
return self._show_magnitude_checkbox.isChecked()
|
||||
|
||||
def _show_phase_curves(self) -> bool:
|
||||
"""Return whether phase curves should be rendered."""
|
||||
return self._show_phase_checkbox.isChecked()
|
||||
|
||||
def _pass_through_fixed_y_range(self) -> tuple[bool, float, float]:
|
||||
"""Return normalized magnitude Y-range override for pass-through mode."""
|
||||
y_min = float(self._pass_through_y_min_db.value())
|
||||
y_max = float(self._pass_through_y_max_db.value())
|
||||
return bool(self._pass_through_fixed_y_enabled.isChecked()), min(y_min, y_max), max(y_min, y_max)
|
||||
|
||||
def _configure_pass_through_magnitude_axis(self, plot: pg.PlotWidget) -> None:
|
||||
"""Apply pass-through magnitude-axis autorange or fixed Y window."""
|
||||
fixed_y_enabled, y_min, y_max = self._pass_through_fixed_y_range()
|
||||
view_box = plot.getViewBox()
|
||||
view_box.invertY(False)
|
||||
view_box.enableAutoRange(x=True, y=not fixed_y_enabled)
|
||||
if fixed_y_enabled:
|
||||
plot.setYRange(y_min, y_max, padding=0.0)
|
||||
|
||||
def _on_trace_visibility_changed(self, *_args) -> None:
|
||||
"""Redraw pass-through traces when magnitude/phase toggles changed."""
|
||||
if self._processing_mode.currentText() in {"bscan", "gpr"}:
|
||||
return
|
||||
if self._result_history:
|
||||
self._draw_results(self._result_history[-1])
|
||||
return
|
||||
self._clear_trace_plots()
|
||||
|
||||
def _clear_trace_plots(self) -> None:
|
||||
"""Clear pass-through magnitude and phase plots."""
|
||||
self._trace_magnitude_plot.clear()
|
||||
self._trace_phase_plot.clear()
|
||||
self._clear_trace_legends()
|
||||
self._trace_magnitude_curves.clear()
|
||||
self._trace_phase_curves.clear()
|
||||
|
||||
def _clear_trace_legends(self) -> None:
|
||||
"""Remove trace plot legends to avoid stale combo-color mappings."""
|
||||
mag_legend = self._trace_magnitude_legend
|
||||
if mag_legend is not None:
|
||||
try:
|
||||
self._trace_magnitude_plot.getPlotItem().removeItem(mag_legend)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
self._log_warning(
|
||||
f"Failed to remove pass-through magnitude legend: {type(exc).__name__}: {exc}",
|
||||
once_key="plot_remove_magnitude_legend_failed",
|
||||
)
|
||||
self._trace_magnitude_legend = None
|
||||
self._trace_magnitude_legend_combo_keys.clear()
|
||||
|
||||
phase_legend = self._trace_phase_legend
|
||||
if phase_legend is not None:
|
||||
try:
|
||||
self._trace_phase_plot.getPlotItem().removeItem(phase_legend)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
self._log_warning(
|
||||
f"Failed to remove pass-through phase legend: {type(exc).__name__}: {exc}",
|
||||
once_key="plot_remove_phase_legend_failed",
|
||||
)
|
||||
self._trace_phase_legend = None
|
||||
self._trace_phase_legend_combo_keys.clear()
|
||||
|
||||
def _draw_trace_lines(self, collection: ResultCollection) -> bool:
|
||||
"""Draw result payload traces as stacked magnitude/phase plots."""
|
||||
show_magnitude = self._show_magnitude_curves()
|
||||
show_phase = self._show_phase_curves()
|
||||
magnitude_plot = self._trace_magnitude_plot
|
||||
phase_plot = self._trace_phase_plot
|
||||
pass_through_channel = "S21"
|
||||
|
||||
magnitude_plot.setVisible(show_magnitude)
|
||||
phase_plot.setVisible(show_phase)
|
||||
if not show_magnitude and not show_phase:
|
||||
self._clear_trace_plots()
|
||||
return False
|
||||
|
||||
if show_magnitude:
|
||||
mag_item = magnitude_plot.getPlotItem()
|
||||
self._configure_pass_through_magnitude_axis(magnitude_plot)
|
||||
mag_item.showAxis("left", show=True)
|
||||
mag_item.showAxis("bottom", show=not show_phase)
|
||||
magnitude_plot.setLabel("left", "Magnitude", units="dB")
|
||||
magnitude_plot.setTitle(f"Pass-Through {pass_through_channel}")
|
||||
if not show_phase:
|
||||
magnitude_plot.setLabel("bottom", "Frequency", units="Hz")
|
||||
|
||||
if show_phase:
|
||||
phase_item = phase_plot.getPlotItem()
|
||||
phase_plot.getViewBox().invertY(False)
|
||||
phase_plot.getViewBox().enableAutoRange(x=True, y=False)
|
||||
phase_item.showAxis("left", show=True)
|
||||
phase_item.showAxis("bottom", show=True)
|
||||
phase_plot.setLabel("left", "Phase", units="deg")
|
||||
phase_plot.setLabel("bottom", "Frequency", units="Hz")
|
||||
phase_plot.setTitle(f"Pass-Through {pass_through_channel}")
|
||||
|
||||
palette = [
|
||||
"#4cc9f0",
|
||||
"#f72585",
|
||||
"#b8f2e6",
|
||||
"#ffd166",
|
||||
"#90be6d",
|
||||
"#ff595e",
|
||||
"#6a4c93",
|
||||
"#1982c4",
|
||||
]
|
||||
|
||||
combo_colors: dict[tuple[int, int], str] = {}
|
||||
legend_source_magnitude: dict[tuple[int, int], pg.PlotCurveItem] = {}
|
||||
legend_source_phase: dict[tuple[int, int], pg.PlotCurveItem] = {}
|
||||
active_magnitude_keys: set[tuple[int, int, int, str]] = set()
|
||||
active_phase_keys: set[tuple[int, int, int, str]] = set()
|
||||
has_data = False
|
||||
x_min = np.inf
|
||||
x_max = -np.inf
|
||||
for block in collection.blocks:
|
||||
combo_key = (int(block.combo.input_pos), int(block.combo.output_pos))
|
||||
if combo_key not in combo_colors:
|
||||
combo_colors[combo_key] = palette[len(combo_colors) % len(palette)]
|
||||
color = combo_colors[combo_key]
|
||||
|
||||
for payload_index, payload in enumerate(block.payloads):
|
||||
if payload.kind != 1 or payload.trace.size == 0:
|
||||
continue
|
||||
if payload.frequency_hz.size == 0 or payload.frequency_hz.size != payload.trace.size:
|
||||
continue
|
||||
curve_key = (
|
||||
combo_key[0],
|
||||
combo_key[1],
|
||||
int(payload_index),
|
||||
str(payload.processing_name),
|
||||
)
|
||||
|
||||
local_x_min = float(np.min(payload.frequency_hz))
|
||||
local_x_max = float(np.max(payload.frequency_hz))
|
||||
x_min = min(x_min, local_x_min)
|
||||
x_max = max(x_max, local_x_max)
|
||||
|
||||
if show_magnitude:
|
||||
magnitude_values = 20.0 * np.log10(np.maximum(np.abs(payload.trace), 1e-12))
|
||||
active_magnitude_keys.add(curve_key)
|
||||
magnitude_curve = self._trace_magnitude_curves.get(curve_key)
|
||||
if magnitude_curve is None:
|
||||
magnitude_curve = pg.PlotCurveItem(pen=pg.mkPen(color, width=1.4))
|
||||
self._trace_magnitude_curves[curve_key] = magnitude_curve
|
||||
magnitude_plot.addItem(magnitude_curve)
|
||||
else:
|
||||
magnitude_curve.setPen(pg.mkPen(color, width=1.4))
|
||||
magnitude_curve.setData(payload.frequency_hz, magnitude_values)
|
||||
legend_source_magnitude.setdefault(combo_key, magnitude_curve)
|
||||
has_data = True
|
||||
|
||||
if show_phase:
|
||||
active_phase_keys.add(curve_key)
|
||||
phase_curve = self._trace_phase_curves.get(curve_key)
|
||||
if phase_curve is None:
|
||||
phase_curve = pg.PlotCurveItem(pen=pg.mkPen(color, width=1.2))
|
||||
self._trace_phase_curves[curve_key] = phase_curve
|
||||
phase_plot.addItem(phase_curve)
|
||||
else:
|
||||
phase_curve.setPen(pg.mkPen(color, width=1.2))
|
||||
phase_x, phase_values = self._phase_display_arrays(payload.frequency_hz, payload.trace)
|
||||
phase_curve.setData(phase_x, phase_values)
|
||||
legend_source_phase.setdefault(combo_key, phase_curve)
|
||||
has_data = True
|
||||
|
||||
if show_magnitude:
|
||||
self._remove_inactive_trace_curves(
|
||||
plot=magnitude_plot,
|
||||
cache=self._trace_magnitude_curves,
|
||||
active_keys=active_magnitude_keys,
|
||||
)
|
||||
else:
|
||||
self._remove_all_trace_curves(plot=magnitude_plot, cache=self._trace_magnitude_curves)
|
||||
|
||||
if show_phase:
|
||||
self._remove_inactive_trace_curves(
|
||||
plot=phase_plot,
|
||||
cache=self._trace_phase_curves,
|
||||
active_keys=active_phase_keys,
|
||||
)
|
||||
else:
|
||||
self._remove_all_trace_curves(plot=phase_plot, cache=self._trace_phase_curves)
|
||||
|
||||
self._sync_trace_legends(
|
||||
show_magnitude=show_magnitude,
|
||||
show_phase=show_phase,
|
||||
magnitude_sources=legend_source_magnitude,
|
||||
phase_sources=legend_source_phase,
|
||||
)
|
||||
|
||||
if has_data:
|
||||
if np.isfinite(x_min) and np.isfinite(x_max):
|
||||
if show_magnitude:
|
||||
magnitude_plot.setXRange(x_min, x_max, padding=0.02)
|
||||
if show_phase:
|
||||
phase_plot.setXRange(x_min, x_max, padding=0.02)
|
||||
if show_phase:
|
||||
phase_plot.setYRange(-180.0, 180.0, padding=0.02)
|
||||
if show_magnitude:
|
||||
self._configure_pass_through_magnitude_axis(magnitude_plot)
|
||||
return has_data
|
||||
|
||||
@staticmethod
|
||||
def _remove_inactive_trace_curves(
|
||||
*,
|
||||
plot: pg.PlotWidget,
|
||||
cache: dict[tuple[int, int, int, str], pg.PlotCurveItem],
|
||||
active_keys: set[tuple[int, int, int, str]],
|
||||
) -> None:
|
||||
"""Delete curve items no longer present in latest result collection."""
|
||||
for key in list(cache.keys()):
|
||||
if key in active_keys:
|
||||
continue
|
||||
curve = cache.pop(key)
|
||||
plot.removeItem(curve)
|
||||
|
||||
@staticmethod
|
||||
def _remove_all_trace_curves(
|
||||
*,
|
||||
plot: pg.PlotWidget,
|
||||
cache: dict[tuple[int, int, int, str], pg.PlotCurveItem],
|
||||
) -> None:
|
||||
"""Delete all cached curves from selected plot."""
|
||||
for curve in cache.values():
|
||||
plot.removeItem(curve)
|
||||
cache.clear()
|
||||
|
||||
def _phase_display_arrays(self, frequency_hz: np.ndarray, trace: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
|
||||
"""Return phase display arrays with decimation for faster rendering."""
|
||||
max_points = int(getattr(self, "_trace_phase_render_max_points", 1200))
|
||||
if max_points > 0 and trace.size > max_points:
|
||||
step = max(1, int(np.ceil(trace.size / max_points)))
|
||||
frequency_hz = frequency_hz[::step]
|
||||
trace = trace[::step]
|
||||
phase_values = np.arctan2(trace.imag, trace.real) * (180.0 / np.pi)
|
||||
return frequency_hz, phase_values
|
||||
|
||||
def _sync_trace_legends(
|
||||
self,
|
||||
*,
|
||||
show_magnitude: bool,
|
||||
show_phase: bool,
|
||||
magnitude_sources: dict[tuple[int, int], pg.PlotCurveItem],
|
||||
phase_sources: dict[tuple[int, int], pg.PlotCurveItem],
|
||||
) -> None:
|
||||
"""Rebuild legends only when active combo set changes."""
|
||||
self._sync_single_trace_legend(
|
||||
show=show_magnitude,
|
||||
plot=self._trace_magnitude_plot,
|
||||
legend_attr="_trace_magnitude_legend",
|
||||
legend_keys_attr="_trace_magnitude_legend_combo_keys",
|
||||
sources=magnitude_sources,
|
||||
)
|
||||
self._sync_single_trace_legend(
|
||||
show=show_phase,
|
||||
plot=self._trace_phase_plot,
|
||||
legend_attr="_trace_phase_legend",
|
||||
legend_keys_attr="_trace_phase_legend_combo_keys",
|
||||
sources=phase_sources,
|
||||
)
|
||||
|
||||
def _sync_single_trace_legend(
|
||||
self,
|
||||
*,
|
||||
show: bool,
|
||||
plot: pg.PlotWidget,
|
||||
legend_attr: str,
|
||||
legend_keys_attr: str,
|
||||
sources: dict[tuple[int, int], pg.PlotCurveItem],
|
||||
) -> None:
|
||||
"""Rebuild one legend from provided combo->curve mapping when needed."""
|
||||
legend = getattr(self, legend_attr)
|
||||
existing_keys = getattr(self, legend_keys_attr)
|
||||
active_keys = set(sources.keys())
|
||||
if not show or not active_keys:
|
||||
if legend is not None:
|
||||
try:
|
||||
plot.getPlotItem().removeItem(legend)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
self._log_warning(
|
||||
f"Failed to clear plot legend: {type(exc).__name__}: {exc}",
|
||||
once_key=f"{legend_attr}_clear_failed",
|
||||
)
|
||||
setattr(self, legend_attr, None)
|
||||
existing_keys.clear()
|
||||
return
|
||||
|
||||
if legend is not None and existing_keys == active_keys:
|
||||
return
|
||||
|
||||
if legend is not None:
|
||||
try:
|
||||
plot.getPlotItem().removeItem(legend)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
self._log_warning(
|
||||
f"Failed to replace plot legend: {type(exc).__name__}: {exc}",
|
||||
once_key=f"{legend_attr}_replace_failed",
|
||||
)
|
||||
|
||||
legend = plot.addLegend(offset=(8, 8))
|
||||
for combo_key in sorted(active_keys):
|
||||
curve = sources[combo_key]
|
||||
legend.addItem(curve, f"in{combo_key[0]}/out{combo_key[1]}")
|
||||
setattr(self, legend_attr, legend)
|
||||
existing_keys.clear()
|
||||
existing_keys.update(active_keys)
|
||||
|
||||
def _draw_single_trace(self, trace: TraceData, title: str, *, channel: str = "s21") -> None:
|
||||
"""Draw one trace on stacked magnitude/phase plots."""
|
||||
show_magnitude = self._show_magnitude_curves()
|
||||
show_phase = self._show_phase_curves()
|
||||
magnitude_plot = self._trace_magnitude_plot
|
||||
phase_plot = self._trace_phase_plot
|
||||
samples = trace.s11 if channel == "s11" else trace.s21
|
||||
|
||||
magnitude_plot.setVisible(show_magnitude)
|
||||
phase_plot.setVisible(show_phase)
|
||||
self._clear_trace_plots()
|
||||
if not show_magnitude and not show_phase:
|
||||
return
|
||||
|
||||
if show_magnitude:
|
||||
self._configure_pass_through_magnitude_axis(magnitude_plot)
|
||||
magnitude_plot.getPlotItem().showAxis("bottom", show=not show_phase)
|
||||
magnitude_plot.setLabel("left", "Magnitude", units="dB")
|
||||
magnitude_plot.setTitle(title)
|
||||
if not show_phase:
|
||||
magnitude_plot.setLabel("bottom", "Frequency", units="Hz")
|
||||
if show_phase:
|
||||
phase_plot.getViewBox().invertY(False)
|
||||
phase_plot.getViewBox().enableAutoRange(x=True, y=False)
|
||||
phase_plot.getPlotItem().showAxis("bottom", show=True)
|
||||
phase_plot.setLabel("left", "Phase", units="deg")
|
||||
phase_plot.setLabel("bottom", "Frequency", units="Hz")
|
||||
phase_plot.setTitle(title)
|
||||
|
||||
if show_magnitude:
|
||||
magnitude_db = 20.0 * np.log10(np.maximum(np.abs(samples), 1e-12))
|
||||
magnitude_curve = pg.PlotCurveItem(
|
||||
trace.frequency_hz,
|
||||
magnitude_db,
|
||||
pen=pg.mkPen("#ffd166", width=1.8),
|
||||
)
|
||||
magnitude_plot.addItem(magnitude_curve)
|
||||
self._trace_magnitude_curves[
|
||||
(int(trace.combo.input_pos), int(trace.combo.output_pos), 0, "__single_trace__")
|
||||
] = magnitude_curve
|
||||
|
||||
if show_phase:
|
||||
phase_values = np.degrees(np.angle(samples))
|
||||
phase_curve = pg.PlotCurveItem(
|
||||
trace.frequency_hz,
|
||||
phase_values,
|
||||
pen=pg.mkPen("#80ed99", width=1.4, style=Qt.PenStyle.DashLine),
|
||||
)
|
||||
phase_plot.addItem(phase_curve)
|
||||
self._trace_phase_curves[
|
||||
(int(trace.combo.input_pos), int(trace.combo.output_pos), 0, "__single_trace__")
|
||||
] = phase_curve
|
||||
phase_plot.setYRange(-180.0, 180.0, padding=0.02)
|
||||
|
||||
if np.size(trace.frequency_hz) > 1:
|
||||
x_min = float(np.min(trace.frequency_hz))
|
||||
x_max = float(np.max(trace.frequency_hz))
|
||||
if show_magnitude:
|
||||
magnitude_plot.setXRange(x_min, x_max, padding=0.02)
|
||||
self._configure_pass_through_magnitude_axis(magnitude_plot)
|
||||
if show_phase:
|
||||
phase_plot.setXRange(x_min, x_max, padding=0.02)
|
||||
Reference in New Issue
Block a user