188 lines
7.9 KiB
Python
188 lines
7.9 KiB
Python
"""Select radar views and supply independent renderers with settings and history."""
|
|
|
|
from collections.abc import Callable
|
|
|
|
import pyqtgraph as pg
|
|
from PyQt6.QtCore import QSignalBlocker
|
|
from PyQt6.QtWidgets import QComboBox, QPushButton, QStackedWidget, QWidget
|
|
|
|
from python_app.gui.controllers.renderers.sar3d import Sar3dRenderer
|
|
from python_app.gui.controllers.history_controller import HistoryController
|
|
from python_app.gui.controllers.renderers.bscan import BscanRenderer
|
|
from python_app.gui.controllers.renderers.gpr import GprRenderer
|
|
from python_app.gui.controllers.renderers.trace import TraceRenderer
|
|
from python_app.models.dataset_model import ResultCollection, TraceData
|
|
from python_app.models.gui_profile_schema import GuiProcessingStateModel
|
|
from python_app.models.run_config_schema import GprModel
|
|
|
|
Renderer = BscanRenderer | GprRenderer | Sar3dRenderer | TraceRenderer
|
|
|
|
|
|
class PlotController:
|
|
"""Own the plot views: renderers draw and clear only through it, and it counts image changes."""
|
|
|
|
def __init__(
|
|
self, history: HistoryController, settings: Callable[[], GuiProcessingStateModel],
|
|
geometry: Callable[[], GprModel], running: Callable[[], bool],
|
|
warning: Callable[..., None],
|
|
) -> None:
|
|
self.history = history
|
|
self._settings = settings
|
|
self._geometry = geometry
|
|
self._running = running
|
|
self._warning = warning
|
|
self.required_revision: int | None = None
|
|
self.live_window = 50
|
|
self.bscan = BscanRenderer()
|
|
self.gpr = GprRenderer()
|
|
self.sar3d = Sar3dRenderer()
|
|
self.trace = TraceRenderer(warning)
|
|
self.stack = QStackedWidget()
|
|
for renderer in (self.bscan, self.gpr, self.sar3d, self.trace):
|
|
self.stack.addWidget(renderer.widget)
|
|
self.stack.setCurrentWidget(self.trace.widget)
|
|
# Counts changes of the shown image (drawing, page, size, view range, a legend click
|
|
# hiding a curve) for observers such as the web stream. QGraphicsScene.changed must stay
|
|
# unconnected: it switches Qt to a legacy update path that leaves stale pixels behind.
|
|
self.revision = 0
|
|
self.stack.currentChanged.connect(self._changed)
|
|
for plot in self.stack.findChildren(pg.PlotWidget):
|
|
item = plot.getPlotItem()
|
|
assert item is not None
|
|
view = item.getViewBox()
|
|
view.sigResized.connect(self._changed)
|
|
view.sigRangeChanged.connect(self._changed)
|
|
for panel in (self.trace.magnitude, self.trace.phase):
|
|
panel.legend.sigSampleClicked.connect(self._changed)
|
|
self._combo: QComboBox | None = None
|
|
self._next: QPushButton | None = None
|
|
|
|
def _changed(self, *_args: object) -> None:
|
|
self.revision += 1
|
|
|
|
def _renderer(self, mode: str) -> Renderer:
|
|
return {"bscan": self.bscan, "gpr": self.gpr, "sar3d": self.sar3d}.get(mode, self.trace)
|
|
|
|
def bind_display_controls(self, combo: QComboBox, next_button: QPushButton) -> None:
|
|
self._combo, self._next = combo, next_button
|
|
combo.currentIndexChanged.connect(self.on_display_combo_changed)
|
|
next_button.clicked.connect(self.select_next_combo)
|
|
self.sync_display_combos()
|
|
|
|
@property
|
|
def display_key(self) -> tuple[int, int] | None:
|
|
return self._combo.currentData() if self._combo is not None else None
|
|
|
|
@property
|
|
def capture_widget(self) -> QWidget | None:
|
|
current = self.stack.currentWidget()
|
|
return self.sar3d.grid if current is self.sar3d.widget else current
|
|
|
|
def set_mode(self, mode: str) -> None:
|
|
self.stack.setCurrentWidget(self._renderer(mode).widget)
|
|
self.sync_display_combos()
|
|
|
|
def select_combo(self, input_pos: int, output_pos: int) -> None:
|
|
key = input_pos, output_pos
|
|
keys = sorted(self.bscan.history_by_combo)
|
|
if key not in keys or self._combo is None:
|
|
raise ValueError(f"Unknown channel combination: {input_pos}:{output_pos}")
|
|
self._combo.setCurrentIndex(keys.index(key))
|
|
|
|
def select_next_combo(self) -> None:
|
|
combo = self._combo
|
|
if combo is not None and combo.count():
|
|
combo.setCurrentIndex((combo.currentIndex() + 1) % combo.count())
|
|
|
|
def sync_display_combos(self) -> None:
|
|
combo, next_button = self._combo, self._next
|
|
if combo is None or next_button is None:
|
|
return
|
|
keys = sorted(self.bscan.history_by_combo)
|
|
current = combo.currentData()
|
|
if keys != [combo.itemData(i) for i in range(combo.count())]:
|
|
with QSignalBlocker(combo):
|
|
combo.clear()
|
|
for key in keys:
|
|
combo.addItem(f"{key[0]} : {key[1]}", key)
|
|
combo.setCurrentIndex(keys.index(current) if current in keys else 0)
|
|
enabled = bool(keys) and self._settings().selected_mode == "bscan"
|
|
combo.setEnabled(enabled)
|
|
next_button.setEnabled(enabled and len(keys) > 1)
|
|
|
|
def window_scans(self) -> int:
|
|
return self.live_window if self._running() else max(1, self._settings().bscan.history_window_scans)
|
|
|
|
def sync_bscan(self) -> None:
|
|
results = self.history.bscan_source()
|
|
self.bscan.sync(results, self.window_scans())
|
|
self.sync_display_combos()
|
|
|
|
def draw_bscan(self) -> bool:
|
|
self._changed()
|
|
self.sync_bscan()
|
|
window, retained = self.window_scans(), len(self.history.preprocessed)
|
|
if retained < window and self.bscan.history_by_combo:
|
|
self._warning(
|
|
f"B-scan is set to show {window} sweeps but only {retained} are retained; "
|
|
"the image shows what history there is.",
|
|
details="Capture more sweeps, or reduce 'Scans to show (stopped)'.",
|
|
once_key=f"bscan_window_underfilled_{window}",
|
|
)
|
|
return self.bscan.draw(self.display_key, self._settings().bscan)
|
|
|
|
def on_display_combo_changed(self, *_args: object) -> None:
|
|
if self._settings().selected_mode == "bscan":
|
|
self.draw_bscan()
|
|
|
|
def draw(self, collection: ResultCollection) -> bool:
|
|
settings = self._settings()
|
|
mode = settings.selected_mode
|
|
if mode in {"gpr", "sar3d"} and self.required_revision is not None:
|
|
if collection.settings_revision != self.required_revision:
|
|
return False
|
|
if mode == "bscan":
|
|
return self.draw_bscan()
|
|
self._changed()
|
|
if mode == "gpr":
|
|
return self.gpr.draw(collection, settings.gpr, self._geometry())
|
|
if mode == "sar3d":
|
|
return self.sar3d.draw(collection, settings.sar3d, self._geometry())
|
|
return self.trace.draw(collection, settings.pass_through)
|
|
|
|
def redraw(self, *_args: object) -> None:
|
|
"""Show the newest result with the current settings."""
|
|
mode = self._settings().selected_mode
|
|
if mode == "bscan":
|
|
self.draw_bscan()
|
|
return
|
|
# Cleared first: a result of another settings revision is not drawn and must not
|
|
# leave the previous image, while a renderer's own failure message is kept.
|
|
self._changed()
|
|
self._renderer(mode).clear()
|
|
if self.history.results:
|
|
self.draw(self.history.results[-1])
|
|
|
|
def draw_trace(self, trace: TraceData, title: str) -> None:
|
|
self._changed()
|
|
self.trace.draw_trace(trace, title, self._settings().pass_through)
|
|
|
|
def clear_trace(self) -> None:
|
|
self._changed()
|
|
self.trace.clear()
|
|
|
|
def clear_bscan(self) -> None:
|
|
self._changed()
|
|
self.bscan.clear()
|
|
self.sync_display_combos()
|
|
|
|
def clear(self) -> None:
|
|
self._changed()
|
|
for renderer in (self.bscan, self.gpr, self.sar3d, self.trace):
|
|
renderer.clear()
|
|
self.sync_display_combos()
|
|
|
|
@staticmethod
|
|
def has_output(collection: ResultCollection) -> bool:
|
|
return bool(collection.collection_payloads) or any(p.kind == 1 for b in collection.blocks for p in b.payloads)
|