GUI: updates to help messages, predictions stream, added polygons to predictions, added beamline recovery panel, controls help dialog, smargon trace panel
This commit is contained in:
+9
-3
@@ -37,8 +37,8 @@ if __name__ == "__main__":
|
||||
default_gonio_camera_id = 3
|
||||
case MXBeamline.X10SA:
|
||||
default_url = "http://127.0.0.1:5210"
|
||||
default_zmq_addr = "tcp://x10sa-spark-01:9091" #"tcp://x10sa-pserv-01:9089"
|
||||
default_pred_zmq_addr = "tcp://x10sa-spark-01:9091"
|
||||
default_zmq_addr = "tcp://x10sa-spark-01:9091" #"tcp://x10sa-pserv-01:9089" #
|
||||
default_pred_zmq_addr = "tcp://x10sa-spark-01:9091" #"tcp://sls-gpu-003:9089"#""
|
||||
default_beamline_cam_addr = "axis-accc8eb02488.psi.ch"
|
||||
default_gonio_cam_addr = "axis-accc8ea5e463.psi.ch"
|
||||
default_gonio_camera_id = 1
|
||||
@@ -103,6 +103,11 @@ if __name__ == "__main__":
|
||||
|
||||
try:
|
||||
token = auth(base_url)
|
||||
if not token or token.count(".") != 2:
|
||||
raise RuntimeError(
|
||||
"Authentication did not return a valid token. "
|
||||
"Please check the server is running (it may still be initialising)."
|
||||
)
|
||||
logger.info("Authentication successful")
|
||||
except Exception as e:
|
||||
logger.error(f"Cannot connect to AareDAQ server. Exiting. {e}")
|
||||
@@ -133,8 +138,9 @@ if __name__ == "__main__":
|
||||
QMessageBox.critical(
|
||||
None,
|
||||
"Fatal Error",
|
||||
f"An error occurred during startup:\n\n{str(e)}\n\nSee console for details."
|
||||
f"An error occurred during startup. See console for details."
|
||||
f"\nPlease check the server is running and your network connection."
|
||||
f"\n\n{str(e)}\n\n"
|
||||
)
|
||||
except:
|
||||
pass
|
||||
|
||||
+185
-18
@@ -1,6 +1,8 @@
|
||||
import time
|
||||
|
||||
import jwt
|
||||
from PySide6.QtCore import Qt, Slot, Signal
|
||||
from PySide6.QtGui import QAction
|
||||
from PySide6.QtCore import Qt, Slot, Signal, QTimer, QSettings
|
||||
from PySide6.QtGui import QAction, QPixmap
|
||||
from PySide6.QtWidgets import (
|
||||
QMainWindow,
|
||||
QWidget,
|
||||
@@ -20,16 +22,19 @@ from aare.gui.panels.LogPanel import LogDock
|
||||
from aare.gui.panels.beamline_controls import BeamlineControls
|
||||
from aare.gui.panels.data_collection_settings import DataCollectionSettings
|
||||
from aare.gui.panels.developer_help_dialog import DeveloperHelpDialog
|
||||
from aare.gui.panels.beamline_recovery_panel import BeamlineRecoveryDialog
|
||||
from aare.gui.panels.manual_sample_panel import ManualSamplePanel
|
||||
from aare.gui.panels.reference_tools_panel import ReferenceToolsPanel
|
||||
from aare.gui.panels.sample_queue_panel import SampleQueuePanel
|
||||
from aare.gui.panels.tell_sample_panel import TellSamplePanel
|
||||
from aare.gui.panels.face_detection_panel import FaceDetectionPanel
|
||||
from aare.gui.panels.smargon_trace_panel import SmargonTracePanel
|
||||
from aare.gui.scan_logic.raster_grid_manager import RasterGridManager
|
||||
from aare.gui.scan_logic.rotation_scan_manager import RotationScanManager
|
||||
from aare.gui.scan_logic.sample_mount_logic import SampleMountLogic
|
||||
from aare.gui.threads.axis_video_thread import VideoThread
|
||||
from aare.gui.tutorials.tutorial_manager import TutorialManager, TutorialStep
|
||||
from aare.gui.tutorials.controls_help_dialog import ControlsHelpDialog
|
||||
|
||||
from aare.gui.threads.camera_thread import SampleCameraThread
|
||||
from aare.gui.threads.prediction_subscriber import PredictionSubscriber
|
||||
@@ -61,15 +66,35 @@ class MainWindow(QMainWindow):
|
||||
self.__token = token
|
||||
self.__mounting = False
|
||||
self._dev_help_dialog = None
|
||||
self._beamline_recovery_dialog = None
|
||||
self._controls_help_dialog = None
|
||||
self._cleanup_done = False
|
||||
|
||||
# Tutorial manager (define tutorials after widgets exist)
|
||||
self.tutorial_manager = TutorialManager(self)
|
||||
|
||||
self.viewer = JFJochDBusClient()
|
||||
|
||||
# Decode the JWT without signature verification
|
||||
self.__decoded_token = TokenData(**jwt.decode(token, options={"verify_signature": False}))
|
||||
logger.debug(self.__decoded_token)
|
||||
try:
|
||||
token_str = (token or "").strip()
|
||||
if token_str.count(".") != 2:
|
||||
raise ValueError(
|
||||
"Invalid authentication token received (not a JWT). "
|
||||
"This usually happens when the server is not running or still starting."
|
||||
)
|
||||
|
||||
payload = jwt.decode(token_str, options={"verify_signature": False})
|
||||
self.__decoded_token = TokenData(**payload)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to decode authentication token: {e}", exc_info=True)
|
||||
QMessageBox.critical(
|
||||
None,
|
||||
"Authentication Error",
|
||||
"Could not start the GUI because authentication data was invalid.\n\n"
|
||||
"Most commonly the server is not running yet (or is still initialising).\n"
|
||||
"Please start/restart the server and try again."
|
||||
)
|
||||
raise
|
||||
|
||||
self.setStyleSheet("background-color: rgb(216, 228, 253);")
|
||||
|
||||
@@ -176,6 +201,7 @@ class MainWindow(QMainWindow):
|
||||
self.addDockWidget(Qt.DockWidgetArea.BottomDockWidgetArea, self.tell_samples_dock)
|
||||
|
||||
self.ref_tools_dock = QDockWidget("Reference Tools", self)
|
||||
self.ref_tools_dock.setObjectName("ref_tools_dock")
|
||||
self.ref_tools_dock.setWidget(self.ref_tools_panel)
|
||||
self.ref_tools_dock.setAllowedAreas(Qt.DockWidgetArea.BottomDockWidgetArea)
|
||||
self.addDockWidget(Qt.DockWidgetArea.BottomDockWidgetArea, self.ref_tools_dock)
|
||||
@@ -184,6 +210,7 @@ class MainWindow(QMainWindow):
|
||||
self.sample_logic = SampleMountLogic()
|
||||
|
||||
self.job_list_dock = QDockWidget("Automation list", self)
|
||||
self.job_list_dock.setObjectName("job_list_dock")
|
||||
self.job_list_dock.setWidget(self.job_list_panel)
|
||||
self.job_list_dock.setAllowedAreas(Qt.DockWidgetArea.BottomDockWidgetArea)
|
||||
self.addDockWidget(Qt.DockWidgetArea.BottomDockWidgetArea, self.job_list_dock)
|
||||
@@ -191,12 +218,14 @@ class MainWindow(QMainWindow):
|
||||
self.manual_sample_panel = ManualSamplePanel()
|
||||
|
||||
self.manual_sample_dock = QDockWidget("Manual sample", self)
|
||||
self.manual_sample_dock.setObjectName("manual_sample_dock")
|
||||
self.manual_sample_dock.setWidget(self.manual_sample_panel)
|
||||
self.manual_sample_dock.setAllowedAreas(Qt.DockWidgetArea.BottomDockWidgetArea)
|
||||
self.addDockWidget(Qt.DockWidgetArea.BottomDockWidgetArea, self.manual_sample_dock)
|
||||
|
||||
self.face_panel = FaceDetectionPanel()
|
||||
self.face_panel_dock = QDockWidget("Face detection", self)
|
||||
self.face_panel_dock.setObjectName("face_panel_dock")
|
||||
self.face_panel_dock.setWidget(self.face_panel)
|
||||
self.face_panel_dock.setAllowedAreas(Qt.DockWidgetArea.RightDockWidgetArea | Qt.DockWidgetArea.LeftDockWidgetArea)
|
||||
self.addDockWidget(Qt.DockWidgetArea.RightDockWidgetArea, self.face_panel_dock)
|
||||
@@ -204,6 +233,7 @@ class MainWindow(QMainWindow):
|
||||
|
||||
self.fluor_panel = FluorescencePanel()
|
||||
self.fluor_panel_dock = QDockWidget("Fluorescence", self)
|
||||
self.fluor_panel_dock.setObjectName("fluor_panel_dock")
|
||||
self.fluor_panel_dock.setWidget(self.fluor_panel)
|
||||
self.fluor_panel_dock.setAllowedAreas(Qt.DockWidgetArea.TopDockWidgetArea | Qt.DockWidgetArea.BottomDockWidgetArea | Qt.DockWidgetArea.RightDockWidgetArea)
|
||||
self.addDockWidget(Qt.DockWidgetArea.BottomDockWidgetArea, self.fluor_panel_dock)
|
||||
@@ -211,16 +241,30 @@ class MainWindow(QMainWindow):
|
||||
|
||||
# Create and add the dock to your main window
|
||||
self.log_dock = LogDock("Console Log", self)
|
||||
self.log_dock.setObjectName("log_dock")
|
||||
self.addDockWidget(Qt.BottomDockWidgetArea, self.log_dock)
|
||||
self.log_dock.attach_logger("")
|
||||
self.log_dock.attach_logger("aareDAQ")
|
||||
self.log_dock.attach_logger("aareGUI")
|
||||
self.log_dock.hide()
|
||||
|
||||
self.smargon_trace_panel = SmargonTracePanel()
|
||||
self.smargon_trace_dock = QDockWidget("Smargon trace", self)
|
||||
self.smargon_trace_dock.setObjectName("smargon_trace_dock")
|
||||
self.smargon_trace_dock.setWidget(self.smargon_trace_panel)
|
||||
self.smargon_trace_dock.setAllowedAreas(
|
||||
Qt.DockWidgetArea.RightDockWidgetArea
|
||||
| Qt.DockWidgetArea.LeftDockWidgetArea
|
||||
| Qt.DockWidgetArea.BottomDockWidgetArea
|
||||
)
|
||||
self.addDockWidget(Qt.DockWidgetArea.RightDockWidgetArea, self.smargon_trace_dock)
|
||||
self.smargon_trace_dock.hide()
|
||||
|
||||
self.setCentralWidget(top_widget)
|
||||
|
||||
self.setWindowTitle("AareGUI")
|
||||
self.create_menu_bar()
|
||||
self._restore_window_state()
|
||||
|
||||
# Define tutorials now that the UI exists
|
||||
|
||||
@@ -233,6 +277,7 @@ class MainWindow(QMainWindow):
|
||||
self.daq.reference_tools.connect(self.ref_tools_panel.new_list)
|
||||
|
||||
self.beamline.samcam.changed.connect(self.daq.samcam_settings)
|
||||
self.beamline.samcam.screenshot_requested.connect(self.daq.send_screenshot_db)
|
||||
self.beamline.loopctr.background.clicked.connect(self.daq.alc_background)
|
||||
self.beamline.loopctr.find_tip.clicked.connect(self.daq.center_loop)
|
||||
self.beamline.loopctr.bounding_box.clicked.connect(self.daq.ml_bounding_box)
|
||||
@@ -267,7 +312,6 @@ class MainWindow(QMainWindow):
|
||||
|
||||
if zmq_addr is not None:
|
||||
self.camera_thread = SampleCameraThread(zmq_url=zmq_addr)
|
||||
self.camera_thread.camera_image.connect(self.sample_camera.update_pixmap)
|
||||
self.camera_thread.start()
|
||||
self.camera_thread.focus_measure.connect(self.status_bar.update_sharpness)
|
||||
self.camera_thread.fps_measure.connect(self.status_bar.update_samcam_fps)
|
||||
@@ -278,14 +322,25 @@ class MainWindow(QMainWindow):
|
||||
if pred_zmq_addr is not None:
|
||||
logger.debug(f"Starting prediction subscriber thread {pred_zmq_addr}")
|
||||
self.prediction_thread = PredictionSubscriber(pred_zmq_url=pred_zmq_addr, topic="detections")
|
||||
self.prediction_thread.image.connect(self.sample_camera.update_pixmap)
|
||||
self.prediction_thread.prediction.connect(self.sample_camera.update_detections)
|
||||
self.prediction_thread.start()
|
||||
else:
|
||||
self.prediction_thread = None
|
||||
|
||||
self._last_pred_image_ts: float | None = None
|
||||
self._pred_preferred_timeout_s: float = 0.7 # tune: how long we "trust" prediction images
|
||||
self._pred_is_preferred: bool = False
|
||||
|
||||
QApplication.instance().aboutToQuit.connect(self.cleanup)
|
||||
if self.camera_thread is not None:
|
||||
self.camera_thread.camera_image.connect(self._on_samcam_camera_pixmap)
|
||||
|
||||
if self.prediction_thread is not None:
|
||||
self.prediction_thread.image.connect(self._on_samcam_prediction_pixmap)
|
||||
|
||||
self._samcam_source_timer = QTimer(self)
|
||||
self._samcam_source_timer.setInterval(200) # ms
|
||||
self._samcam_source_timer.timeout.connect(self._update_samcam_source_preference)
|
||||
self._samcam_source_timer.start()
|
||||
|
||||
#
|
||||
# self.data_collection.helical.helical_scan.connect(self.worker.helical_scan)
|
||||
@@ -395,6 +450,31 @@ class MainWindow(QMainWindow):
|
||||
|
||||
register_tutorials(self, self.tutorial_manager)
|
||||
|
||||
@Slot(QPixmap)
|
||||
def _on_samcam_prediction_pixmap(self, pix: QPixmap) -> None:
|
||||
self._last_pred_image_ts = time.monotonic()
|
||||
self._pred_is_preferred = True
|
||||
self.sample_camera.update_pixmap(pix)
|
||||
|
||||
@Slot(QPixmap)
|
||||
def _on_samcam_camera_pixmap(self, pix: QPixmap) -> None:
|
||||
# Only show camera frames when prediction is not currently "healthy"
|
||||
if not self._pred_is_preferred:
|
||||
self.sample_camera.update_pixmap(pix)
|
||||
|
||||
@Slot()
|
||||
def _update_samcam_source_preference(self) -> None:
|
||||
if self.prediction_thread is None:
|
||||
self._pred_is_preferred = False
|
||||
return
|
||||
|
||||
if self._last_pred_image_ts is None:
|
||||
self._pred_is_preferred = False
|
||||
return
|
||||
|
||||
age_s = time.monotonic() - self._last_pred_image_ts
|
||||
self._pred_is_preferred = age_s <= self._pred_preferred_timeout_s
|
||||
|
||||
def start_text_tutorial(self) -> None:
|
||||
self.tutorial_manager.start("intro_text")
|
||||
|
||||
@@ -453,6 +533,15 @@ class MainWindow(QMainWindow):
|
||||
self.fluor_panel_dock.visibilityChanged.connect(show_fluor_panel_action.setChecked)
|
||||
view_menu.addAction(show_fluor_panel_action)
|
||||
|
||||
show_smargon_trace_action = QAction("Show Smargon trace", self)
|
||||
show_smargon_trace_action.setCheckable(True)
|
||||
show_smargon_trace_action.setChecked(False)
|
||||
show_smargon_trace_action.triggered.connect(lambda checked: self.smargon_trace_dock.setVisible(checked))
|
||||
self.smargon_trace_dock.visibilityChanged.connect(
|
||||
lambda visible: self.smargon_trace_panel.refresh_plot(force=True) if visible else None
|
||||
)
|
||||
view_menu.addAction(show_smargon_trace_action)
|
||||
|
||||
show_log_action = QAction("Show Log", self)
|
||||
show_log_action.setCheckable(True)
|
||||
show_log_action.setChecked(False)
|
||||
@@ -465,10 +554,19 @@ class MainWindow(QMainWindow):
|
||||
about_action.triggered.connect(self.show_about_dialog)
|
||||
help_menu.addAction(about_action)
|
||||
|
||||
controls_help_action = QAction("Mouse / Keyboard Controls", self)
|
||||
controls_help_action.triggered.connect(self.show_controls_help)
|
||||
help_menu.addAction(controls_help_action)
|
||||
|
||||
dev_help_action = QAction("Developer / Help", self)
|
||||
dev_help_action.triggered.connect(self.show_developer_help)
|
||||
help_menu.addAction(dev_help_action)
|
||||
|
||||
if self.__decoded_token.staff:
|
||||
beamline_recovery_action = QAction("Beamline Recovery", self)
|
||||
beamline_recovery_action.triggered.connect(self.show_beamline_recovery)
|
||||
help_menu.addAction(beamline_recovery_action)
|
||||
|
||||
help_menu.addSeparator()
|
||||
|
||||
start_text_tutorial_action = QAction("Start Tutorial (Text)", self)
|
||||
@@ -485,6 +583,14 @@ class MainWindow(QMainWindow):
|
||||
"About",
|
||||
"Aare Macromolecular Crystallography GUI\nVersion: 1.0\nCopyright: Paul Scherrer Institute 2024-2025",
|
||||
)
|
||||
|
||||
def show_controls_help(self) -> None:
|
||||
if self._controls_help_dialog is None:
|
||||
self._controls_help_dialog = ControlsHelpDialog(parent=self)
|
||||
self._controls_help_dialog.show()
|
||||
self._controls_help_dialog.raise_()
|
||||
self._controls_help_dialog.activateWindow()
|
||||
|
||||
def show_developer_help(self) -> None:
|
||||
if self._dev_help_dialog is None:
|
||||
self._dev_help_dialog = DeveloperHelpDialog(
|
||||
@@ -497,6 +603,18 @@ class MainWindow(QMainWindow):
|
||||
self._dev_help_dialog.raise_()
|
||||
self._dev_help_dialog.activateWindow()
|
||||
|
||||
def show_beamline_recovery(self) -> None:
|
||||
if not bool(getattr(self.__decoded_token, "staff", False)):
|
||||
return
|
||||
if self._beamline_recovery_dialog is None:
|
||||
self._beamline_recovery_dialog = BeamlineRecoveryDialog(
|
||||
daq=self.daq,
|
||||
parent=self,
|
||||
)
|
||||
self._beamline_recovery_dialog.show()
|
||||
self._beamline_recovery_dialog.raise_()
|
||||
self._beamline_recovery_dialog.activateWindow()
|
||||
|
||||
@Slot(str)
|
||||
def show_sample_missing_dialog(self, msg: str):
|
||||
if self.job_list_panel.is_running():
|
||||
@@ -514,8 +632,12 @@ class MainWindow(QMainWindow):
|
||||
|
||||
@Slot(DAQStatusModel)
|
||||
def update_daq_status(self, s: DAQStatusModel):
|
||||
self.beamline_camera_thread.set_busy(s.busy)
|
||||
self.gonio_camera_thread.set_busy(s.busy)
|
||||
if hasattr(self, "beamline_camera_thread") and self.beamline_camera_thread is not None:
|
||||
self.beamline_camera_thread.set_busy(s.busy)
|
||||
|
||||
if hasattr(self, "gonio_camera_thread") and self.gonio_camera_thread is not None:
|
||||
self.gonio_camera_thread.set_busy(s.busy)
|
||||
|
||||
if not self.__mounting and s.state == BeamlineStateEnum.RobotSampleExchange:
|
||||
self.__mounting = True
|
||||
self.video_tab.setCurrentIndex(3)
|
||||
@@ -532,12 +654,57 @@ class MainWindow(QMainWindow):
|
||||
self.status_bar.setStyleSheet("color: green;")
|
||||
self.status_bar.showMessage(f'<span style="color: green; "> {msg} </span>', 10000)
|
||||
|
||||
def _restore_window_state(self) -> None:
|
||||
settings = QSettings()
|
||||
geometry = settings.value("main_window/geometry")
|
||||
state = settings.value("main_window/state")
|
||||
|
||||
if geometry is not None:
|
||||
self.restoreGeometry(geometry)
|
||||
if state is not None:
|
||||
self.restoreState(state)
|
||||
|
||||
def closeEvent(self, event) -> None:
|
||||
try:
|
||||
settings = QSettings()
|
||||
settings.setValue("main_window/geometry", self.saveGeometry())
|
||||
settings.setValue("main_window/state", self.saveState())
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to save main window state: {e}")
|
||||
|
||||
try:
|
||||
self.cleanup()
|
||||
except Exception as e:
|
||||
logger.warning(f"Cleanup during closeEvent failed: {e}")
|
||||
|
||||
super().closeEvent(event)
|
||||
|
||||
def cleanup(self):
|
||||
if self.camera_thread is not None:
|
||||
self.camera_thread.stop()
|
||||
if self.prediction_thread is not None:
|
||||
self.prediction_thread.stop()
|
||||
if self.beamline_camera_thread is not None:
|
||||
self.beamline_camera_thread.stop()
|
||||
if self.gonio_camera_thread is not None:
|
||||
self.gonio_camera_thread.stop()
|
||||
if getattr(self, "_cleanup_done", False):
|
||||
return
|
||||
self._cleanup_done = True
|
||||
|
||||
try:
|
||||
if hasattr(self, "_samcam_source_timer") and self._samcam_source_timer is not None:
|
||||
self._samcam_source_timer.stop()
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to stop _samcam_source_timer: {e}")
|
||||
|
||||
for attr_name in (
|
||||
"camera_thread",
|
||||
"prediction_thread",
|
||||
"beamline_camera_thread",
|
||||
"gonio_camera_thread",
|
||||
):
|
||||
thread = getattr(self, attr_name, None)
|
||||
if thread is None:
|
||||
continue
|
||||
|
||||
logger.debug(f"Stopping {attr_name}")
|
||||
|
||||
try:
|
||||
thread.stop()
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to stop {attr_name}: {e}")
|
||||
|
||||
setattr(self, attr_name, None)
|
||||
@@ -0,0 +1,282 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from PySide6.QtCore import Slot
|
||||
from PySide6.QtWidgets import (
|
||||
QDialog,
|
||||
QVBoxLayout,
|
||||
QLabel,
|
||||
QPushButton,
|
||||
QWidget,
|
||||
QDialogButtonBox,
|
||||
QInputDialog,
|
||||
QLineEdit,
|
||||
QMessageBox,
|
||||
)
|
||||
|
||||
from aare.common.models import DAQStatusModel
|
||||
from aare.gui.threads.daq_worker import DAQWorker
|
||||
|
||||
|
||||
class RecoveryPanel(QWidget):
|
||||
def __init__(self, *, daq: DAQWorker, parent=None):
|
||||
super().__init__(parent)
|
||||
self._daq = daq
|
||||
self._last_status: DAQStatusModel | None = None
|
||||
|
||||
layout = QVBoxLayout(self)
|
||||
layout.setSpacing(10)
|
||||
|
||||
self._warning_primary = QLabel(
|
||||
"⚠ Recovery actions are staff-only and intentionally dangerous.",
|
||||
self,
|
||||
)
|
||||
self._warning_primary.setWordWrap(True)
|
||||
self._warning_primary.setStyleSheet(
|
||||
"QLabel {"
|
||||
" background: #fff3cd;"
|
||||
" color: #7a4b00;"
|
||||
" border: 1px solid #f0c36d;"
|
||||
" border-radius: 6px;"
|
||||
" padding: 8px;"
|
||||
" font-weight: 600;"
|
||||
"}"
|
||||
)
|
||||
layout.addWidget(self._warning_primary)
|
||||
|
||||
self._warning_secondary = QLabel(
|
||||
"Only use these commands when beamline is stuck and certain beamline is unrecoverable through normal operation.",
|
||||
self,
|
||||
)
|
||||
self._warning_secondary.setWordWrap(True)
|
||||
self._warning_secondary.setStyleSheet(
|
||||
"QLabel {"
|
||||
" background: #fdeaea;"
|
||||
" color: #8b1e1e;"
|
||||
" border: 1px solid #e6a8a8;"
|
||||
" border-radius: 6px;"
|
||||
" padding: 8px;"
|
||||
" font-weight: 600;"
|
||||
"}"
|
||||
)
|
||||
layout.addWidget(self._warning_secondary)
|
||||
|
||||
self._status = QLabel("Current status: waiting for DAQ status update…", self)
|
||||
self._status.setWordWrap(True)
|
||||
self._status.setStyleSheet(
|
||||
"QLabel {"
|
||||
" background: #fafafa;"
|
||||
" border: 1px solid #d0d0d0;"
|
||||
" border-radius: 6px;"
|
||||
" padding: 8px;"
|
||||
"}"
|
||||
)
|
||||
layout.addWidget(self._status)
|
||||
|
||||
self._take_over_btn = QPushButton("Take over beamline", self)
|
||||
self._take_over_btn.setStyleSheet(
|
||||
"QPushButton {"
|
||||
" background: #fff3cd;"
|
||||
" border: 1px solid #f0c36d;"
|
||||
" border-radius: 6px;"
|
||||
" padding: 8px;"
|
||||
" font-weight: 600;"
|
||||
"}"
|
||||
)
|
||||
self._take_over_btn.clicked.connect(self._take_over_beamline)
|
||||
layout.addWidget(self._take_over_btn)
|
||||
|
||||
self._free_beamline_btn = QPushButton("Free beamline", self)
|
||||
self._free_beamline_btn.setStyleSheet(
|
||||
"QPushButton {"
|
||||
" background: #fff3cd;"
|
||||
" border: 1px solid #f0c36d;"
|
||||
" border-radius: 6px;"
|
||||
" padding: 8px;"
|
||||
" font-weight: 600;"
|
||||
"}"
|
||||
)
|
||||
self._free_beamline_btn.clicked.connect(self._free_beamline)
|
||||
layout.addWidget(self._free_beamline_btn)
|
||||
|
||||
self._recover_beamline_btn = QPushButton("Recover beamline", self)
|
||||
self._recover_beamline_btn.setStyleSheet(
|
||||
"QPushButton {"
|
||||
" background: #fdeaea;"
|
||||
" color: #8b1e1e;"
|
||||
" border: 1px solid #e6a8a8;"
|
||||
" border-radius: 6px;"
|
||||
" padding: 8px;"
|
||||
" font-weight: 700;"
|
||||
"}"
|
||||
)
|
||||
self._recover_beamline_btn.clicked.connect(self._recover_beamline)
|
||||
layout.addWidget(self._recover_beamline_btn)
|
||||
|
||||
self._recovery_unmount_btn = QPushButton("Unmount sample (recovery)", self)
|
||||
self._recovery_unmount_btn.setStyleSheet(
|
||||
"QPushButton {"
|
||||
" background: #fdeaea;"
|
||||
" color: #8b1e1e;"
|
||||
" border: 1px solid #e6a8a8;"
|
||||
" border-radius: 6px;"
|
||||
" padding: 8px;"
|
||||
" font-weight: 700;"
|
||||
"}"
|
||||
)
|
||||
self._recovery_unmount_btn.clicked.connect(self._recovery_unmount_sample)
|
||||
layout.addWidget(self._recovery_unmount_btn)
|
||||
|
||||
layout.addStretch(1)
|
||||
|
||||
self._daq.update.connect(self._set_daq_status)
|
||||
self._refresh_buttons()
|
||||
|
||||
def _prompt_recovery_code(self, action_name: str) -> str | None:
|
||||
code, ok = QInputDialog.getText(
|
||||
self,
|
||||
action_name,
|
||||
"Enter recovery confirmation code:",
|
||||
QLineEdit.EchoMode.Password,
|
||||
)
|
||||
if not ok:
|
||||
return None
|
||||
code = code.strip()
|
||||
return code or None
|
||||
|
||||
def _sample_appears_mounted(self) -> bool:
|
||||
try:
|
||||
return self._last_status is not None and self._last_status.sample is not None
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
def _beamline_appears_busy(self) -> bool:
|
||||
try:
|
||||
return self._last_status is not None and bool(self._last_status.busy)
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
def _status_text(self) -> str:
|
||||
if self._last_status is None:
|
||||
return "Current status: waiting for DAQ status update…"
|
||||
|
||||
state_name = getattr(self._last_status.state, "name", str(self._last_status.state))
|
||||
busy = bool(getattr(self._last_status, "busy", False))
|
||||
sample_mounted = self._sample_appears_mounted()
|
||||
tell_connected = bool(getattr(self._last_status, "tell_connected", False))
|
||||
|
||||
return (
|
||||
f"State: {state_name}\n"
|
||||
f"Busy: {busy}\n"
|
||||
f"Sample mounted: {sample_mounted}\n"
|
||||
f"TELL connected: {tell_connected}"
|
||||
)
|
||||
|
||||
def _refresh_buttons(self) -> None:
|
||||
sample_mounted = self._sample_appears_mounted()
|
||||
beamline_busy = self._beamline_appears_busy()
|
||||
|
||||
self._recovery_unmount_btn.setEnabled(sample_mounted)
|
||||
self._recovery_unmount_btn.setToolTip(
|
||||
"" if sample_mounted else "Disabled because no mounted sample is visible in current status."
|
||||
)
|
||||
|
||||
self._free_beamline_btn.setEnabled(beamline_busy)
|
||||
self._free_beamline_btn.setToolTip(
|
||||
"" if beamline_busy else "Disabled because beamline does not currently appear busy."
|
||||
)
|
||||
|
||||
def _confirm(self, title: str, msg: str) -> bool:
|
||||
reply = QMessageBox.warning(
|
||||
self,
|
||||
title,
|
||||
msg,
|
||||
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
|
||||
QMessageBox.StandardButton.No,
|
||||
)
|
||||
return reply == QMessageBox.StandardButton.Yes
|
||||
|
||||
@Slot(DAQStatusModel)
|
||||
def _set_daq_status(self, s: DAQStatusModel) -> None:
|
||||
self._last_status = s
|
||||
self._status.setText(self._status_text())
|
||||
self._refresh_buttons()
|
||||
|
||||
@Slot()
|
||||
def _take_over_beamline(self) -> None:
|
||||
if not self._confirm(
|
||||
"Take over beamline",
|
||||
"This will forcefully grab the active beamline session.\n\nDo you want to continue?",
|
||||
):
|
||||
return
|
||||
code = self._prompt_recovery_code("Take over beamline")
|
||||
if not code:
|
||||
return
|
||||
self._daq.take_over_beamline(code)
|
||||
|
||||
@Slot()
|
||||
def _free_beamline(self) -> None:
|
||||
if not self._confirm(
|
||||
"Free beamline",
|
||||
"This will clear the beamline busy flag.\n\nDo you want to continue?",
|
||||
):
|
||||
return
|
||||
code = self._prompt_recovery_code("Free beamline")
|
||||
if not code:
|
||||
return
|
||||
self._daq.free_beamline(code)
|
||||
|
||||
@Slot()
|
||||
def _recover_beamline(self) -> None:
|
||||
if self._sample_appears_mounted():
|
||||
if not self._confirm(
|
||||
"Recover beamline",
|
||||
"A sample appears to be mounted.\n\n"
|
||||
"Recovering the beamline may damage the sample or leave hardware in an unsafe state.\n\n"
|
||||
"Only continue if you are sure this is the correct recovery action.",
|
||||
):
|
||||
return
|
||||
else:
|
||||
if not self._confirm(
|
||||
"Recover beamline",
|
||||
"This will take over the beamline, clear the busy flag, and set the state to Maintenance.\n\n"
|
||||
"Do you want to continue?",
|
||||
):
|
||||
return
|
||||
|
||||
code = self._prompt_recovery_code("Recover beamline")
|
||||
if not code:
|
||||
return
|
||||
self._daq.recover_beamline(code)
|
||||
|
||||
@Slot()
|
||||
def _recovery_unmount_sample(self) -> None:
|
||||
if not self._confirm(
|
||||
"Unmount sample (recovery)",
|
||||
"This will force-take the session and attempt a controlled recovery unmount.\n\n"
|
||||
"Use this only if normal unmount is not possible.",
|
||||
):
|
||||
return
|
||||
|
||||
code = self._prompt_recovery_code("Unmount sample (recovery)")
|
||||
if not code:
|
||||
return
|
||||
self._daq.recovery_unmount_sample(code)
|
||||
|
||||
|
||||
class BeamlineRecoveryDialog(QDialog):
|
||||
def __init__(self, *, daq: DAQWorker, parent=None):
|
||||
super().__init__(parent)
|
||||
self.setWindowTitle("Beamline Recovery")
|
||||
self.setMinimumSize(560, 420)
|
||||
|
||||
layout = QVBoxLayout(self)
|
||||
layout.setContentsMargins(12, 12, 12, 12)
|
||||
layout.setSpacing(8)
|
||||
|
||||
self._panel = RecoveryPanel(daq=daq, parent=self)
|
||||
layout.addWidget(self._panel, 1)
|
||||
|
||||
buttons = QDialogButtonBox(QDialogButtonBox.StandardButton.Close, parent=self)
|
||||
buttons.rejected.connect(self.reject)
|
||||
buttons.accepted.connect(self.accept)
|
||||
layout.addWidget(buttons)
|
||||
@@ -29,6 +29,7 @@ from aare.common.logger_config import QtLogEmitter, QtLogHandler, find_existing_
|
||||
from aare.gui.threads.daq_worker import DAQWorker
|
||||
|
||||
|
||||
|
||||
class DeveloperHelpDialog(QDialog):
|
||||
def __init__(self, *, daq: DAQWorker, is_staff: bool, parent=None):
|
||||
super().__init__(parent)
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
# Python
|
||||
from PySide6.QtCore import Signal
|
||||
from PySide6.QtWidgets import QWidget, QGridLayout, QLabel, QPushButton, QVBoxLayout
|
||||
from matplotlib.backends.backend_qt5agg import FigureCanvasQTAgg as FigureCanvas
|
||||
@@ -12,6 +11,7 @@ from aare.common.logger_config import setup_logger
|
||||
|
||||
logger = setup_logger("aareGUI")
|
||||
|
||||
|
||||
class FaceDetectionPanel(QWidget):
|
||||
|
||||
face_detection = Signal(int, int)
|
||||
@@ -20,6 +20,7 @@ class FaceDetectionPanel(QWidget):
|
||||
super().__init__(parent)
|
||||
self.steps = 14
|
||||
self.step_size = 15
|
||||
self._manual_run_requested = False
|
||||
|
||||
def _set_steps(val: float):
|
||||
self.steps = int(val)
|
||||
@@ -29,27 +30,22 @@ class FaceDetectionPanel(QWidget):
|
||||
|
||||
self.fig = Figure(figsize=(5, 4), tight_layout=True)
|
||||
self.canvas = FigureCanvas(self.fig)
|
||||
self.ax1 = self.fig.add_subplot(2, 1, 1) # Height vs angle
|
||||
self.ax2 = self.fig.add_subplot(2, 1, 2) # Area vs angle
|
||||
self.ax1 = self.fig.add_subplot(2, 1, 1)
|
||||
self.ax2 = self.fig.add_subplot(2, 1, 2)
|
||||
|
||||
self._top_layout = QGridLayout()
|
||||
self._top_layout.addWidget(TitleLabel("TELL sample changer", self), 0, 0, 1, 3)
|
||||
self.status_lbl = QLabel("")
|
||||
self.status_lbl = QLabel("Idle")
|
||||
self._top_layout.addWidget(self.status_lbl, 0, 2)
|
||||
|
||||
self._top_layout.addWidget(QLabel("step size", parent=self), 1, 0)
|
||||
self.step_size_enter = NumberLineEdit(
|
||||
0, 50, 15, decimals=4, parent=self
|
||||
)
|
||||
|
||||
self.step_size_enter = NumberLineEdit(0, 50, 15, decimals=4, parent=self)
|
||||
self._top_layout.addWidget(self.step_size_enter, 1, 1, 1, 3)
|
||||
self.step_size_enter.newValue.connect(_set_step_size)
|
||||
self._top_layout.addWidget(QLabel("°", parent=self), 1, 4)
|
||||
|
||||
self._top_layout.addWidget(QLabel("number of steps", parent=self), 2, 0)
|
||||
self.steps_enter = NumberLineEdit(
|
||||
0, 50, 14, decimals=4, parent=self
|
||||
)
|
||||
self.steps_enter = NumberLineEdit(0, 50, 14, decimals=4, parent=self)
|
||||
self._top_layout.addWidget(self.steps_enter, 2, 1, 1, 3)
|
||||
self._top_layout.addWidget(QLabel("", parent=self), 2, 4)
|
||||
self.steps_enter.newValue.connect(_set_steps)
|
||||
@@ -69,49 +65,68 @@ class FaceDetectionPanel(QWidget):
|
||||
|
||||
def run_and_refresh(self):
|
||||
try:
|
||||
self._manual_run_requested = True
|
||||
self.status_lbl.setText("Starting...")
|
||||
self.face_detection_button.setEnabled(False)
|
||||
self.face_detection.emit(int(self.steps), int(self.step_size))
|
||||
except Exception as e:
|
||||
self._manual_run_requested = False
|
||||
self.status_lbl.setText(f"Error: {e}")
|
||||
self.face_detection_button.setEnabled(True)
|
||||
|
||||
def update_plot(self, data):
|
||||
samples = data.get("samples", [])
|
||||
print(samples)
|
||||
samples = data.get("samples", []) or []
|
||||
running = bool(data.get("running", False))
|
||||
angle = data.get("current_angle_deg")
|
||||
status = data.get("status", "")
|
||||
|
||||
if running:
|
||||
if self._manual_run_requested:
|
||||
self.status_lbl.setText(f"Running... angle {angle}" if angle is not None else "Running...")
|
||||
else:
|
||||
self.status_lbl.setText(f"Automation running... angle {angle}" if angle is not None else "Automation running...")
|
||||
else:
|
||||
if self._manual_run_requested:
|
||||
self.status_lbl.setText("Done")
|
||||
self.face_detection_button.setEnabled(True)
|
||||
self._manual_run_requested = False
|
||||
elif samples:
|
||||
self.status_lbl.setText("Showing latest result")
|
||||
else:
|
||||
self.status_lbl.setText("Idle")
|
||||
|
||||
self.ax1.clear()
|
||||
self.ax2.clear()
|
||||
|
||||
if not samples:
|
||||
self.ax1.clear()
|
||||
self.ax2.clear()
|
||||
self.ax1.text(0.5, 0.5, "No data", ha="center", va="center")
|
||||
self.ax2.text(0.5, 0.5, "No data", ha="center", va="center")
|
||||
self.canvas.draw_idle()
|
||||
logger.info("No data")
|
||||
return
|
||||
|
||||
angles = np.array([s["angle_deg"] for s in samples], dtype=float)
|
||||
heights = np.array([s["height"] for s in samples], dtype=float)
|
||||
areas = np.array([s["area"] for s in samples], dtype=float)
|
||||
self.ax1.clear()
|
||||
self.ax2.clear()
|
||||
|
||||
self.ax1.scatter(angles, heights, s=16, c="tab:blue", label="Height")
|
||||
self.ax2.scatter(angles, areas, s=16, c="tab:green", label="Area")
|
||||
|
||||
ang_grid = np.linspace(angles.min(),angles.max(), 400)
|
||||
#ang_grid_wrapped = ((ang_grid + 180) % 360) - 180
|
||||
hf = data.get("height_fit", {})
|
||||
ang_grid = np.linspace(angles.min(), angles.max(), 400)
|
||||
|
||||
hf = data.get("height_fit", {}) or {}
|
||||
if {"A", "B", "phi_rad", "C"} <= hf.keys():
|
||||
A, B, phi, C = hf["A"], hf["B"], hf["phi_rad"], hf["C"]
|
||||
height_fit = A + B * np.cos(C*np.deg2rad(ang_grid) - phi)
|
||||
height_fit = A + B * np.cos(C * np.deg2rad(ang_grid) - phi)
|
||||
self.ax1.plot(ang_grid, height_fit, color="tab:orange", label="Height fit")
|
||||
if "best_angle_deg" in hf:
|
||||
logger.info(f"best angle: {hf['best_angle_deg']}")
|
||||
self.ax1.axvline(hf["best_angle_deg"], color="tab:orange", ls="--", alpha=0.6)
|
||||
|
||||
af = data.get("area_fit", {})
|
||||
if {"A", "B", "phi_rad", "C"} <= hf.keys():
|
||||
af = data.get("area_fit", {}) or {}
|
||||
if {"A", "B", "phi_rad", "C"} <= af.keys():
|
||||
A2, B2, phi2, C2 = af["A"], af["B"], af["phi_rad"], af["C"]
|
||||
area_fit = A2 + B2 * np.cos(C2 * np.deg2rad(ang_grid) - phi2)
|
||||
self.ax2.plot(ang_grid, area_fit, color="tab:red", label="Area fit")
|
||||
if "best_angle_deg" in af:
|
||||
logger.info(f"best angle: {af['best_angle_deg']}")
|
||||
self.ax2.axvline(af["best_angle_deg"], color="tab:red", ls="--", alpha=0.6)
|
||||
|
||||
self.ax1.set_xlabel("Angle (deg)")
|
||||
|
||||
@@ -182,6 +182,7 @@ class FilePathPanel(QWidget):
|
||||
self.__puck_pos = 99
|
||||
self.directory_edit.setText(f"{self.__formatted_date}/test")
|
||||
else:
|
||||
self.__sample_id = sample.db_id
|
||||
self.__sample_name = sample.sample_name
|
||||
self.__dewar_pos = sample.loc_str()
|
||||
self.__puck_name = sample.puck_name
|
||||
|
||||
@@ -85,7 +85,8 @@ class SmargonPanel(QWidget):
|
||||
|
||||
@Slot()
|
||||
def home(self):
|
||||
self.smargon.emit(Smargon.SMARGON_HOME)
|
||||
#TODO move SMARGON_HOME to REDIS, allow GUI to read this value
|
||||
self.smargon.emit(SmargonCoordinate(sh_mm=Coordinate(x=0, y=0, z=18), phi_deg=0, chi_deg=0))
|
||||
|
||||
@Slot(float)
|
||||
def phi(self, f: float):
|
||||
|
||||
@@ -0,0 +1,639 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
from math import sqrt
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
from matplotlib.backends.backend_qtagg import FigureCanvasQTAgg as FigureCanvas
|
||||
from matplotlib.figure import Figure
|
||||
from PySide6.QtCore import QSettings, QTimer, Qt
|
||||
from PySide6.QtWidgets import (
|
||||
QCheckBox,
|
||||
QComboBox,
|
||||
QFileDialog,
|
||||
QFormLayout,
|
||||
QGroupBox,
|
||||
QHBoxLayout,
|
||||
QLabel,
|
||||
QPushButton,
|
||||
QTableWidget,
|
||||
QTableWidgetItem,
|
||||
QVBoxLayout,
|
||||
QWidget,
|
||||
)
|
||||
|
||||
|
||||
class SmargonTracePanel(QWidget):
|
||||
HOME_X_MM = 0.0
|
||||
HOME_Y_MM = 0.0
|
||||
HOME_Z_MM = 18.0
|
||||
TABLE_MAX_ROWS = 20
|
||||
|
||||
COLOR_X = "tab:red"
|
||||
COLOR_Y = "tab:green"
|
||||
COLOR_Z = "tab:blue"
|
||||
COLOR_DISTANCE = "tab:purple"
|
||||
|
||||
SETTINGS_GROUP = "smargon_trace_panel"
|
||||
|
||||
def __init__(self, csv_path: str | Path = "logs/smargon_trace.csv", parent=None):
|
||||
super().__init__(parent)
|
||||
|
||||
self._csv_path = Path(csv_path)
|
||||
self._active_csv_path: Path | None = None
|
||||
self._last_mtime_ns: int | None = None
|
||||
self._last_rows: list[dict[str, object]] = []
|
||||
self._last_distances: list[float] = []
|
||||
self._last_lengths: list[float] = []
|
||||
|
||||
self._status = QLabel("Waiting for smargon trace data...")
|
||||
self._status.setWordWrap(True)
|
||||
|
||||
self._summary = QLabel(
|
||||
f"Home position: "
|
||||
f"X={self.HOME_X_MM:.3f} mm, "
|
||||
f"Y={self.HOME_Y_MM:.3f} mm, "
|
||||
f"Z={self.HOME_Z_MM:.3f} mm"
|
||||
)
|
||||
self._summary.setWordWrap(True)
|
||||
|
||||
self._metrics = QLabel("Latest metrics: distance=n/a | length=n/a")
|
||||
self._metrics.setWordWrap(True)
|
||||
|
||||
self._show_summary_cb = QCheckBox("Summary")
|
||||
self._show_delta_cb = QCheckBox("Delta")
|
||||
self._show_absolute_cb = QCheckBox("Absolute")
|
||||
self._show_distance_cb = QCheckBox("Total distance")
|
||||
self._show_table_cb = QCheckBox("Table")
|
||||
|
||||
self._show_summary_cb.setChecked(True)
|
||||
self._show_delta_cb.setChecked(True)
|
||||
self._show_absolute_cb.setChecked(False)
|
||||
self._show_distance_cb.setChecked(False)
|
||||
self._show_table_cb.setChecked(False)
|
||||
|
||||
self._units_combo = QComboBox()
|
||||
self._units_combo.addItems(["mm", "µm"])
|
||||
self._units_combo.setCurrentText("mm")
|
||||
self._units_combo.setMaximumWidth(90)
|
||||
|
||||
self._export_btn = QPushButton("Export table CSV")
|
||||
self._export_btn.clicked.connect(self._export_table_csv)
|
||||
|
||||
display_group = QGroupBox("Display")
|
||||
display_layout = QHBoxLayout(display_group)
|
||||
display_layout.setContentsMargins(8, 6, 8, 6)
|
||||
display_layout.setSpacing(10)
|
||||
display_layout.addWidget(self._show_summary_cb)
|
||||
display_layout.addWidget(self._show_delta_cb)
|
||||
display_layout.addWidget(self._show_absolute_cb)
|
||||
display_layout.addWidget(self._show_distance_cb)
|
||||
display_layout.addWidget(self._show_table_cb)
|
||||
display_layout.addStretch()
|
||||
display_layout.addWidget(QLabel("Units:"))
|
||||
display_layout.addWidget(self._units_combo)
|
||||
display_layout.addWidget(self._export_btn)
|
||||
|
||||
self._summary_group = QGroupBox("Summary")
|
||||
summary_layout = QFormLayout(self._summary_group)
|
||||
summary_layout.setContentsMargins(8, 6, 8, 6)
|
||||
summary_layout.setSpacing(6)
|
||||
summary_layout.addRow("Status:", self._status)
|
||||
summary_layout.addRow("Position:", self._summary)
|
||||
summary_layout.addRow("Metrics:", self._metrics)
|
||||
|
||||
self._figure = Figure(figsize=(7, 7))
|
||||
self._canvas = FigureCanvas(self._figure)
|
||||
|
||||
self._table = QTableWidget(self)
|
||||
self._table.setColumnCount(8)
|
||||
self._table.setHorizontalHeaderLabels(
|
||||
["Point", "Event", "Sample", "X", "Y", "Z", "Distance", "Length"]
|
||||
)
|
||||
self._table.verticalHeader().setVisible(False)
|
||||
self._table.setAlternatingRowColors(True)
|
||||
self._table.setVisible(False)
|
||||
|
||||
layout = QVBoxLayout(self)
|
||||
layout.setContentsMargins(6, 6, 6, 6)
|
||||
layout.setSpacing(8)
|
||||
layout.addWidget(display_group)
|
||||
layout.addWidget(self._summary_group)
|
||||
layout.addWidget(self._canvas)
|
||||
layout.addWidget(self._table)
|
||||
|
||||
self._show_summary_cb.toggled.connect(self._on_controls_changed)
|
||||
self._show_delta_cb.toggled.connect(self._on_controls_changed)
|
||||
self._show_absolute_cb.toggled.connect(self._on_controls_changed)
|
||||
self._show_distance_cb.toggled.connect(self._on_controls_changed)
|
||||
self._show_table_cb.toggled.connect(self._on_controls_changed)
|
||||
self._units_combo.currentTextChanged.connect(lambda _text: self._on_controls_changed())
|
||||
|
||||
self._load_settings()
|
||||
|
||||
self._timer = QTimer(self)
|
||||
self._timer.setInterval(1000)
|
||||
self._timer.timeout.connect(self.refresh_plot)
|
||||
self._timer.start()
|
||||
|
||||
self._apply_visibility_settings()
|
||||
self.refresh_plot(force=True)
|
||||
|
||||
def refresh_plot(self, force: bool = False) -> None:
|
||||
csv_path = self._resolve_csv_path()
|
||||
|
||||
if csv_path is None or not csv_path.exists():
|
||||
self._active_csv_path = None
|
||||
self._last_mtime_ns = None
|
||||
self._last_rows = []
|
||||
self._last_distances = []
|
||||
self._last_lengths = []
|
||||
self._status.setText(
|
||||
"No trace file found. Tried: "
|
||||
+ ", ".join(str(p) for p in self._candidate_paths())
|
||||
)
|
||||
self._summary.setText(
|
||||
f"Home position: "
|
||||
f"X={self.HOME_X_MM:.3f} mm, "
|
||||
f"Y={self.HOME_Y_MM:.3f} mm, "
|
||||
f"Z={self.HOME_Z_MM:.3f} mm"
|
||||
)
|
||||
self._metrics.setText("Latest metrics: distance=n/a | length=n/a")
|
||||
self._draw_empty("No smargon trace file found yet")
|
||||
self._clear_table()
|
||||
return
|
||||
|
||||
try:
|
||||
stat = csv_path.stat()
|
||||
if (
|
||||
not force
|
||||
and self._active_csv_path == csv_path
|
||||
and self._last_mtime_ns == stat.st_mtime_ns
|
||||
):
|
||||
return
|
||||
|
||||
self._active_csv_path = csv_path
|
||||
self._last_mtime_ns = stat.st_mtime_ns
|
||||
|
||||
rows = self._read_rows(csv_path)
|
||||
if not rows:
|
||||
self._last_rows = []
|
||||
self._last_distances = []
|
||||
self._last_lengths = []
|
||||
self._status.setText(f"Trace file is empty: {csv_path}")
|
||||
self._summary.setText(
|
||||
f"Home position: "
|
||||
f"X={self.HOME_X_MM:.3f} mm, "
|
||||
f"Y={self.HOME_Y_MM:.3f} mm, "
|
||||
f"Z={self.HOME_Z_MM:.3f} mm"
|
||||
)
|
||||
self._metrics.setText("Latest metrics: distance=n/a | length=n/a")
|
||||
self._draw_empty("Smargon trace file is empty")
|
||||
self._clear_table()
|
||||
return
|
||||
|
||||
unit_name, unit_scale = self._unit_settings()
|
||||
|
||||
x = list(range(1, len(rows) + 1))
|
||||
|
||||
shx_mm = [row["shx_mm"] for row in rows]
|
||||
shy_mm = [row["shy_mm"] for row in rows]
|
||||
shz_mm = [row["shz_mm"] for row in rows]
|
||||
|
||||
dx_mm = [value - self.HOME_X_MM for value in shx_mm]
|
||||
dy_mm = [value - self.HOME_Y_MM for value in shy_mm]
|
||||
dz_mm = [value - self.HOME_Z_MM for value in shz_mm]
|
||||
|
||||
shx = [value * unit_scale for value in shx_mm]
|
||||
shy = [value * unit_scale for value in shy_mm]
|
||||
shz = [value * unit_scale for value in shz_mm]
|
||||
|
||||
dx = [value * unit_scale for value in dx_mm]
|
||||
dy = [value * unit_scale for value in dy_mm]
|
||||
dz = [value * unit_scale for value in dz_mm]
|
||||
|
||||
distances_mm = [
|
||||
sqrt(dx0**2 + dy0**2 + dz0**2)
|
||||
for dx0, dy0, dz0 in zip(dx_mm, dy_mm, dz_mm)
|
||||
]
|
||||
distances = [value * unit_scale for value in distances_mm]
|
||||
|
||||
lengths_mm = [self._projected_length_mm(row) for row in rows]
|
||||
lengths = [value * unit_scale for value in lengths_mm]
|
||||
|
||||
self._last_rows = rows
|
||||
self._last_distances = distances
|
||||
self._last_lengths = lengths
|
||||
|
||||
self._redraw_plots(
|
||||
x=x,
|
||||
shx=shx,
|
||||
shy=shy,
|
||||
shz=shz,
|
||||
dx=dx,
|
||||
dy=dy,
|
||||
dz=dz,
|
||||
distances=distances,
|
||||
unit_name=unit_name,
|
||||
)
|
||||
|
||||
last = rows[-1]
|
||||
last_dx_mm = dx_mm[-1]
|
||||
last_dy_mm = dy_mm[-1]
|
||||
last_dz_mm = dz_mm[-1]
|
||||
|
||||
distance_value = distances[-1]
|
||||
length_value = lengths[-1]
|
||||
|
||||
self._status.setText(
|
||||
f"Loaded {len(rows)} points from {csv_path} | "
|
||||
f"last event={last['event']} | "
|
||||
f"sample_id={last['sample_id']}"
|
||||
)
|
||||
self._summary.setText(
|
||||
f"Home: X={self.HOME_X_MM * unit_scale:.3f}, "
|
||||
f"Y={self.HOME_Y_MM * unit_scale:.3f}, "
|
||||
f"Z={self.HOME_Z_MM * unit_scale:.3f} {unit_name} | "
|
||||
f"Latest: X={last['shx_mm'] * unit_scale:.5f}, "
|
||||
f"Y={last['shy_mm'] * unit_scale:.5f}, "
|
||||
f"Z={last['shz_mm'] * unit_scale:.5f} {unit_name} | "
|
||||
f"Δ: X={last_dx_mm * unit_scale:+.5f}, "
|
||||
f"Y={last_dy_mm * unit_scale:+.5f}, "
|
||||
f"Z={last_dz_mm * unit_scale:+.5f} {unit_name}"
|
||||
)
|
||||
self._metrics.setText(
|
||||
f"distance={distance_value:.5f} {unit_name} | "
|
||||
f"length={length_value:.5f} {unit_name} (beamline-plane projected)"
|
||||
)
|
||||
|
||||
self._populate_table(
|
||||
rows=rows,
|
||||
distances=distances,
|
||||
lengths=lengths,
|
||||
unit_name=unit_name,
|
||||
unit_scale=unit_scale,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
self._last_rows = []
|
||||
self._last_distances = []
|
||||
self._last_lengths = []
|
||||
self._status.setText(f"Failed to load trace from {csv_path}: {e}")
|
||||
self._summary.setText(
|
||||
f"Home position: "
|
||||
f"X={self.HOME_X_MM:.3f} mm, "
|
||||
f"Y={self.HOME_Y_MM:.3f} mm, "
|
||||
f"Z={self.HOME_Z_MM:.3f} mm"
|
||||
)
|
||||
self._metrics.setText("Latest metrics: distance=n/a | length=n/a")
|
||||
self._draw_empty("Failed to parse smargon trace")
|
||||
self._clear_table()
|
||||
|
||||
def showEvent(self, event) -> None:
|
||||
super().showEvent(event)
|
||||
self.refresh_plot(force=True)
|
||||
|
||||
def _on_controls_changed(self) -> None:
|
||||
self._save_settings()
|
||||
self._apply_visibility_settings()
|
||||
|
||||
def _load_settings(self) -> None:
|
||||
settings = QSettings()
|
||||
settings.beginGroup(self.SETTINGS_GROUP)
|
||||
self._show_summary_cb.setChecked(settings.value("show_summary", True, type=bool))
|
||||
self._show_delta_cb.setChecked(settings.value("show_delta", True, type=bool))
|
||||
self._show_absolute_cb.setChecked(settings.value("show_absolute", False, type=bool))
|
||||
self._show_distance_cb.setChecked(settings.value("show_distance", False, type=bool))
|
||||
self._show_table_cb.setChecked(settings.value("show_table", False, type=bool))
|
||||
self._units_combo.setCurrentText(settings.value("units", "mm", type=str))
|
||||
settings.endGroup()
|
||||
|
||||
def _save_settings(self) -> None:
|
||||
settings = QSettings()
|
||||
settings.beginGroup(self.SETTINGS_GROUP)
|
||||
settings.setValue("show_summary", self._show_summary_cb.isChecked())
|
||||
settings.setValue("show_delta", self._show_delta_cb.isChecked())
|
||||
settings.setValue("show_absolute", self._show_absolute_cb.isChecked())
|
||||
settings.setValue("show_distance", self._show_distance_cb.isChecked())
|
||||
settings.setValue("show_table", self._show_table_cb.isChecked())
|
||||
settings.setValue("units", self._units_combo.currentText())
|
||||
settings.endGroup()
|
||||
|
||||
def _apply_visibility_settings(self) -> None:
|
||||
if not (
|
||||
self._show_delta_cb.isChecked()
|
||||
or self._show_absolute_cb.isChecked()
|
||||
or self._show_distance_cb.isChecked()
|
||||
):
|
||||
self._show_delta_cb.blockSignals(True)
|
||||
self._show_delta_cb.setChecked(True)
|
||||
self._show_delta_cb.blockSignals(False)
|
||||
|
||||
self._summary_group.setVisible(self._show_summary_cb.isChecked())
|
||||
self._table.setVisible(self._show_table_cb.isChecked())
|
||||
self.refresh_plot(force=True)
|
||||
|
||||
def _export_table_csv(self) -> None:
|
||||
if not self._last_rows:
|
||||
return
|
||||
|
||||
unit_name, unit_scale = self._unit_settings()
|
||||
file_path, _ = QFileDialog.getSaveFileName(
|
||||
self,
|
||||
"Export Smargon Trace Table",
|
||||
"smargon_trace_export.csv",
|
||||
"CSV Files (*.csv)",
|
||||
)
|
||||
if not file_path:
|
||||
return
|
||||
|
||||
with open(file_path, "w", encoding="utf-8", newline="") as f:
|
||||
writer = csv.writer(f)
|
||||
writer.writerow([
|
||||
"point",
|
||||
"timestamp",
|
||||
"event",
|
||||
"sample_id",
|
||||
f"shx_{unit_name}",
|
||||
f"shy_{unit_name}",
|
||||
f"shz_{unit_name}",
|
||||
f"distance_{unit_name}",
|
||||
f"length_{unit_name}",
|
||||
"omega_deg",
|
||||
"phi_deg",
|
||||
"chi_deg",
|
||||
])
|
||||
|
||||
for idx, (row, distance_value, length_value) in enumerate(
|
||||
zip(self._last_rows, self._last_distances, self._last_lengths),
|
||||
start=1,
|
||||
):
|
||||
writer.writerow([
|
||||
idx,
|
||||
row["timestamp"],
|
||||
row["event"],
|
||||
row["sample_id"],
|
||||
float(row["shx_mm"]) * unit_scale,
|
||||
float(row["shy_mm"]) * unit_scale,
|
||||
float(row["shz_mm"]) * unit_scale,
|
||||
distance_value,
|
||||
length_value,
|
||||
row["omega_deg"],
|
||||
row["phi_deg"],
|
||||
row["chi_deg"],
|
||||
])
|
||||
|
||||
def _redraw_plots(
|
||||
self,
|
||||
*,
|
||||
x: list[int],
|
||||
shx: list[float],
|
||||
shy: list[float],
|
||||
shz: list[float],
|
||||
dx: list[float],
|
||||
dy: list[float],
|
||||
dz: list[float],
|
||||
distances: list[float],
|
||||
unit_name: str,
|
||||
) -> None:
|
||||
self._figure.clear()
|
||||
|
||||
enabled = []
|
||||
if self._show_delta_cb.isChecked():
|
||||
enabled.append("delta")
|
||||
if self._show_absolute_cb.isChecked():
|
||||
enabled.append("absolute")
|
||||
if self._show_distance_cb.isChecked():
|
||||
enabled.append("distance")
|
||||
|
||||
axes = self._figure.subplots(len(enabled), 1, squeeze=False)
|
||||
axes_list = [row[0] for row in axes]
|
||||
|
||||
for ax, plot_name in zip(axes_list, enabled):
|
||||
if plot_name == "delta":
|
||||
ax.plot(x, dx, marker="o", color=self.COLOR_X, label=f"ΔSHX [{unit_name}]")
|
||||
ax.plot(x, dy, marker="o", color=self.COLOR_Y, label=f"ΔSHY [{unit_name}]")
|
||||
ax.plot(x, dz, marker="o", color=self.COLOR_Z, label=f"ΔSHZ [{unit_name}]")
|
||||
ax.axhline(0.0, color="black", linewidth=1.0, alpha=0.5)
|
||||
ax.set_title("Smargon displacement from home")
|
||||
ax.set_ylabel(f"Δ position [{unit_name}]")
|
||||
ax.grid(True, alpha=0.3)
|
||||
ax.legend(loc="best")
|
||||
|
||||
elif plot_name == "absolute":
|
||||
ax.plot(x, shx, marker="o", color=self.COLOR_X, label=f"SHX [{unit_name}]")
|
||||
ax.plot(x, shy, marker="o", color=self.COLOR_Y, label=f"SHY [{unit_name}]")
|
||||
ax.plot(x, shz, marker="o", color=self.COLOR_Z, label=f"SHZ [{unit_name}]")
|
||||
ax.set_title("Smargon absolute position")
|
||||
ax.set_ylabel(f"Position [{unit_name}]")
|
||||
ax.grid(True, alpha=0.3)
|
||||
ax.legend(loc="best")
|
||||
|
||||
elif plot_name == "distance":
|
||||
ax.plot(x, distances, marker="o", color=self.COLOR_DISTANCE, label=f"Total distance [{unit_name}]")
|
||||
ax.set_title("Total distance from home")
|
||||
ax.set_ylabel(f"Distance [{unit_name}]")
|
||||
ax.grid(True, alpha=0.3)
|
||||
ax.legend(loc="best")
|
||||
|
||||
ax.set_xlabel("Trace point")
|
||||
|
||||
self._figure.tight_layout()
|
||||
self._canvas.draw_idle()
|
||||
|
||||
def _populate_table(
|
||||
self,
|
||||
*,
|
||||
rows: list[dict[str, object]],
|
||||
distances: list[float],
|
||||
lengths: list[float],
|
||||
unit_name: str,
|
||||
unit_scale: float,
|
||||
) -> None:
|
||||
recent_rows = rows[-self.TABLE_MAX_ROWS:]
|
||||
recent_distances = distances[-self.TABLE_MAX_ROWS:]
|
||||
recent_lengths = lengths[-self.TABLE_MAX_ROWS:]
|
||||
|
||||
self._table.setRowCount(len(recent_rows))
|
||||
self._table.setHorizontalHeaderLabels(
|
||||
[
|
||||
"Point",
|
||||
"Event",
|
||||
"Sample",
|
||||
f"X [{unit_name}]",
|
||||
f"Y [{unit_name}]",
|
||||
f"Z [{unit_name}]",
|
||||
f"Distance [{unit_name}]",
|
||||
f"Length [{unit_name}]",
|
||||
]
|
||||
)
|
||||
|
||||
start_idx = len(rows) - len(recent_rows) + 1
|
||||
for row_idx, (row, distance_value, length_value) in enumerate(
|
||||
zip(recent_rows, recent_distances, recent_lengths)
|
||||
):
|
||||
values = [
|
||||
str(start_idx + row_idx),
|
||||
str(row["event"]),
|
||||
str(row["sample_id"]),
|
||||
f"{float(row['shx_mm']) * unit_scale:.5f}",
|
||||
f"{float(row['shy_mm']) * unit_scale:.5f}",
|
||||
f"{float(row['shz_mm']) * unit_scale:.5f}",
|
||||
f"{distance_value:.5f}",
|
||||
f"{length_value:.5f}",
|
||||
]
|
||||
for col_idx, value in enumerate(values):
|
||||
item = QTableWidgetItem(value)
|
||||
item.setTextAlignment(Qt.AlignmentFlag.AlignCenter)
|
||||
self._table.setItem(row_idx, col_idx, item)
|
||||
|
||||
self._table.resizeColumnsToContents()
|
||||
|
||||
def _clear_table(self) -> None:
|
||||
self._table.setRowCount(0)
|
||||
|
||||
def _unit_settings(self) -> tuple[str, float]:
|
||||
if self._units_combo.currentText() == "µm":
|
||||
return "µm", 1000.0
|
||||
return "mm", 1.0
|
||||
|
||||
def _candidate_paths(self) -> list[Path]:
|
||||
here = Path(__file__).resolve()
|
||||
project_root = here.parents[4]
|
||||
|
||||
candidates = [
|
||||
Path.cwd() / self._csv_path,
|
||||
project_root / self._csv_path,
|
||||
project_root / "src" / "aare" / "daq" / "logs" / "smargon_trace.csv",
|
||||
project_root / "src" / "aare" / "gui" / "logs" / "smargon_trace.csv",
|
||||
]
|
||||
|
||||
out: list[Path] = []
|
||||
seen: set[str] = set()
|
||||
for path in candidates:
|
||||
key = str(path.resolve()) if path.exists() else str(path)
|
||||
if key not in seen:
|
||||
out.append(path)
|
||||
seen.add(key)
|
||||
return out
|
||||
|
||||
def _resolve_csv_path(self) -> Path | None:
|
||||
existing = [p for p in self._candidate_paths() if p.exists()]
|
||||
if not existing:
|
||||
return None
|
||||
return max(existing, key=lambda p: p.stat().st_mtime_ns)
|
||||
|
||||
def _read_rows(self, csv_path: Path) -> list[dict[str, object]]:
|
||||
rows: list[dict[str, object]] = []
|
||||
|
||||
with csv_path.open("r", encoding="utf-8", newline="") as f:
|
||||
reader = csv.DictReader(f)
|
||||
for row in reader:
|
||||
try:
|
||||
rows.append(
|
||||
{
|
||||
"timestamp": row.get("timestamp", row.get("ts", "")),
|
||||
"event": row.get("event", ""),
|
||||
"sample_id": row.get("sample_id", ""),
|
||||
"omega_deg": float(row.get("omega_deg", "nan")),
|
||||
"zoom": float(row.get("zoom", "nan")),
|
||||
"shx_mm": float(row.get("shx_mm", "nan")),
|
||||
"shy_mm": float(row.get("shy_mm", "nan")),
|
||||
"shz_mm": float(row.get("shz_mm", "nan")),
|
||||
"phi_deg": float(row.get("phi_deg", "nan")),
|
||||
"chi_deg": float(row.get("chi_deg", "nan")),
|
||||
}
|
||||
)
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
|
||||
return rows
|
||||
|
||||
def _projected_length_mm(self, row: dict[str, object]) -> float:
|
||||
rel = np.array(
|
||||
[
|
||||
float(row["shx_mm"]) - self.HOME_X_MM,
|
||||
float(row["shy_mm"]) - self.HOME_Y_MM,
|
||||
float(row["shz_mm"]) - self.HOME_Z_MM,
|
||||
],
|
||||
dtype=float,
|
||||
)
|
||||
|
||||
vec_x = self._smargon_nudge_basis(
|
||||
axis="x",
|
||||
omega_deg=float(row["omega_deg"]),
|
||||
phi_deg=float(row["phi_deg"]),
|
||||
chi_deg=float(row["chi_deg"]),
|
||||
)
|
||||
vec_y = self._smargon_nudge_basis(
|
||||
axis="y",
|
||||
omega_deg=float(row["omega_deg"]),
|
||||
phi_deg=float(row["phi_deg"]),
|
||||
chi_deg=float(row["chi_deg"]),
|
||||
)
|
||||
|
||||
beam_x = float(np.dot(rel, vec_x))
|
||||
beam_y = float(np.dot(rel, vec_y))
|
||||
|
||||
return sqrt(beam_x**2 + beam_y**2)
|
||||
|
||||
def _smargon_nudge_basis(
|
||||
self,
|
||||
*,
|
||||
axis: str,
|
||||
omega_deg: float,
|
||||
phi_deg: float,
|
||||
chi_deg: float,
|
||||
) -> np.ndarray:
|
||||
phi = np.radians(np.around(phi_deg, decimals=1))
|
||||
chi = np.radians(np.around(chi_deg, decimals=1))
|
||||
omega = np.radians(np.around(omega_deg, decimals=1))
|
||||
|
||||
if axis == "x":
|
||||
coord_x, coord_y, coord_z = 1.0, 0.0, 0.0
|
||||
elif axis == "y":
|
||||
coord_x, coord_y, coord_z = 0.0, 1.0, 0.0
|
||||
else:
|
||||
raise ValueError(f"Unsupported axis: {axis}")
|
||||
|
||||
co = np.cos(omega)
|
||||
so = np.sin(omega)
|
||||
cp = np.cos(phi)
|
||||
sp = np.sin(phi)
|
||||
cc = np.cos(chi)
|
||||
sc = np.sin(chi)
|
||||
|
||||
offset_x = (
|
||||
-coord_z * co * sp
|
||||
- coord_y * so * sp
|
||||
+ coord_x * cp * sc
|
||||
- coord_y * co * cc * cp
|
||||
+ coord_z * so * cc * cp
|
||||
)
|
||||
offset_y = (
|
||||
-coord_z * co * cp
|
||||
- coord_y * so * cp
|
||||
- coord_x * sc * sp
|
||||
+ coord_y * co * cc * sp
|
||||
- coord_z * so * cc * sp
|
||||
)
|
||||
offset_z = -coord_x * cc - coord_y * co * sc + coord_z * so * sc
|
||||
|
||||
vec = np.array([offset_x, offset_y, offset_z], dtype=float)
|
||||
norm = np.linalg.norm(vec)
|
||||
|
||||
if norm == 0.0:
|
||||
return np.array([0.0, 0.0, 0.0], dtype=float)
|
||||
|
||||
return vec / norm
|
||||
|
||||
def _draw_empty(self, message: str) -> None:
|
||||
self._figure.clear()
|
||||
ax = self._figure.add_subplot(111)
|
||||
ax.set_title("Smargon trace")
|
||||
ax.text(0.5, 0.5, message, ha="center", va="center", transform=ax.transAxes)
|
||||
ax.set_xticks([])
|
||||
ax.set_yticks([])
|
||||
self._figure.tight_layout()
|
||||
self._canvas.draw_idle()
|
||||
@@ -67,7 +67,7 @@ class SampleCameraThread(QThread):
|
||||
self.__fps_window_start = now
|
||||
self.__fps_frame_count = 0
|
||||
|
||||
if len(r) != 2:
|
||||
if len(r) < 2:
|
||||
continue
|
||||
|
||||
data = r[-1]
|
||||
@@ -82,31 +82,25 @@ class SampleCameraThread(QThread):
|
||||
except:
|
||||
continue
|
||||
|
||||
#meta, data = r
|
||||
#header = json.loads(meta)
|
||||
#header_shape = header["shape"]
|
||||
if header and header.get("type") == "uint8":# and len(header_shape) == 2:
|
||||
header_shape = header["shape"]
|
||||
bayer_image = np.frombuffer(data, dtype=np.uint8)
|
||||
bayer_image = bayer_image.reshape(header_shape)
|
||||
rgb_image = cv2.cvtColor(bayer_image, cv2.COLOR_BAYER_GB2RGB)
|
||||
#cv2.COLOR_BAYER_GB2RGB for ethernet connection
|
||||
rgb_image = rgb_image[:, ::-1, :].copy()
|
||||
if header and header.get("type") == "uint8":
|
||||
h, w = header["shape"][:2]
|
||||
raw = np.frombuffer(data, np.uint8).reshape((h, w))
|
||||
|
||||
rgb = cv2.cvtColor(raw, cv2.COLOR_BAYER_GB2RGB) # choose correct Bayer order
|
||||
#rgb = cv2.flip(rgb, 0)
|
||||
|
||||
if self.__measure_focus:
|
||||
gray = cv2.cvtColor(rgb_image, cv2.COLOR_RGB2GRAY)
|
||||
gray = cv2.cvtColor(rgb, cv2.COLOR_RGB2GRAY)
|
||||
if self.__focus_mask is None or self.__focus_mask.shape != gray.shape:
|
||||
height, width = gray.shape
|
||||
y, x = np.ogrid[:height, :width]
|
||||
self.__focus_mask = (x - self.__beam_x) ** 2 + (y - self.__beam_y) ** 2 <= self.__radius ** 2
|
||||
self.__focus_mask = (x - self.__beam_x) ** 2 + (
|
||||
y - self.__beam_y) ** 2 <= self.__radius ** 2
|
||||
|
||||
sharpness = focus_measure_edges(gray, self.__focus_mask)
|
||||
# print(f"GUI: mask pixels: {self.__focus_mask.sum()}, "
|
||||
# f"center=({self.__beam_x:.1f}, {self.__beam_y:.1f}), "
|
||||
# f"radius={self.__radius}, "
|
||||
# f"focus={sharpness:.2f}")
|
||||
self.focus_measure.emit(sharpness)
|
||||
|
||||
qimage = QImage(rgb_image.data, header_shape[1], header_shape[0], QImage.Format.Format_RGB888)
|
||||
qimage = QImage(rgb.data, rgb.shape[1], rgb.shape[0], QImage.Format.Format_RGB888).copy()
|
||||
self.camera_image.emit(QPixmap.fromImage(qimage))
|
||||
else:
|
||||
print("Sample camera image has wrong dimensions")
|
||||
|
||||
@@ -8,6 +8,9 @@ from PySide6.QtGui import QImage, QPixmap
|
||||
# If you need Bayer conversion like your SampleCameraThread did:
|
||||
import cv2
|
||||
|
||||
from aare.common.logger_config import setup_logger
|
||||
|
||||
logger = setup_logger("aareGUI")
|
||||
|
||||
class PredictionSubscriber(QThread):
|
||||
# emits parsed JSON payload (dict with keys: time, frame_id, shape, boxes)
|
||||
@@ -20,12 +23,16 @@ class PredictionSubscriber(QThread):
|
||||
super().__init__(parent)
|
||||
self._ctx = zmq.Context()
|
||||
self._sock = self._ctx.socket(zmq.SUB)
|
||||
self._sock.setsockopt(zmq.RCVTIMEO, 500)
|
||||
self._sock.setsockopt(zmq.LINGER, 0)
|
||||
|
||||
if isinstance(topic, str):
|
||||
self._sock.setsockopt_string(zmq.SUBSCRIBE, topic)
|
||||
elif isinstance(topic, bytes):
|
||||
self._sock.setsockopt(zmq.SUBSCRIBE, topic)
|
||||
else:
|
||||
self._sock.setsockopt(zmq.SUBSCRIBE, b"")
|
||||
|
||||
self._sock.connect(pred_zmq_url)
|
||||
self.running = True
|
||||
|
||||
@@ -65,23 +72,23 @@ class PredictionSubscriber(QThread):
|
||||
else:
|
||||
return None
|
||||
|
||||
# If you still need the horizontal flip you had before:
|
||||
rgb = rgb[:, ::-1, :].copy()
|
||||
|
||||
qimage = QImage(rgb.data, rgb.shape[1], rgb.shape[0], QImage.Format.Format_RGB888)
|
||||
qimage = QImage(rgb.data, rgb.shape[1], rgb.shape[0], QImage.Format.Format_RGB888).copy()
|
||||
return QPixmap.fromImage(qimage)
|
||||
|
||||
def run(self):
|
||||
while self.running:
|
||||
try:
|
||||
parts = self._sock.recv_multipart()
|
||||
try:
|
||||
while self.running:
|
||||
try:
|
||||
parts = self._sock.recv_multipart()
|
||||
except zmq.Again:
|
||||
continue
|
||||
|
||||
if not parts:
|
||||
continue
|
||||
|
||||
json_dicts: list[dict] = []
|
||||
non_json_parts: list[bytes] = []
|
||||
|
||||
# Parse all parts; keep non-JSON as candidates for image bytes
|
||||
for p in parts:
|
||||
d = self._try_parse_json(p)
|
||||
if d is not None:
|
||||
@@ -89,26 +96,51 @@ class PredictionSubscriber(QThread):
|
||||
else:
|
||||
non_json_parts.append(p)
|
||||
|
||||
header = next((d for d in json_dicts if "shape" in d and d.get("type") == "uint8"), None)
|
||||
header = next(
|
||||
(d for d in json_dicts if "shape" in d and d.get("type") == "uint8"),
|
||||
None,
|
||||
)
|
||||
detections = next((d for d in json_dicts if "boxes" in d), None)
|
||||
|
||||
# Heuristic: image payload is usually the largest non-JSON part
|
||||
image_bytes = max(non_json_parts, key=len) if non_json_parts else None
|
||||
|
||||
if header and image_bytes:
|
||||
pix = self._decode_image(header, image_bytes)
|
||||
if pix is not None:
|
||||
if pix is not None and self.running:
|
||||
self.image.emit(pix)
|
||||
|
||||
if detections:
|
||||
if detections and self.running:
|
||||
self.prediction.emit(detections)
|
||||
|
||||
except Exception as e:
|
||||
logger.error("PredictionSubscriber error:", e)
|
||||
break
|
||||
except Exception as e:
|
||||
if self.running:
|
||||
logger.exception(f"PredictionSubscriber error: {e}")
|
||||
finally:
|
||||
try:
|
||||
if self._sock is not None:
|
||||
self._sock.close(0)
|
||||
except Exception:
|
||||
pass
|
||||
finally:
|
||||
self._sock = None
|
||||
|
||||
try:
|
||||
if self._ctx is not None:
|
||||
self._ctx.term()
|
||||
except Exception:
|
||||
pass
|
||||
finally:
|
||||
self._ctx = None
|
||||
|
||||
def stop(self):
|
||||
self.running = False
|
||||
self._sock.close()
|
||||
self.quit()
|
||||
self.wait()
|
||||
self.requestInterruption()
|
||||
|
||||
try:
|
||||
if self._sock is not None:
|
||||
self._sock.close(0)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if not self.wait(1500):
|
||||
logger.warning("PredictionSubscriber did not stop within timeout")
|
||||
@@ -0,0 +1,156 @@
|
||||
from PySide6.QtCore import Qt
|
||||
from PySide6.QtWidgets import (
|
||||
QDialog,
|
||||
QVBoxLayout,
|
||||
QTextEdit,
|
||||
QDialogButtonBox,
|
||||
QTabWidget,
|
||||
QWidget,
|
||||
)
|
||||
|
||||
|
||||
class ControlsHelpDialog(QDialog):
|
||||
def __init__(self, parent=None):
|
||||
super().__init__(parent)
|
||||
|
||||
self.setWindowTitle("Mouse / Keyboard Controls")
|
||||
self.setMinimumSize(760, 560)
|
||||
|
||||
layout = QVBoxLayout(self)
|
||||
|
||||
self._tabs = QTabWidget(self)
|
||||
self._tabs.addTab(
|
||||
self._create_tab(
|
||||
"""
|
||||
<h2>Sample Camera</h2>
|
||||
|
||||
<h3>Mouse Wheel</h3>
|
||||
<ul>
|
||||
<li><b>Mouse wheel</b>: rotate omega by 90°</li>
|
||||
<li><b>Shift + Mouse wheel</b>: rotate omega by 10°</li>
|
||||
<li><b>Ctrl + Mouse wheel</b>: change sample camera exposure by a large step</li>
|
||||
<li><b>Alt + Mouse wheel</b>: change sample camera exposure by a small step</li>
|
||||
</ul>
|
||||
|
||||
<h3>Mouse Click / Drag</h3>
|
||||
<ul>
|
||||
<li><b>Left click</b>: move sample to clicked position</li>
|
||||
<li><b>Shift + Left click</b>: move using the special Z-alignment click behaviour</li>
|
||||
<li><b>Left drag on active raster grid</b>: move active raster grid</li>
|
||||
<li><b>Right click</b>: open sample camera context menu</li>
|
||||
<li><b>Right drag on empty area</b>: draw raster grid</li>
|
||||
<li><b>Right drag on active raster grid</b>: resize raster grid</li>
|
||||
</ul>
|
||||
|
||||
<h3>Mouse Move</h3>
|
||||
<ul>
|
||||
<li><b>Shift + Mouse move</b>: inspect/load raster image under cursor</li>
|
||||
<li><b>Mouse move</b>: show completed-grid tooltip or coordinates, depending on mode</li>
|
||||
</ul>
|
||||
|
||||
<h3>Beam Mark Mode</h3>
|
||||
<ul>
|
||||
<li><b>Shift + Left click</b>: set beam mark at clicked position</li>
|
||||
<li><b>Mouse wheel</b>: change exposure by a large step</li>
|
||||
<li><b>Alt + Mouse wheel</b>: change exposure by a small step</li>
|
||||
</ul>
|
||||
"""
|
||||
),
|
||||
"Sample Camera",
|
||||
)
|
||||
self._tabs.addTab(
|
||||
self._create_tab(
|
||||
"""
|
||||
<h2>Sample Camera Context Menu</h2>
|
||||
<ul>
|
||||
<li><b>Scale to fit</b></li>
|
||||
<li><b>Show coordinates</b></li>
|
||||
<li><b>Grab</b></li>
|
||||
<li><b>Grab with overlay</b></li>
|
||||
<li><b>Auto-focus</b></li>
|
||||
<li><b>Mark beam center</b></li>
|
||||
<li><b>Delete grid</b> <i>(when on active grid)</i></li>
|
||||
<li><b>Evaluate grid</b> <i>(when on active grid)</i></li>
|
||||
<li><b>Delete completed grids</b></li>
|
||||
</ul>
|
||||
|
||||
<h3>Notes</h3>
|
||||
<ul>
|
||||
<li>Some actions appear only when the cursor is over the active raster grid.</li>
|
||||
<li>Grid actions depend on the current sample camera state and current raster visibility.</li>
|
||||
</ul>
|
||||
"""
|
||||
),
|
||||
"Context Menu",
|
||||
)
|
||||
self._tabs.addTab(
|
||||
self._create_tab(
|
||||
"""
|
||||
<h2>Video Views</h2>
|
||||
<p>Applies to gonio camera, beamline view, and combined beamline views.</p>
|
||||
|
||||
<h3>Mouse</h3>
|
||||
<ul>
|
||||
<li><b>Ctrl + Mouse wheel</b>: zoom in/out</li>
|
||||
<li><b>Mouse wheel</b>: normal scrolling when Ctrl is not pressed</li>
|
||||
<li><b>Mouse drag</b>: rubber-band drag/selection behaviour is enabled</li>
|
||||
</ul>
|
||||
|
||||
<h3>Keyboard</h3>
|
||||
<ul>
|
||||
<li><b>F</b>: fit to view</li>
|
||||
<li><b>R</b>: reset zoom</li>
|
||||
<li><b>+</b> or <b>=</b>: zoom in</li>
|
||||
<li><b>-</b>: zoom out</li>
|
||||
</ul>
|
||||
"""
|
||||
),
|
||||
"Video Views",
|
||||
)
|
||||
self._tabs.addTab(
|
||||
self._create_tab(
|
||||
"""
|
||||
<h2>Other Interactive Areas</h2>
|
||||
|
||||
<h3>Sample Queue</h3>
|
||||
<ul>
|
||||
<li><b>Delete</b>: remove selected samples from the queue</li>
|
||||
</ul>
|
||||
|
||||
<h3>Fluorescence Plot</h3>
|
||||
<ul>
|
||||
<li><b>Mouse move</b>: show energy/count tooltip and vertical marker line</li>
|
||||
<li><b>Right click</b>: save spectrum as CSV</li>
|
||||
</ul>
|
||||
|
||||
<h3>General Widgets</h3>
|
||||
<ul>
|
||||
<li><b>Left click</b> on clickable labels/value labels: trigger the widget's click action</li>
|
||||
</ul>
|
||||
|
||||
<h3>Notes</h3>
|
||||
<ul>
|
||||
<li>Some wheel interactions are intentionally disabled in certain scroll areas to prevent accidental scrolling.</li>
|
||||
<li>Available actions can depend on beamline state, current mode, and widget focus.</li>
|
||||
</ul>
|
||||
"""
|
||||
),
|
||||
"Other Panels",
|
||||
)
|
||||
layout.addWidget(self._tabs)
|
||||
|
||||
buttons = QDialogButtonBox(QDialogButtonBox.StandardButton.Close, parent=self)
|
||||
buttons.rejected.connect(self.reject)
|
||||
buttons.accepted.connect(self.accept)
|
||||
layout.addWidget(buttons)
|
||||
|
||||
def _create_tab(self, html: str) -> QWidget:
|
||||
text = QTextEdit(self)
|
||||
text.setReadOnly(True)
|
||||
text.setLineWrapMode(QTextEdit.LineWrapMode.WidgetWidth)
|
||||
text.setTextInteractionFlags(
|
||||
Qt.TextInteractionFlag.TextSelectableByMouse
|
||||
| Qt.TextInteractionFlag.TextSelectableByKeyboard
|
||||
)
|
||||
text.setHtml(html)
|
||||
return text
|
||||
@@ -10,7 +10,7 @@ from PySide6.QtGui import (
|
||||
QWheelEvent,
|
||||
QTransform,
|
||||
QCursor,
|
||||
QLinearGradient, QFont, QFontMetrics,
|
||||
QLinearGradient, QFont, QFontMetrics, QPolygonF,
|
||||
)
|
||||
from PySide6.QtWidgets import (
|
||||
QMenu,
|
||||
@@ -526,6 +526,7 @@ class SampleCameraImageLabel(QGraphicsView):
|
||||
y1 = det['y1'] * sy
|
||||
x2 = det['x2'] * sx
|
||||
y2 = det['y2'] * sy
|
||||
poly = det.get('poly', None)
|
||||
label = str(det.get('label', '')).lower()
|
||||
conf = det.get('conf', 0.0)
|
||||
except Exception as e:
|
||||
@@ -540,6 +541,12 @@ class SampleCameraImageLabel(QGraphicsView):
|
||||
painter.setBrush(Qt.BrushStyle.NoBrush)
|
||||
detection_rect = QRect(int(x1), int(y1), int(max(1, x2 - x1)), int(max(1, y2 - y1)))
|
||||
painter.drawRect(detection_rect)
|
||||
if poly and len(poly) >= 3:
|
||||
polygon = QPolygonF([
|
||||
QPointF(x1 + float(p[0]) * sx, y1 + float(p[1]) * sy)
|
||||
for p in poly
|
||||
])
|
||||
painter.drawPolygon(polygon)
|
||||
|
||||
# Draw label text with white text on colored background
|
||||
painter.setPen(QPen(QColor(255, 255, 255), 1))
|
||||
|
||||
Reference in New Issue
Block a user