mx/AareDAQ
0
1
Fork 0
Code Issues 49 Pull Requests Actions Packages Projects Releases 84 Wiki Activity
Files
fbbb23684485d3d7fe9d0a92f2b7ca3fa96f2b86
AareDAQ/src/aare/gui/main_window.py
T
duan_jandClaude Fable 5 fbbb236844 feat: cross-fade and palette flip on theme switch 
Freeze the old look in a click-through screenshot overlay and fade it
out over the restyled window — QSS has no transitions and animating the
palette would re-polish every widget per frame. Sunset also flips the
app-palette text roles so native primitives QSS can't recolor (spin and
combo arrow glyphs) follow the theme.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-08-08 15:07:23 +02:00

2954 lines
135 KiB
Python
Raw Blame History

import time
import jwt
from aarecommon.config.logger import setup_logger
from aarecommon.math.coordinate import Coordinate, SmargonCoordinate
from aarecommon.math.diffraction_geometry import DiffractionGeometry
from aarecommon.math.sample_geometry import SampleGeometryModel
# Common imports
from aarecommon.models.auth import BatonStatus
from aarecommon.models.models import (
BeamlineStateEnum,
DAQStatusModel,
SampleShortInfoList,
SessionsStateEnum,
TokenData,
)
from PySide6.QtCore import QEvent, QPropertyAnimation, QSettings, Qt, QTimer, Signal, Slot
from PySide6.QtGui import (
QAction,
QActionGroup,
QColor,
QCursor,
QGuiApplication,
QKeySequence,
QPalette,
)
from PySide6.QtWidgets import (
QApplication,
QCheckBox,
QDockWidget,
QFrame,
QGraphicsColorizeEffect,
QGraphicsOpacityEffect,
QHBoxLayout,
QLabel,
QMainWindow,
QMessageBox,
QPushButton,
QScrollArea,
QSizePolicy,
QStackedWidget,
QTabWidget,
QToolBar,
QVBoxLayout,
QWidget,
)
from aare.gui.about import about_text
from aare.gui.constants import LOGGER_NAME
# Gui Models
from aare.gui.models.gui_state_manager import UIStateManager
from aare.gui.panels.abr_tweak_panel import AbrTweakWidget
from aare.gui.panels.automation_panel import AutomationProgressWidget
from aare.gui.panels.axis_video_panel import AxisVideoPanel
from aare.gui.panels.beamline_controls import BeamConfigPanel, BeamlineControls
from aare.gui.panels.beamline_recovery_panel import BeamlineRecoveryDialog
from aare.gui.panels.beamline_state_panel import BeamlineStatePanel
from aare.gui.panels.compact_automation_panel import CompactAutomationPanel
from aare.gui.panels.data_collection_settings import DataCollectionSettings
from aare.gui.panels.developer_help_dialog import DeveloperHelpDialog
from aare.gui.panels.face_detection_panel import FaceDetectionPanel
from aare.gui.panels.fluorescence_panel import FluorescencePanel
from aare.gui.panels.local_contact_panel import LocalContactDialog
# panels
from aare.gui.panels.log_panel import LogDock
from aare.gui.panels.monochromator_panel import MonochromatorPanel
from aare.gui.panels.portrait_mode import PortraitModePanel
from aare.gui.panels.prediction_metrics_panel import PredictionMetricsPanel
from aare.gui.panels.reference_tools_panel import ReferenceToolsPanel
from aare.gui.panels.samcam_panel import SamcamPanel
from aare.gui.panels.sample_queue_panel import SampleQueuePanel
from aare.gui.panels.smargon_trace_panel import SmargonTracePanel
from aare.gui.panels.target_stability_panel import TargetStabilityPanel
from aare.gui.panels.tell_sample_panel import TellSamplePanel
# Scan Logic
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.styles import (
APP_BACKGROUND,
DARK_TEXT,
DOCK_CONTENT_LEFT_PAD,
THEME_FADE_MS,
THEME_ORIGINAL,
THEME_PORTRAIT,
build_app_stylesheet,
qcolor,
)
# Threads
from aare.gui.threads.axis_video_thread import VideoThread
from aare.gui.threads.daq_worker import DAQWorker
from aare.gui.threads.jfjoch_viewer import JFJochDBusClient
from aare.gui.threads.prediction_subscriber import PredictionSubscriber
from aare.gui.tutorials.controls_help_dialog import ControlsHelpDialog
# Tutorials
from aare.gui.tutorials.tutorial_actions import TutorialActionExecutor
from aare.gui.tutorials.tutorial_manager import TutorialManager
from aare.gui.tutorials.tutorial_registration import register_tutorials
from aare.gui.tutorials.tutorial_runtime import DictionaryTextResolver, TutorialEventBus
from aare.gui.tutorials.tutorial_targets import MainWindowTutorialTargetResolver
from aare.gui.tutorials.tutroial_texts import MANUAL_MOUNT_TUTORIAL
# Widgets
from aare.gui.widgets.alert_banner import AlertBanner
from aare.gui.widgets.baton_request_dialog import BatonPendingDialog, BatonRequestDialog
from aare.gui.widgets.busy_overlay import build_busy_overlay_style
from aare.gui.widgets.camera_image import SampleCameraImageLabel
from aare.gui.widgets.message_box import precondition_check
from aare.gui.widgets.no_wheel_scroll_area import NoWheelScrollArea
from aare.gui.widgets.popout_window import DockTitleBar, PopoutWindow
from aare.gui.widgets.status_bar import StatusBar
from aare.gui.widgets.title_label import TitleLabel, tighten_column
from aare.gui.widgets.video_image import VideoGraphicsView
from aare.gui.widgets.wheel_value_guard import WheelValueGuard
logger = setup_logger(LOGGER_NAME)
class MainWindow(QMainWindow):
sample_geometry = Signal(SampleGeometryModel)
# Set lazily outside __init__ (first use guards with getattr/default);
# declared for the basedpyright gate.
_session_operations_enabled: bool | None = None
_default_dock_split_done: bool = False
def __init__(
self,
base_url: str | None,
token: str,
default_image: str | None,
zmq_addr: str | None,
pred_zmq_addr: str | None,
beamline_cam_addr: str | None,
gonio_cam_addr: str | None,
gonio_cam_id: int | None,
):
super().__init__()
# Banners open before a baton-vacancy fold; in __init__ (not a class
# default) so the mutable list is per-instance (RUF012).
self._pre_vacancy_open_banners: list[TitleLabel] = []
self._theme_mode = THEME_ORIGINAL
self._theme_action_group = None
self._use_legacy_theme_action = None
self._use_portrait_theme_action = None
self._base_url = base_url
self._token = token
self._mounting = False
self._samcam_feed_banner_active = False
self._samcam_feed_banner_message = "Sample camera feed unavailable"
self._automation_critical_banner_active = False
self._dev_help_dialog = None
self._beamline_recovery_dialog = None
self._local_contact_dialog = None
self._controls_help_dialog = None
self._cleanup_done = False
self._default_window_state = None
self._pre_automation_window_state = None
self._pre_automation_left_column_visible = True
self._pre_automation_right_column_visible = True
self._in_compact_automation_view = False
self._enter_automation_view_action = None
self._return_main_view_action = None
self._beamline_cam_addr = beamline_cam_addr
self._gonio_cam_addr = gonio_cam_addr
self._gonio_cam_id = gonio_cam_id
self._axis_camera_refresh_interval_ms = 60 * 60 * 1000
self.beamline_camera_thread = None
self.gonio_camera_thread = None
self._waiting_for_baton_response: bool = False
self._baton_request_dialog: BatonRequestDialog | None = None
self._baton_pending_dialog: BatonPendingDialog | None = None
self._last_user_interaction_ts = time.time()
self._last_interaction_report_ts = 0.0
self._interaction_report_min_interval_s = 15.0
self._idle_close_timeout_s = 60 * 60 * 8
self._remote_close_deadline_ts: float | None = None
self._remote_close_reason: str | None = None
self._remote_close_banner_active: bool = False
self._latest_daq_status: DAQStatusModel | None = None
self._tutorial_event_bus = TutorialEventBus(self)
self._tutorial_text_resolver = DictionaryTextResolver(MANUAL_MOUNT_TUTORIAL)
self.state_manager = UIStateManager("PSI", "AareGUI")
# Wheel safety: sliders/spin boxes/combos only react to the wheel
# while the right mouse button is held; a bare wheel just scrolls
# the page — it can never nudge a value or move a motor.
self._wheel_value_guard = WheelValueGuard(self)
app_instance = QApplication.instance()
if app_instance is not None:
app_instance.installEventFilter(self._wheel_value_guard)
self.viewer = JFJochDBusClient()
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:
logger.exception("Failed to decode authentication token")
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(f"background-color: {APP_BACKGROUND};")
root_widget = QWidget(parent=self)
root_widget.setObjectName("mainContentRoot")
root_layout = QVBoxLayout(root_widget)
root_layout.setContentsMargins(0, 0, 0, 0)
root_layout.setSpacing(0)
self.alert_banner = AlertBanner(parent=root_widget)
root_layout.addWidget(self.alert_banner)
self.alert_banner_secondary = AlertBanner(parent=root_widget)
root_layout.addWidget(self.alert_banner_secondary)
self.content_stack = QStackedWidget(parent=root_widget)
root_layout.addWidget(self.content_stack, 1)
top_widget = QWidget(parent=root_widget)
top_widget.setObjectName("standardMainPage")
top_widget_layout = QHBoxLayout(top_widget)
top_widget.setLayout(top_widget_layout)
diffraction = DiffractionGeometry(
energy_keV=12.4,
dtz_mm=100,
detector_size_pxl=(1553, 1630),
pixel_size_mm=0.150, # PILATUS 4
beam_center_pxl=(750, 750),
detector_description="PILATUS 4",
detector_serial_number="1",
poni_rot1_rad=-0.001396263,
poni_rot2_rad=-0.003839724,
)
geom = SampleGeometryModel(
beam_location_pxl=Coordinate(x=1000, y=1000),
pixel_in_mm=0.001,
aerotech=Coordinate(),
smargon=SmargonCoordinate(sh_mm=Coordinate(), phi_deg=0, chi_deg=0),
omega_deg=0,
beam_size_mm=Coordinate(x=0.01, y=0.01),
aerotech_meas=Coordinate(),
)
self.raster = RasterGridManager(geom=geom)
self.rotation = RotationScanManager()
self.collection_controls_scroll = NoWheelScrollArea(top_widget)
self.left_column = QWidget(parent=top_widget)
self.left_column_layout = QVBoxLayout(self.left_column)
self.left_column_layout.setContentsMargins(0, 0, 0, 0)
self.data_collection = DataCollectionSettings(
s=geom, parent=self.left_column, raster_mgr=self.raster, diffraction=diffraction
)
# The beamline state strip lives in a bottom toolbar row (created
# after the docks), not in the left column. Always visible.
self.beamline_state_panel = BeamlineStatePanel(parent=self)
# Beamline / Experiment as tabs (like the Dewar samples dock) instead
# of two stacked banner groups; the pages keep their banner children.
# documentMode: no pane frame, so the fixed-width panels aren't inset.
self.left_column_tabs = QTabWidget(self.left_column)
self.left_column_tabs.setDocumentMode(True)
# documentMode draws a grey base line across the bar's full width.
self.left_column_tabs.tabBar().setDrawBase(False)
beamline_page = QWidget()
beamline_layout = QVBoxLayout(beamline_page)
beamline_layout.setContentsMargins(0, 0, 0, 0)
self.samcam = SamcamPanel(beamline_page)
if self._decoded_token.staff:
self.monochromator_panel = MonochromatorPanel(beamline_page)
self.abr_tweak = AbrTweakWidget(beamline_page)
self.beam_config = BeamConfigPanel(beamline_page)
self.beam_mark = self.beam_config.beam_mark
self.beam_center = self.beam_config.beam_center
self.beam_size = self.beam_config.beam_size
beamline_layout.addWidget(self.monochromator_panel)
beamline_layout.addWidget(self.abr_tweak)
beamline_layout.addWidget(self.beam_config)
# Samcam last: the beam panels are the ones tweaked most.
beamline_layout.addWidget(self.samcam)
beamline_layout.addStretch()
tighten_column(beamline_layout)
experiment_page = QWidget()
experiment_layout = QVBoxLayout(experiment_page)
experiment_layout.setContentsMargins(0, 0, 0, 0)
experiment_layout.addWidget(self.data_collection)
experiment_layout.addStretch()
tighten_column(experiment_layout)
self.left_column_tabs.addTab(beamline_page, "Beamline")
self.left_column_tabs.addTab(experiment_page, "Experiment")
# Only the visible page counts toward the height — same trick as the
# content stack below, else the taller page pads the other tab.
def _only_current_left_tab_counts(index: int) -> None:
for i in range(self.left_column_tabs.count()):
page = self.left_column_tabs.widget(i)
vertical = (
QSizePolicy.Policy.Preferred if i == index else QSizePolicy.Policy.Ignored
)
page.setSizePolicy(QSizePolicy.Policy.Preferred, vertical)
self.left_column_tabs.currentChanged.connect(_only_current_left_tab_counts)
_only_current_left_tab_counts(self.left_column_tabs.currentIndex())
# Dewar-tabs look: first banner flush under the tab bar (no top
# margin) and the tab row starting at the banners' left edge. The
# Experiment side needs two levels: the frame AND its first panel.
beamline_first = self.monochromator_panel if self._decoded_token.staff else self.samcam
for first in (beamline_first, self.data_collection, self.data_collection.file_path_panel):
first_layout = first.layout()
assert first_layout is not None # panels build their layouts in __init__
m = first_layout.contentsMargins()
first_layout.setContentsMargins(m.left(), 0, m.right(), m.bottom())
samcam_layout = self.samcam.layout()
assert samcam_layout is not None
self.left_column_tabs.setStyleSheet(
f"QTabWidget::tab-bar {{ left: {samcam_layout.contentsMargins().left()}px; }}"
)
self.left_column_layout.addWidget(self.left_column_tabs)
self.left_column_layout.addStretch()
top_widget_layout.addWidget(self.collection_controls_scroll)
self.collection_controls_scroll.setWidget(self.left_column)
self.collection_controls_scroll.setHorizontalScrollBarPolicy(
Qt.ScrollBarPolicy.ScrollBarAlwaysOff
)
self.collection_controls_scroll.setWidgetResizable(True)
# No frame: its border drew a line above the tab bar (Dewar tabs have
# none). Freeze the inner column width: widgetResizable makes it track
# the viewport, so the scrollbar appearing used to re-flow every
# banner. Fixed width + a permanent 10px scrollbar gutter means the
# scrollbar pops into spare space and nothing moves.
self.collection_controls_scroll.setFrameShape(QFrame.Shape.NoFrame)
self.left_column.setFixedWidth(self.data_collection.set_width)
self.collection_controls_scroll.setFixedWidth(self.data_collection.set_width + 10)
self.video_tab = QTabWidget(parent=top_widget)
self.sample_camera = SampleCameraImageLabel(
geom=geom, raster=self.raster, parent=top_widget, default_image=default_image
)
self.beamline_view = VideoGraphicsView()
self.beamline_view_panel = AxisVideoPanel(
"Beamline view", self.beamline_view, parent=top_widget
)
self.beamline_view_panel.refresh_requested.connect(self.refresh_axis_cameras)
self.gonio_view = VideoGraphicsView()
self.gonio_view_panel = AxisVideoPanel("Gonio camera", self.gonio_view, parent=top_widget)
self.gonio_view_panel.refresh_requested.connect(self.refresh_axis_cameras)
self.beamline_view_container = QWidget(parent=top_widget)
self.beamline_view_layout = QVBoxLayout(self.beamline_view_container)
self.beamline_view_layout.setContentsMargins(0, 0, 0, 0)
self.beamline_view_layout.setSpacing(6)
self.beamline_view_1_combined = VideoGraphicsView()
self.beamline_view_2_combined = VideoGraphicsView()
self.beamline_view_layout.addWidget(self.beamline_view_1_combined)
self.beamline_view_layout.addWidget(self.beamline_view_2_combined)
self.beamline_combined_panel = AxisVideoPanel(
"Beamline combined view", self.beamline_view_container, parent=top_widget
)
self.beamline_combined_panel.refresh_requested.connect(self.refresh_axis_cameras)
self.video_tab.addTab(self.sample_camera, "Sample camera")
self.video_tab.addTab(self.gonio_view_panel, "Gonio camera")
self.video_tab.addTab(self.beamline_view_panel, "Beamline view")
self.video_tab.addTab(self.beamline_combined_panel, "Beamline combined view")
# if cfg_get("gui.cameras.secondary_beamline_camera_url", None):
# self.secondary_beamline_view = VideoGraphicsView()
# self.secondary_beamline_view_panel = AxisVideoPanel("Secondary view", self.secondary_beamline_view,
# parent=top_widget)
# self.secondary_beamline_view_panel.refresh_requested.connect(self.refresh_axis_cameras)
# self.video_tab.addTab(self.secondary_beamline_view_panel, "Secondary view")
self.compact_sample_camera = SampleCameraImageLabel(
geom=geom, raster=self.raster, parent=root_widget, default_image=default_image
)
self.compact_automation_panel = CompactAutomationPanel(
self.compact_sample_camera, parent=root_widget
)
self.compact_automation_page = QWidget(parent=root_widget)
self.compact_automation_page.setObjectName("compactAutomationPage")
self.compact_automation_page_layout = QVBoxLayout(self.compact_automation_page)
self.compact_automation_page_layout.setContentsMargins(18, 18, 18, 18)
self.compact_automation_page_layout.setSpacing(0)
self.compact_automation_page_layout.addWidget(self.compact_automation_panel)
# ── Portrait mode page ──────────────────────────────────────────
self.portrait_sample_camera = SampleCameraImageLabel(
geom=geom, raster=self.raster, parent=root_widget, default_image=default_image
)
self.portrait_mode_panel = PortraitModePanel(
sample_camera_widget=self.portrait_sample_camera, parent=root_widget
)
self.portrait_mode_page = QWidget(parent=root_widget)
self.portrait_mode_page.setObjectName("portraitModePage")
portrait_page_layout = QHBoxLayout(self.portrait_mode_page)
portrait_page_layout.setContentsMargins(0, 0, 0, 0)
portrait_page_layout.setSpacing(0)
self.portrait_mode_page.setFixedWidth(self.portrait_mode_panel.PORTRAIT_WIDTH + 24)
portrait_page_layout.addWidget(
self.portrait_mode_panel, alignment=Qt.AlignmentFlag.AlignHCenter
)
top_widget_layout.addWidget(self.video_tab)
self._start_axis_camera_threads()
self.beamline_controls_scroll = NoWheelScrollArea(top_widget)
self.beamline = BeamlineControls(self.beamline_controls_scroll)
top_widget_layout.addWidget(self.beamline_controls_scroll)
self.beamline_controls_scroll.setWidget(self.beamline)
# Resizable so the column shrinks when panels collapse; without it the
# scrollbar keeps dead range below the collapsed panels.
self.beamline_controls_scroll.setWidgetResizable(True)
self.beamline_controls_scroll.setHorizontalScrollBarPolicy(
Qt.ScrollBarPolicy.ScrollBarAlwaysOff
)
# Same gutter math as the left column: 10px scrollbar + 2px frame, so
# the fixed-width controls are never clipped when the scrollbar shows.
self.beamline_controls_scroll.setFixedWidth(self.beamline.set_width + 12)
self.tell_samples = TellSamplePanel(samples=SampleShortInfoList(s=[]))
self.ref_tools_panel = ReferenceToolsPanel(samples=SampleShortInfoList(s=[]))
self.job_list_panel = SampleQueuePanel(show_user=self._decoded_token.staff)
self.compact_automation_panel.play_pause_clicked.connect(self.job_list_panel.run)
self.compact_automation_panel.skip_clicked.connect(self.job_list_panel.skip_current_sample)
self.compact_automation_panel.step_through_toggled.connect(
self.job_list_panel.set_step_through
)
self.compact_automation_panel.show_full_view_requested.connect(
self._return_from_compact_automation_view
)
self.compact_automation_panel.annotation_selected.connect(self._handle_compact_annotation)
# One dock for both lists: the old tabified Reference Tools dock was
# staff-only and hid behind the Sample List tab, so it "sometimes"
# showed. aaregui2 concept: one panel, Dewar + Auxiliary-puck tabs
# that always travel together, and the dewar table doubles as the
# queue view (status tints), so the automation controls sit under it
# and the Automation list dock is gone. SampleQueuePanel stays alive,
# hidden, as the queue engine; its buttons are reparented here so all
# their existing wiring keeps working.
dewar_tab = QWidget()
dewar_layout = QVBoxLayout(dewar_tab)
dewar_layout.setContentsMargins(0, 0, 0, 0)
dewar_layout.setSpacing(2)
dewar_layout.addWidget(self.tell_samples)
self.quick_unmount_button = QPushButton("⏏ Unmount", dewar_tab)
self.quick_unmount_button.clicked.connect(lambda: self._on_manual_unmount_requested())
automation_row = QHBoxLayout()
for w in (
self.job_list_panel.play_button,
self.job_list_panel.remove_button,
self.job_list_panel.clear_button,
self.quick_unmount_button,
self.job_list_panel.park_and_dry_when_cleared,
self.job_list_panel.pause_on_conditions_cb,
):
automation_row.addWidget(w)
dewar_layout.addLayout(automation_row)
# "Remove selected" now unqueues the dewar-table selection — the
# queue's own table is no longer displayed.
self.job_list_panel.remove_button.clicked.disconnect(
self.job_list_panel.remove_selected_samples
)
self.job_list_panel.remove_button.clicked.connect(self._remove_selected_from_queue)
self.job_list_panel.hide()
self.sample_lists_tabs = QTabWidget()
self.sample_lists_tabs.addTab(dewar_tab, "Dewar samples")
self.sample_lists_tabs.addTab(self.ref_tools_panel, "Auxiliary puck")
# Non-staff never get reference-tools data (the DAQ connect below is
# staff-gated). Grey the tab out instead of hiding it: every role sees
# the same view, and clicking the locked tab says why it is locked.
if not self._decoded_token.staff:
self.sample_lists_tabs.setTabEnabled(1, False)
self.sample_lists_tabs.setTabToolTip(1, "Staff only")
# Disabled tabs are skipped by tabBarClicked's hit-test, so the
# click is caught in eventFilter via the geometric tabAt() instead.
self.sample_lists_tabs.tabBar().installEventFilter(self)
# Wrapper for the left inset: QTabWidget ignores its own contents
# margins for the tab bar, so the padding lives one level up. Aligns
# the panel's left edge with the left column above (Loop centering).
sample_lists_wrap = QWidget()
sample_lists_wrap_layout = QVBoxLayout(sample_lists_wrap)
sample_lists_wrap_layout.setContentsMargins(DOCK_CONTENT_LEFT_PAD, 0, 0, 0)
sample_lists_wrap_layout.setSpacing(0)
sample_lists_wrap_layout.addWidget(self.sample_lists_tabs)
self.tell_samples_dock = QDockWidget("Sample List", self)
self.tell_samples_dock.setObjectName("tell_samples_dock")
self.tell_samples_dock.setWidget(sample_lists_wrap)
self.tell_samples_dock.setAllowedAreas(Qt.DockWidgetArea.BottomDockWidgetArea)
self.addDockWidget(Qt.DockWidgetArea.BottomDockWidgetArea, self.tell_samples_dock)
# No floating: popping the dock out ripped the panel from the row and
# reshuffled the rest. The ⤢ button (in the title bar, next to ✕)
# opens an ADDITIONAL fully-wired window; closing it changes nothing.
self.tell_samples_dock.setFeatures(
QDockWidget.DockWidgetFeature.DockWidgetMovable
| QDockWidget.DockWidgetFeature.DockWidgetClosable
)
self._sample_popout: PopoutWindow | None = None
self.tell_samples_dock.setTitleBarWidget(
DockTitleBar(self.tell_samples_dock, self._open_sample_popout)
)
self.sample_logic = SampleMountLogic()
# Manual sample lives in the left column (DataCollectionSettings)
# between Dataset path and Exp. Config., collapsible like its
# neighbors — it is no longer a bottom dock.
self.manual_sample_panel = self.data_collection.manual_sample_panel
self.automation_progress_panel = AutomationProgressWidget()
self.automation_progress_dock = QDockWidget("Automation progress", self)
self.automation_progress_dock.setObjectName("automation_progress_dock")
# Scroll host: the panel's ~420px minimum otherwise dictates the whole
# bottom row's height and squeezes the Beamline column into a scrollbar.
automation_scroll = NoWheelScrollArea(self.automation_progress_dock)
automation_scroll.setWidget(self.automation_progress_panel)
automation_scroll.setWidgetResizable(True)
automation_scroll.setFrameShape(QFrame.Shape.NoFrame)
self.automation_progress_dock.setWidget(automation_scroll)
self.automation_progress_dock.setAllowedAreas(Qt.DockWidgetArea.BottomDockWidgetArea)
self.addDockWidget(Qt.DockWidgetArea.BottomDockWidgetArea, self.automation_progress_dock)
# Same title-bar icons (⤢ pop-out + ✕) as Sample List / Console Log.
self._automation_popout: PopoutWindow | None = None
self.automation_progress_dock.setTitleBarWidget(
DockTitleBar(self.automation_progress_dock, self._open_automation_popout)
)
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)
self.face_panel_dock.hide()
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)
self.fluor_panel_dock.hide()
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("aareGUI")
self.log_dock.hide()
self.tabifyDockWidget(self.automation_progress_dock, self.log_dock)
self.job_list_panel.samples_in_queue_changed.connect(
self.automation_progress_panel.set_samples_in_queue
)
self.job_list_panel.automation_running_changed.connect(
self.automation_progress_panel.set_running
)
self.job_list_panel.automation_running_changed.connect(self._on_automation_running_changed)
self.job_list_panel.automation_running_changed.connect(
self.compact_automation_panel.set_running
)
self.job_list_panel.step_through_changed.connect(
self.compact_automation_panel.set_step_through
)
self.job_list_panel.samples_in_queue_changed.connect(
self.compact_automation_panel.set_samples_in_queue
)
self.job_list_panel.samples_in_queue_changed.connect(self._refresh_compact_queue_preview)
# Portrait mode: queue size + running state + preview
self.job_list_panel.automation_running_changed.connect(self.portrait_mode_panel.set_running)
self.job_list_panel.samples_in_queue_changed.connect(
self.portrait_mode_panel.set_samples_in_queue
)
self.job_list_panel.samples_in_queue_changed.connect(self._refresh_portrait_queue_preview)
self.automation_progress_panel.set_samples_in_queue(
len(self.job_list_panel.table_model.samples)
)
self.automation_progress_panel.set_running(self.job_list_panel.is_running())
self.compact_automation_panel.set_running(self.job_list_panel.is_running())
self.compact_automation_panel.set_step_through(self.job_list_panel.is_step_through())
self.compact_automation_panel.set_samples_in_queue(
len(self.job_list_panel.table_model.samples)
)
self._refresh_compact_queue_preview()
# smargon trace panel
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()
# Target stability panel
self.target_stability_panel = TargetStabilityPanel()
self.target_stability_dock = QDockWidget("Target stability", self)
self.target_stability_dock.setObjectName("target_stability_dock")
self.target_stability_dock.setWidget(self.target_stability_panel)
self.target_stability_dock.setAllowedAreas(
Qt.DockWidgetArea.RightDockWidgetArea
| Qt.DockWidgetArea.LeftDockWidgetArea
| Qt.DockWidgetArea.BottomDockWidgetArea
)
self.addDockWidget(Qt.DockWidgetArea.RightDockWidgetArea, self.target_stability_dock)
self.target_stability_dock.hide()
# Prediction Metrics Panel
self.prediction_metrics_panel = PredictionMetricsPanel()
self.prediction_metrics_dock = QDockWidget("Prediction Metrics", self)
self.prediction_metrics_dock.setObjectName("prediction_metrics_dock")
self.prediction_metrics_dock.setWidget(self.prediction_metrics_panel)
self.prediction_metrics_dock.setAllowedAreas(
Qt.DockWidgetArea.RightDockWidgetArea
| Qt.DockWidgetArea.LeftDockWidgetArea
| Qt.DockWidgetArea.BottomDockWidgetArea
)
self.addDockWidget(Qt.DockWidgetArea.RightDockWidgetArea, self.prediction_metrics_dock)
self.prediction_metrics_dock.hide()
# Full-width state strip directly above the status bar: the bottom
# toolbar area is guaranteed to sit below the bottom dock area (a
# bottom dock would land beside the existing docks instead).
self.beamline_state_toolbar = QToolBar("Beamline state", self)
self.beamline_state_toolbar.setObjectName("beamline_state_toolbar")
self.beamline_state_toolbar.setMovable(False)
self.beamline_state_toolbar.setFloatable(False)
self.beamline_state_toolbar.setAllowedAreas(Qt.ToolBarArea.BottomToolBarArea)
self.beamline_state_panel.setSizePolicy(
QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Preferred
)
self.beamline_state_toolbar.addWidget(self.beamline_state_panel)
self.addToolBar(Qt.ToolBarArea.BottomToolBarArea, self.beamline_state_toolbar)
# The standard page's combined panel minimum (~1400x1200) exceeds many
# monitors, which pushed the status bar off-screen. Scrolling instead of
# clamping lets the window shrink to any screen; scrollbars only appear
# when the monitor is actually smaller than the panels.
standard_page_scroll = QScrollArea(parent=root_widget)
standard_page_scroll.setWidgetResizable(True)
standard_page_scroll.setFrameShape(QFrame.Shape.NoFrame)
standard_page_scroll.setWidget(top_widget)
self.content_stack.addWidget(standard_page_scroll)
self.content_stack.addWidget(self.compact_automation_page)
self.content_stack.addWidget(self.portrait_mode_page)
self.content_stack.setCurrentWidget(standard_page_scroll)
self._standard_main_page = standard_page_scroll
# QStackedWidget's minimum size is the max over ALL pages, so the hidden
# portrait/compact pages inflated the window minimum past small monitors
# (portrait alone demands 720px height). Only the visible page should
# count — the mode-switch code resizes the window explicitly anyway.
def _only_current_page_counts(index: int) -> None:
for i in range(self.content_stack.count()):
page = self.content_stack.widget(i)
policy = QSizePolicy.Policy.Preferred if i == index else QSizePolicy.Policy.Ignored
page.setSizePolicy(policy, policy)
self.content_stack.currentChanged.connect(_only_current_page_counts)
_only_current_page_counts(self.content_stack.currentIndex())
self.setCentralWidget(root_widget)
self.setWindowTitle("AareGUI")
self._restore_theme_settings()
self._apply_theme()
self.create_menu_bar()
self._update_view_mode_actions()
self._setup_global_shortcuts()
# Default bottom-dock height: the sample list used to grab ~40% of the
# window and squeeze the Beamline column behind a scrollbar. Before
# the default-state capture so "reset layout" gets it too; a saved
# user layout (restored below) still wins.
self.resizeDocks(
[self.tell_samples_dock, self.log_dock], [240, 240], Qt.Orientation.Vertical
)
# Equal oversized requests -> Qt distributes proportionally = 50/50.
self.resizeDocks(
[self.tell_samples_dock, self.automation_progress_dock],
[10000, 10000],
Qt.Orientation.Horizontal,
)
self._capture_default_window_state()
self._restore_window_state()
# Define tutorials now that the UI exists
self._tutorial_target_resolver = MainWindowTutorialTargetResolver(self)
self._tutorial_action_executor = TutorialActionExecutor(self)
self.tutorial_manager = TutorialManager(
self,
target_resolver=self._tutorial_target_resolver,
text_resolver=self._tutorial_text_resolver,
action_executor=self._tutorial_action_executor,
event_bus=self._tutorial_event_bus,
)
self.status_bar = StatusBar(self._decoded_token, parent=self)
self.setStatusBar(self.status_bar)
self.daq = DAQWorker(base_url=self._base_url, token=self._token)
self.installEventFilter(self)
self._idle_timer = QTimer(self)
self._idle_timer.setInterval(60_000)
self._idle_timer.timeout.connect(self._check_idle_timeout)
self._idle_timer.start()
self._remote_close_timer = QTimer(self)
self._remote_close_timer.setInterval(1000)
self._remote_close_timer.timeout.connect(self._check_remote_close_deadline)
self._axis_camera_refresh_timer = QTimer(self)
self._axis_camera_refresh_timer.setInterval(self._axis_camera_refresh_interval_ms)
self._axis_camera_refresh_timer.timeout.connect(self.refresh_axis_cameras)
self._axis_camera_refresh_timer.start()
self.portrait_mode_panel.wire_to_main_window(
job_list_panel=self.job_list_panel, tell_samples=self.tell_samples
)
self.portrait_mode_panel._back_btn.clicked.connect(self._return_from_portrait_mode)
self.portrait_mode_panel.grab_session_requested.connect(self.status_bar.request_baton)
# Route alert banner signals through portrait-aware interceptors
self.daq.polled_devices_status.connect(self._portrait_alert_primary)
self.daq.detector_error.connect(self._portrait_alert_secondary)
self.daq.baton_status_changed.connect(self.status_bar.update_baton_status)
self.daq.baton_status_changed.connect(self._on_baton_status_changed)
self.daq.baton_request_result.connect(self._on_baton_request_result)
self.daq.baton_response_result.connect(self._on_baton_response_result)
self.daq.baton_timeout_checked.connect(self._on_baton_timeout_checked)
self.daq.automation_progress.connect(self.automation_progress_panel.set_progress)
self.daq.automation_progress.connect(self.compact_automation_panel.set_progress)
self.daq.automation_progress.connect(self.portrait_mode_panel.set_progress)
self.status_bar.baton_request_received.connect(self._show_baton_request_dialog)
self.status_bar.baton_request_accepted.connect(self._accept_baton_request)
self.status_bar.baton_request_refused.connect(self._refuse_baton_request)
self.daq.spreadsheet.connect(self.tell_samples.new_sample_list)
if self._decoded_token.staff:
self.daq.reference_tools.connect(self.ref_tools_panel.new_list)
self.samcam.changed.connect(self.daq.samcam_settings)
self.samcam.screenshot_requested.connect(self.daq.send_screenshot_db)
self.samcam.save_beam_location_setting.connect(self.daq.save_beam_location_camera_setting)
self.data_collection.find_tip.clicked.connect(self.daq.center_loop)
self.data_collection.bounding_box.clicked.connect(self.daq.ml_bounding_box)
self.daq.raster_generated_by_ml.connect(self.raster.update_active_grid_request)
# Clicking the big session badge (SESSION VACANT / Guest Mode) opens
# the grab/request menu right at the cursor.
self.sample_camera.session_badge_clicked.connect(
lambda: self.status_bar.show_session_menu(QCursor.pos())
)
self.sample_camera.smargon.connect(self.daq.move_smargon)
self.beamline.smargon_panel.smargon.connect(self.daq.move_smargon)
self.sample_camera.samcam_updated.connect(self.daq.samcam_settings)
self.raster.omega.connect(self.daq.set_omega)
self.raster.smargon.connect(self.daq.move_smargon)
self.beamline.omega_panel.set_omega_rel.connect(self.daq.set_omega_rel)
self.beamline.omega_panel.set_omega.connect(self.daq.set_omega)
self.sample_camera.set_omega.connect(self.daq.set_omega)
self.beamline.zoom_panel.zoom.connect(self.daq.zoom)
self.beamline.illumination_panel.front_light.connect(self.daq.front_light)
self.beamline.illumination_panel.back_light.connect(self.daq.back_light)
if self._decoded_token.staff:
self.monochromator_panel.mono_pitch_scan.connect(self.daq.mono_pitch_scan)
self.monochromator_panel.change_energy.connect(self.daq.change_energy)
self.abr_tweak.abr_tweak.connect(self.daq.abr_tweak)
self.abr_tweak.abr_save.connect(self.daq.abr_save)
self.abr_tweak.abr_goto_meas.connect(self.daq.abr_goto_meas)
self.beam_mark.beam_mark_clear.connect(self.daq.beam_mark_clear)
self.sample_camera.update_beam_mark.connect(self.daq.beam_mark_add)
self.beam_center.beam_center.connect(self.daq.beam_center)
self.beam_size.beam_size.connect(self.daq.beam_size_mm)
self.sample_camera.load_image.connect(self.raster.load_image)
self.sample_camera.switch_raster_grid.connect(self.data_collection.switch_to_raster)
self.samcam.show_detections_changed.connect(self.sample_camera.set_show_detections)
self.samcam.show_detection_polygons_changed.connect(
self.sample_camera.set_show_detection_polygons
)
self.samcam.show_target_point_changed.connect(self.sample_camera.set_show_target_point)
self.samcam.show_target_coordinates_changed.connect(
self.sample_camera.set_show_target_coordinates
)
self.samcam.show_overlay_legend_changed.connect(self.sample_camera.set_show_overlay_legend)
self.samcam.compact_overlay_legend_changed.connect(
self.sample_camera.set_compact_overlay_legend
)
self.samcam.target_color_changed.connect(self.sample_camera.set_target_color)
self._restore_samcam_overlay_settings()
sample_feed_addr = pred_zmq_addr or zmq_addr
if sample_feed_addr is not None:
logger.debug(f"Starting prediction subscriber thread {sample_feed_addr}")
self.prediction_thread = PredictionSubscriber(pred_zmq_url=sample_feed_addr, topic=b"")
self.prediction_thread.image.connect(self.sample_camera.update_pixmap)
self.prediction_thread.image.connect(self.compact_sample_camera.update_pixmap)
self.prediction_thread.image.connect(self.portrait_sample_camera.update_pixmap)
self.prediction_thread.prediction.connect(self.sample_camera.update_detections)
self.prediction_thread.prediction.connect(self.compact_sample_camera.update_detections)
self.prediction_thread.prediction.connect(self.portrait_sample_camera.update_detections)
self.prediction_thread.prediction.connect(
self.prediction_metrics_panel.update_from_prediction
)
self.prediction_thread.target_point.connect(self.sample_camera.update_target_point)
self.prediction_thread.target_point.connect(
self.compact_sample_camera.update_target_point
)
self.prediction_thread.target_point.connect(
self.portrait_sample_camera.update_target_point
)
self.prediction_thread.target_point.connect(
self.target_stability_panel.update_target_point
)
self.prediction_thread.focus_measure.connect(self.status_bar.update_sharpness)
self.prediction_thread.fps_measure.connect(self.status_bar.update_samcam_fps)
self.prediction_thread.camera_availability_changed.connect(
self.sample_camera.set_camera_available
)
self.prediction_thread.camera_availability_changed.connect(
self.compact_sample_camera.set_camera_available
)
self.prediction_thread.camera_availability_changed.connect(
self.portrait_sample_camera.set_camera_available
)
self.prediction_thread.camera_availability_changed.connect(
self._on_sample_camera_availability_changed
)
self.prediction_thread.camera_error.connect(self._on_sample_camera_error)
self.prediction_thread.start()
else:
self.prediction_thread = None
self.sample_camera.set_camera_available(False)
self.compact_sample_camera.set_camera_available(False)
self.portrait_sample_camera.set_camera_available(False)
self._show_samcam_feed_banner("Sample camera feed unavailable: no stream configured")
#
# self.data_collection.helical.helical_scan.connect(self.worker.helical_scan)
# self.data_collection.helical.update_bookmarks.connect(
# self.camera_image.update_bookmarks
# )
#
self.sample_camera.autofocus.connect(self.daq.autofocus)
self.sample_camera.evaluate_grid.connect(self.raster.run_grid_scan)
self.data_collection.raster.evaluate_grid.connect(self.raster.run_grid_scan)
self.data_collection.raster.evaluate_grid_auto.connect(self.raster.run_grid_scan_auto)
self.sample_camera.clear_evaluated_grids.connect(self.raster.clear_completed_grids)
self.sample_camera.clear_grid.connect(self.raster.clear_active_grid)
self.daq.run_number_incremented.connect(
self.data_collection.file_path_panel.increment_run_number
)
self.job_list_panel.auto_scan.connect(self.daq.automated_scan)
self.job_list_panel.unmount.connect(self.daq.unmount)
self.job_list_panel.park_and_dry.connect(self.daq.park_and_dry)
# Gate manual mounts/unmounts on the hutch PSS state so the user gets an
# immediate pop-up instead of the robot failing to move server-side.
self.tell_samples.mount.connect(self._on_manual_mount_requested)
self.tell_samples.unmount.connect(self._on_manual_unmount_requested)
self.ref_tools_panel.mount.connect(self._on_manual_mount_requested)
self.ref_tools_panel.unmount.connect(self._on_manual_unmount_requested)
# Dewar table doubles as the queue view: right-click edits queue
# membership, and every queue change repaints the status tints.
self.tell_samples.add_to_queue.connect(
lambda lst: self.job_list_panel.queue_samples(lst.s, replace=False)
)
self.tell_samples.remove_from_queue.connect(
lambda lst: self.job_list_panel.remove_samples([s.db_id for s in lst.s])
)
self.job_list_panel.table_model.modelReset.connect(self._sync_queue_row_tints)
self.data_collection.raster.grid_size_updated.connect(self.raster.update_grid_size)
self.data_collection.raster.exp_time_updated.connect(self.raster.update_exposure_time)
self.data_collection.raster.transmission_updated.connect(self.raster.update_transmission)
self.data_collection.raster.dtz_updated.connect(self.raster.update_dtz)
self.data_collection.raster.grid_metric_updated.connect(self.raster.metric)
self.data_collection.raster.raster_alpha_changed.connect(self.sample_camera.raster_alpha)
self.data_collection.cancel.connect(self.daq.cancel)
self.raster.grid_scan.connect(self.daq.raster_scan)
self.raster.grid_scan_auto.connect(self.daq.raster_scan_auto)
self.data_collection.screening.rotation_scan.connect(self.daq.standard_scan)
self.data_collection.simple.rotation_scan.connect(self._on_simple_rotation_requested)
self.data_collection.simple.parameters_changed.connect(self.daq.smart_params)
self.raster.grid_scan_size_changed.connect(
self.data_collection.raster.grid_scan_size_change
)
self.status_bar.set_pgroup.connect(self.daq.set_pgroup)
self.status_bar.end_session.connect(self.daq.end_session)
self.status_bar.force_session.connect(self.daq.force_session)
self.status_bar.request_baton.connect(self.daq.request_baton)
self.status_bar.cancel_baton_request.connect(self.daq.cancel_baton_request)
self.status_bar.release_baton.connect(self.daq.release_baton)
self.status_bar.baton_request_accepted.connect(
lambda: self.daq.respond_to_baton_request(True)
)
self.status_bar.baton_request_refused.connect(
lambda: self.daq.respond_to_baton_request(False)
)
self.status_bar.dewar_exchange.connect(self.daq.dewar_exchange)
self.status_bar.sample_exchange.connect(self.daq.sample_exchange)
self.status_bar.sample_alignment.connect(self.daq.sample_alignment)
self.status_bar.beam_location.connect(self.daq.beam_location)
self.status_bar.close_shutter.connect(self.daq.close_shutter)
self.status_bar.open_shutter.connect(self.daq.open_shutter)
self.beamline_state_panel.dewar_exchange.connect(self.daq.dewar_exchange)
self.beamline_state_panel.sample_exchange.connect(self.daq.sample_exchange)
self.beamline_state_panel.sample_alignment.connect(self.daq.sample_alignment)
self.beamline_state_panel.beam_location.connect(self.daq.beam_location)
self.beamline_state_panel.beamstop_alignment.connect(self.daq.beamstop_alignment)
self.beamline_state_panel.flux_measurement.connect(self.daq.flux_measurement)
self.beamline_state_panel.data_collection.connect(self.daq.data_collection)
self.beamline_state_panel.xtal_snapshot.connect(self.daq.xtal_snapshot)
self.beamline_state_panel.xray_fluorescence.connect(self.daq.xray_fluorescence)
self.beamline_state_panel.robot_sample_exchange.connect(self.daq.robot_sample_exchange)
self.rotation.file_ready.connect(self.viewer.load_image)
self.raster.image_selected.connect(self.viewer.load_image)
self.raster.viewer_track_online.connect(self.viewer.load_online)
self.data_collection.screening.viewer_track_online.connect(self.viewer.load_online)
self.job_list_panel.viewer_track_online.connect(self.viewer.load_online)
self.sample_logic.sample_changed.connect(self.data_collection.file_path_panel.update_sample)
self.daq.update.connect(self.beamline.omega_panel.update_daq_status)
self.daq.update.connect(self.beamline.smargon_panel.update_daq_status)
self.daq.update.connect(self.job_list_panel.update_daq_status)
self.daq.update.connect(self.data_collection.file_path_panel.update_daq_status)
self.daq.update.connect(self.data_collection.update_daq_status)
self.daq.update.connect(self.beamline.illumination_panel.update_daq_status)
self.daq.update.connect(self.raster.update_daq_status)
self.daq.update.connect(self.sample_camera.update_daq_status)
self.daq.update.connect(self.compact_sample_camera.update_daq_status)
self.daq.update.connect(self.portrait_sample_camera.update_daq_status)
self.daq.update.connect(self.tell_samples.update_daq_status)
self.daq.update.connect(self.ref_tools_panel.update_daq_status)
if self.prediction_thread is not None:
self.daq.update.connect(self.prediction_thread.update_daq_status)
if self._decoded_token.staff:
self.daq.update.connect(self.monochromator_panel.update_daq_status)
self.daq.update.connect(self.beam_size.update_daq_status)
self.daq.update.connect(self.beam_center.update_daq_status)
self.daq.update.connect(self.abr_tweak.update_daq_status)
self.daq.update.connect(self.beam_mark.update_daq_status)
self.daq.update.connect(self.status_bar.update_daq_status)
self.daq.update.connect(self.samcam.update_daq_status)
self.daq.update.connect(self.beamline.zoom_panel.update_daq_status)
self.daq.update.connect(self.sample_logic.update_daq_status)
self.daq.update.connect(self.manual_sample_panel.update_daq_status)
self.daq.update.connect(self.fluor_panel.update_daq_status)
self.daq.update.connect(self.beamline_state_panel.update_daq_status)
self.daq.update.connect(self.update_daq_status)
self.daq.sample_missing.connect(self.show_sample_missing_dialog)
# Generic/background HTTP errors -> non-modal runtime alert; failed
# user operations (mount/unmount/...) -> modal pop-up.
self.daq.http_error.connect(self._on_http_error)
self.daq.operation_failed.connect(self._on_operation_failed)
self.daq.pss_alarm_changed.connect(self._on_pss_alarm_changed)
self.daq.standard_scan_completed.connect(self.rotation.scan_completed)
self.daq.raster_scan_completed.connect(self.raster.grid_scan_completed)
self.daq.automated_scan_done.connect(self.job_list_panel.automated_scan_done)
self.daq.automated_scan_done.connect(self._mark_scan_result)
self.daq.automation_critical_failure.connect(self._on_automation_critical_failure)
self.daq.manual_collection_critical_failure.connect(
self._on_manual_collection_critical_failure
)
self.daq.recovery_action_completed.connect(
lambda _msg: self._clear_automation_critical_banner()
)
self.manual_sample_panel.sample_manual.connect(self.daq.sample_manual)
self.face_panel.face_detection.connect(self.daq.face_detection)
self.daq.face_detection_result.connect(self.face_panel.update_plot)
self.status_bar.get_all_pgroups.connect(self.daq.get_all_pgroups)
self.daq.staff_pgroups_loaded.connect(self.status_bar.staff_pgroups_loaded)
self.data_collection.fluo.fluo_scan.connect(self._on_fluo_scan_requested)
self.daq.fluorimeter_spectrum_update.connect(self.fluor_panel.update_plot)
self.daq.fluorimeter_spectrum_update.connect(lambda: self.fluor_panel_dock.setVisible(True))
# === Alert/Status Message Routing ===
# Status bar: General status messages (not device connection status)
self.daq.status_message.connect(self.status_bar.show_connection_message)
register_tutorials(self, self.tutorial_manager)
# Every dock: closing its popped-out window brings it back into the
# layout instead of hiding it (eventFilter), and popping out opens
# an enlarged window (2x width, 3x height, clamped to the screen).
for dock in self.findChildren(QDockWidget):
dock.installEventFilter(self)
dock.topLevelChanged.connect(self._on_dock_top_level_changed)
def _setup_global_shortcuts(self) -> None:
self._shortcut_manual_sample = QAction("Expand Manual Sample", self)
self._shortcut_manual_sample.setShortcut(QKeySequence("Ctrl+M"))
self._shortcut_manual_sample.triggered.connect(
lambda: self.manual_sample_panel.title.expand()
)
self.addAction(self._shortcut_manual_sample)
self._shortcut_raise_sample_list = QAction("Raise sample list", self)
self._shortcut_raise_sample_list.setShortcut(QKeySequence("Ctrl+L"))
self._shortcut_raise_sample_list.triggered.connect(
lambda: (
self.tell_samples_dock.setVisible(True),
self.tell_samples_dock.raise_(),
self.sample_lists_tabs.setCurrentIndex(0),
)
)
self.addAction(self._shortcut_raise_sample_list)
# Reference tools now live inside the Sample List dock as the
# Auxiliary-puck tab — Ctrl+R raises the dock on that tab.
self._shortcut_raise_reference_tools_list = QAction("Raise reference tools", self)
self._shortcut_raise_reference_tools_list.setShortcut(QKeySequence("Ctrl+R"))
self._shortcut_raise_reference_tools_list.triggered.connect(self._raise_reference_tools)
self.addAction(self._shortcut_raise_reference_tools_list)
# The automation controls live under the Dewar samples tab now.
self._shortcut_raise_job_list = QAction("Raise automation list", self)
self._shortcut_raise_job_list.setShortcut(QKeySequence("Ctrl+J"))
self._shortcut_raise_job_list.triggered.connect(
lambda: (
self.tell_samples_dock.setVisible(True),
self.tell_samples_dock.raise_(),
self.sample_lists_tabs.setCurrentIndex(0),
)
)
self.addAction(self._shortcut_raise_job_list)
self._shortcut_toggle_target_stability = QAction("Toggle target stability panel", self)
self._shortcut_toggle_target_stability.setShortcut(QKeySequence("Ctrl+Shift+T"))
self._shortcut_toggle_target_stability.triggered.connect(
lambda: self.target_stability_dock.setVisible(
not self.target_stability_dock.isVisible()
)
)
self.addAction(self._shortcut_toggle_target_stability)
self._shortcut_toggle_prediction_metrics = QAction("Toggle prediction metrics panel", self)
self._shortcut_toggle_prediction_metrics.setShortcut(QKeySequence("Ctrl+Shift+P"))
self._shortcut_toggle_prediction_metrics.triggered.connect(
lambda: self.prediction_metrics_dock.setVisible(
not self.prediction_metrics_dock.isVisible()
)
)
self.addAction(self._shortcut_toggle_prediction_metrics)
self._shortcut_toggle_smargon_trace = QAction("Toggle smargon trace panel", self)
self._shortcut_toggle_smargon_trace.setShortcut(QKeySequence("Ctrl+Shift+S"))
self._shortcut_toggle_smargon_trace.triggered.connect(
lambda: self.smargon_trace_dock.setVisible(not self.smargon_trace_dock.isVisible())
)
self.addAction(self._shortcut_toggle_smargon_trace)
self._shortcut_console_log = QAction("Toggle Console Log", self)
self._shortcut_console_log.setShortcut(QKeySequence("Ctrl+Shift+L"))
self._shortcut_console_log.triggered.connect(
lambda: (self.log_dock.setVisible(True), self.log_dock.raise_())
)
self.addAction(self._shortcut_console_log)
@Slot()
def _sync_queue_row_tints(self) -> None:
self.tell_samples.table_model.set_queued_ids(
sample.db_id for sample in self.job_list_panel.table_model.samples
)
@Slot()
def _remove_selected_from_queue(self) -> None:
self._unqueue_panel_selection(self.tell_samples)
def _unqueue_panel_selection(self, panel: TellSamplePanel) -> None:
rows = panel.table_view.selectionModel().selectedRows()
db_ids = [panel.table_model.get_id(index.row()).db_id for index in rows]
if db_ids:
self.job_list_panel.remove_samples(db_ids)
@Slot(int, bool, str)
def _mark_scan_result(self, db_id: int, success: bool, reply: str) -> None:
# Display only: a failed run tints the row pale red until a later
# success clears it. Auth errors are not the sample's fault — skip.
if reply == "Authentication Error":
return
self.tell_samples.table_model.set_flagged(db_id, not success)
@Slot()
def _open_sample_popout(self) -> None:
if self._sample_popout is None:
# Fully operational duplicate of the Sample List tab: second panel
# instances SHARING the docked panels' models, wired to the same
# slots — mounts, queue edits, chips and filters all work here.
dewar_panel = TellSamplePanel(model=self.tell_samples.table_model)
aux_panel = ReferenceToolsPanel(model=self.ref_tools_panel.table_model)
dewar_panel.mount.connect(self._on_manual_mount_requested)
dewar_panel.unmount.connect(self._on_manual_unmount_requested)
dewar_panel.add_to_queue.connect(
lambda lst: self.job_list_panel.queue_samples(lst.s, replace=False)
)
dewar_panel.remove_from_queue.connect(
lambda lst: self.job_list_panel.remove_samples([s.db_id for s in lst.s])
)
aux_panel.mount.connect(self._on_manual_mount_requested)
aux_panel.unmount.connect(self._on_manual_unmount_requested)
# One shared filter state → keep the two chip rows visually in sync.
self.tell_samples.status_chips.buttonClicked.connect(
lambda chip: dewar_panel.set_status_chip(chip.property("status_key"))
)
dewar_panel.status_chips.buttonClicked.connect(
lambda chip: self.tell_samples.set_status_chip(chip.property("status_key"))
)
dewar_panel.set_status_chip(self.tell_samples.table_model.status_filter)
dewar_tab = QWidget()
dewar_layout = QVBoxLayout(dewar_tab)
dewar_layout.setContentsMargins(0, 0, 0, 0)
dewar_layout.setSpacing(2)
dewar_layout.addWidget(dewar_panel)
dewar_layout.addLayout(self._clone_automation_row(dewar_panel))
tabs = QTabWidget()
tabs.addTab(dewar_tab, "Dewar samples")
tabs.addTab(aux_panel, "Auxiliary puck")
if not self._decoded_token.staff:
tabs.setTabEnabled(1, False)
tabs.setTabToolTip(1, "Staff only")
self._sample_popout = PopoutWindow("Sample List", tabs, parent=self)
self._sample_popout.resize(1200, 500)
self._sample_popout.show()
self._sample_popout.raise_()
self._sample_popout.activateWindow()
def _open_automation_popout(self) -> None:
if self._automation_popout is None:
# Mirror wired to the same feeds as the docked panel.
panel = AutomationProgressWidget()
self.job_list_panel.samples_in_queue_changed.connect(panel.set_samples_in_queue)
self.job_list_panel.automation_running_changed.connect(panel.set_running)
self.daq.automation_progress.connect(panel.set_progress)
panel.set_samples_in_queue(len(self.job_list_panel.table_model.samples))
panel.set_running(self.job_list_panel.is_running())
self._automation_popout = PopoutWindow("Automation progress", panel, parent=self)
self._automation_popout.resize(420, 520)
self._automation_popout.show()
self._automation_popout.raise_()
self._automation_popout.activateWindow()
def _clone_automation_row(self, dewar_panel: TellSamplePanel) -> QHBoxLayout:
"""Pop-out copy of the automation controls, driving the same queue
engine. Run/Pause text and checkbox states stay mirrored; 'Remove
selected' acts on the pop-out's own table selection."""
jl = self.job_list_panel
run_button = QPushButton(jl.play_button.text())
run_button.clicked.connect(jl.run)
jl.automation_running_changed.connect(
lambda running: run_button.setText("⏸ Pause" if running else "▶ Run")
)
remove_button = QPushButton("🗑 Remove selected")
remove_button.clicked.connect(lambda: self._unqueue_panel_selection(dewar_panel))
clear_button = QPushButton("✖ Clear list")
clear_button.clicked.connect(jl.clear)
unmount_button = QPushButton("⏏ Unmount")
unmount_button.clicked.connect(lambda: self._on_manual_unmount_requested())
row = QHBoxLayout()
for button in (run_button, remove_button, clear_button, unmount_button):
row.addWidget(button)
for source in (jl.park_and_dry_when_cleared, jl.pause_on_conditions_cb):
clone = QCheckBox(source.text())
clone.setToolTip(source.toolTip())
clone.setChecked(source.isChecked())
# setChecked with an unchanged value emits nothing, so the
# cross-connection cannot loop.
clone.toggled.connect(source.setChecked)
source.toggled.connect(clone.setChecked)
row.addWidget(clone)
return row
def _show_reference_tools_staff_only_popup(self) -> None:
QMessageBox.information(
self,
"Staff only",
"The Auxiliary puck (reference tools) view is available to staff accounts only.",
)
@Slot()
def _raise_reference_tools(self) -> None:
if not self._decoded_token.staff:
self._show_reference_tools_staff_only_popup()
return
self.tell_samples_dock.setVisible(True)
self.tell_samples_dock.raise_()
self.sample_lists_tabs.setCurrentIndex(1)
def _return_to_main_view_for_shutdown(self) -> None:
try:
if getattr(self, "content_stack", None) is None:
return
current_widget = self.content_stack.currentWidget()
if hasattr(self, "portrait_mode_page") and current_widget is self.portrait_mode_page:
self._return_from_portrait_mode()
elif bool(getattr(self, "_in_compact_automation_view", False)):
self._return_from_compact_automation_view()
if hasattr(self, "content_stack") and hasattr(self, "_standard_main_page"):
self.content_stack.setCurrentWidget(self._standard_main_page)
except Exception as e:
logger.warning(f"Failed to return to main view during shutdown: {e}", exc_info=True)
def _restore_samcam_overlay_settings(self) -> None:
settings = QSettings("PSI", "AareGUI")
# bool()/str() wraps: QSettings.value is typed "object" even with
# type=..., so the wraps are runtime no-ops for the pyright gate.
show_detections = bool(settings.value("samcam/show_detections", True, type=bool))
show_detection_polygons = bool(
settings.value("samcam/show_detection_polygons", True, type=bool)
)
show_target_point = bool(settings.value("samcam/show_target_point", True, type=bool))
show_target_coordinates = bool(
settings.value("samcam/show_target_coordinates", True, type=bool)
)
show_overlay_legend = bool(settings.value("samcam/show_overlay_legend", True, type=bool))
compact_overlay_legend = bool(
settings.value("samcam/compact_overlay_legend", False, type=bool)
)
target_color = str(settings.value("samcam/target_color", "Cyan", type=str))
self.samcam.apply_overlay_settings(
show_detections=show_detections,
show_detection_polygons=show_detection_polygons,
show_target_point=show_target_point,
show_target_coordinates=show_target_coordinates,
show_overlay_legend=show_overlay_legend,
compact_overlay_legend=compact_overlay_legend,
target_color=target_color,
)
self.sample_camera.set_show_detections(show_detections)
self.sample_camera.set_show_detection_polygons(show_detection_polygons)
self.sample_camera.set_show_target_point(show_target_point)
self.sample_camera.set_show_target_coordinates(show_target_coordinates)
self.sample_camera.set_show_overlay_legend(show_overlay_legend)
self.sample_camera.set_compact_overlay_legend(compact_overlay_legend)
self.sample_camera.set_target_color(target_color)
def _save_samcam_overlay_settings(self) -> None:
settings = QSettings("PSI", "AareGUI")
overlay = self.sample_camera.target_overlay_settings()
settings.setValue("samcam/show_detections", overlay["show_detections"])
settings.setValue(
"samcam/show_detection_polygons", overlay["show_detection_polygons"]
) # NEW
settings.setValue("samcam/show_target_point", overlay["show_target_point"])
settings.setValue("samcam/show_target_coordinates", overlay["show_target_coordinates"])
settings.setValue("samcam/show_overlay_legend", overlay["show_overlay_legend"])
settings.setValue("samcam/compact_overlay_legend", overlay["compact_overlay_legend"])
settings.setValue("samcam/target_color", overlay["target_color"])
@Slot(bool)
def _on_sample_camera_availability_changed(self, available: bool) -> None:
if available:
self._clear_samcam_feed_banner()
return
self._show_samcam_feed_banner("Sample camera feed unavailable")
@Slot(str)
def _on_sample_camera_error(self, message: str) -> None:
logger.warning(message)
self._show_samcam_feed_banner(message or "Sample camera feed unavailable")
def _show_samcam_feed_banner(self, message: str) -> None:
self._samcam_feed_banner_message = message
self._show_runtime_notification(
title="Sample camera", message=message, level="warning", sticky=True
)
self._samcam_feed_banner_active = True
def _clear_samcam_feed_banner(self) -> None:
if not self._samcam_feed_banner_active:
return
self._clear_runtime_notification()
self._samcam_feed_banner_active = False
def _stop_axis_camera_threads(self) -> None:
for attr_name in ("beamline_camera_thread", "gonio_camera_thread"):
thread = getattr(self, attr_name, None)
if thread is None:
continue
try:
thread.stop()
except Exception as e:
logger.warning(f"Failed to stop {attr_name}: {e}", exc_info=True)
setattr(self, attr_name, None)
@staticmethod
def _axis_busy_text_from_status(s: DAQStatusModel) -> str:
tell_state = getattr(s, "tell_state", None)
if tell_state is None:
return "BEAMLINE BUSY"
activity_value = str(getattr(tell_state.activity, "value", "") or "").lower()
if activity_value not in {"mounting", "unmounting", "drying", "cooling"}:
return "BEAMLINE BUSY"
try:
activity_name = tell_state.activity.display_name()
except Exception:
logger.debug("Could not derive the axis busy text from status", exc_info=True)
activity_name = activity_value.capitalize() if activity_value else "Busy"
return f"TELL {activity_name}".upper()
def _start_axis_camera_threads(self) -> None:
self._stop_axis_camera_threads()
if self._beamline_cam_addr:
self.beamline_camera_thread = VideoThread(ip=self._beamline_cam_addr)
self.beamline_camera_thread.frame_ready.connect(self.beamline_view.update_frame)
self.beamline_camera_thread.frame_ready.connect(
self.beamline_view_2_combined.update_frame
)
self.beamline_camera_thread.start()
if self._gonio_cam_addr and self._gonio_cam_id:
self.gonio_camera_thread = VideoThread(
ip=self._gonio_cam_addr, camera=self._gonio_cam_id
)
self.gonio_camera_thread.frame_ready.connect(self.gonio_view.update_frame)
self.gonio_camera_thread.frame_ready.connect(self.beamline_view_1_combined.update_frame)
self.gonio_camera_thread.start()
def _all_tell_samples_in_default_order(self) -> list:
samples = list(getattr(self.tell_samples.table_model, "samples", []))
return sorted(
[sample for sample in samples if sample.location is not None],
key=lambda sample: sample.loc_str_sort(),
)
def _update_view_mode_actions(self) -> None:
in_automation_view = bool(self._in_compact_automation_view)
if self._enter_automation_view_action is not None:
self._enter_automation_view_action.setVisible(not in_automation_view)
self._enter_automation_view_action.setEnabled(not in_automation_view)
if self._return_main_view_action is not None:
self._return_main_view_action.setVisible(in_automation_view)
self._return_main_view_action.setEnabled(in_automation_view)
@Slot()
def enter_compact_automation_view(self) -> None:
self.job_list_panel.ensure_default_queue_from_samples(
self._all_tell_samples_in_default_order()
)
self._refresh_compact_queue_preview()
if not self._in_compact_automation_view:
self._pre_automation_window_state = self.saveState()
self._pre_automation_left_column_visible = self.collection_controls_scroll.isVisible()
self._pre_automation_right_column_visible = self.beamline_controls_scroll.isVisible()
self.tell_samples_dock.setVisible(False)
self.automation_progress_dock.setVisible(False)
self.face_panel_dock.setVisible(False)
self.fluor_panel_dock.setVisible(False)
self.smargon_trace_dock.setVisible(False)
self.target_stability_dock.setVisible(False)
self.prediction_metrics_dock.setVisible(False)
self.log_dock.setVisible(False)
self.collection_controls_scroll.setVisible(False)
self.beamline_controls_scroll.setVisible(False)
self.content_stack.setCurrentWidget(self.compact_automation_page)
self._in_compact_automation_view = True
self._update_view_mode_actions()
@Slot()
def _return_from_compact_automation_view(self) -> None:
self.content_stack.setCurrentWidget(self._standard_main_page)
if self._pre_automation_window_state is not None:
self.restoreState(self._pre_automation_window_state)
self.collection_controls_scroll.setVisible(self._pre_automation_left_column_visible)
self.beamline_controls_scroll.setVisible(self._pre_automation_right_column_visible)
self._in_compact_automation_view = False
self._update_view_mode_actions()
self.tell_samples_dock.setVisible(True)
self.tell_samples_dock.raise_()
@Slot()
def _refresh_compact_queue_preview(self) -> None:
current_sample, next_sample, next_next_sample = self.job_list_panel.queue_preview()
self.compact_automation_panel.set_samples(current_sample, next_sample, next_next_sample)
@Slot()
def enter_portrait_mode(self) -> None:
"""Switch to the portrait / phone-screen view and resize the window."""
self._pre_portrait_geometry = self.saveGeometry()
self.portrait_mode_panel.set_running(self.job_list_panel.is_running())
self.portrait_mode_panel.set_samples_in_queue(len(self.job_list_panel.table_model.samples))
self._refresh_portrait_queue_preview()
self.content_stack.setCurrentWidget(self.portrait_mode_page)
# ── Camera: scale-to-fit + hide legend ─────────────────────────────
self.portrait_sample_camera.set_show_overlay_legend(False)
try:
self.portrait_sample_camera._autoscale = True
self.portrait_sample_camera._scaling()
except Exception:
logger.debug("Could not autoscale the portrait-mode camera", exc_info=True)
# ── Hide all chrome that contributes to window width ────────────────
if self.status_bar is not None:
self.status_bar.setVisible(False)
self.menuBar().setVisible(False)
# Alert banners take up horizontal space even when hidden via QFrame
# — force them to zero height so they cannot influence the minimum width.
self.alert_banner.setVisible(False)
self.alert_banner.setMaximumHeight(0)
self.alert_banner_secondary.setVisible(False)
self.alert_banner_secondary.setMaximumHeight(0)
# Hide all dock widgets
for dock_attr in (
"tell_samples_dock",
"automation_progress_dock",
"face_panel_dock",
"fluor_panel_dock",
"smargon_trace_dock",
"target_stability_dock",
"prediction_metrics_dock",
"log_dock",
):
dock = getattr(self, dock_attr, None)
if dock is not None:
dock.setVisible(False)
# ── Resize to phone footprint ───────────────────────────────────────
screen = QGuiApplication.screenAt(self.geometry().center())
if screen is None:
screen = QGuiApplication.primaryScreen()
available = screen.availableGeometry()
portrait_w = self.portrait_mode_panel.PORTRAIT_WIDTH + 24
portrait_h = min(860, available.height() - 40)
new_x = available.x() + (available.width() - portrait_w) // 2
new_y = available.y() + (available.height() - portrait_h) // 2
self.setMinimumWidth(portrait_w)
self.setMaximumWidth(portrait_w)
self.resize(portrait_w, portrait_h)
self.move(new_x, new_y)
@Slot()
def _return_from_portrait_mode(self) -> None:
"""Restore the window to its pre-portrait geometry and switch page."""
# ── Lift hard width cap before restoring geometry ───────────────────
self.setMinimumWidth(0)
self.setMaximumWidth(16777215) # Qt's QWIDGETSIZE_MAX
self.content_stack.setCurrentWidget(self._standard_main_page)
# ── Restore chrome ──────────────────────────────────────────────────
if self.status_bar is not None:
self.status_bar.setVisible(True)
self.menuBar().setVisible(True)
# Restore alert banners to normal operation
self.alert_banner.setMaximumHeight(16777215)
self.alert_banner_secondary.setMaximumHeight(16777215)
# Replay any pending messages that arrived during portrait mode
self.portrait_mode_panel._flush_portrait_alerts_to_banners(
self.alert_banner, self.alert_banner_secondary
)
# ── Restore camera legend ───────────────────────────────────────────
try:
settings = self.portrait_sample_camera.target_overlay_settings()
self.portrait_sample_camera.set_show_overlay_legend(
settings.get("show_overlay_legend", True)
)
except Exception:
logger.debug("Could not restore the camera overlay legend", exc_info=True)
if hasattr(self, "_pre_portrait_geometry") and self._pre_portrait_geometry:
self.restoreGeometry(self._pre_portrait_geometry)
self._pre_portrait_geometry = None
self.tell_samples_dock.setVisible(True)
self.automation_progress_dock.setVisible(False)
self.face_panel_dock.setVisible(False)
self.fluor_panel_dock.setVisible(False)
self.smargon_trace_dock.setVisible(False)
self.target_stability_dock.setVisible(False)
self.prediction_metrics_dock.setVisible(False)
self.log_dock.setVisible(False)
@Slot(str, bool)
def _portrait_alert_primary(self, msg: str, is_error: bool) -> None:
"""Route primary alerts into the runtime dock."""
if self.content_stack.currentWidget() is self.portrait_mode_page:
self.portrait_mode_panel.show_portrait_alert(msg, is_error)
else:
self._show_runtime_notification(
title="Runtime alert" if is_error else "Runtime update",
message=msg,
level="error" if is_error else "success",
sticky=is_error,
auto_clear_ms=None if is_error else 4000,
)
@Slot(str, bool)
def _portrait_alert_secondary(self, msg: str, is_error: bool) -> None:
"""Route secondary alerts into the runtime dock."""
if self.content_stack.currentWidget() is self.portrait_mode_page:
self.portrait_mode_panel.show_portrait_alert(msg, is_error)
else:
self._show_runtime_notification(
title="Device alert" if is_error else "Device update",
message=msg,
level="warning" if is_error else "info",
sticky=is_error,
auto_clear_ms=None if is_error else 4000,
)
@Slot()
def _refresh_portrait_queue_preview(self) -> None:
self.portrait_mode_panel.refresh_queue_preview()
@staticmethod
def _annotation_token(annotation: str) -> str:
mapping = {
"Heart": "❤️",
"Thumbs Up": "👍",
"Thumbs Down": "👎",
"Eyes": "👀",
"Scan Again": "scan again",
}
return mapping.get(annotation, str(annotation).strip())
@staticmethod
def _append_annotation_to_comment(existing_comment: str | None, annotation: str) -> str:
token = MainWindow._annotation_token(annotation)
current = str(existing_comment or "").strip()
if not current:
return token
current_tokens = [part.strip() for part in current.split(" | ") if part.strip()]
if token in current_tokens:
return current
return f"{current} | {token}"
@Slot(str)
def _handle_compact_annotation(self, annotation: str) -> None:
current_sample, _, _ = self.job_list_panel.queue_preview()
if current_sample is None:
self.status_bar.show_connection_message("No sample selected for annotation.", True)
return
updated_comment = self._append_annotation_to_comment(
getattr(current_sample, "comment", None), annotation
)
self.job_list_panel.annotate_sample_comment(current_sample.db_id, updated_comment)
self.tell_samples.annotate_sample_comment(current_sample.db_id, updated_comment)
self._refresh_compact_queue_preview()
self.status_bar.show_connection_message(
f"Annotation added: {self._annotation_token(annotation)} — {current_sample.sample_name}",
False,
)
@Slot()
def refresh_axis_cameras(self) -> None:
logger.info("Refreshing Axis camera threads")
self._stop_axis_camera_threads()
self._start_axis_camera_threads()
def start_text_tutorial(self) -> None:
self.tutorial_manager.start("manual_workflow_demo")
def start_interactive_tutorial(self) -> None:
self.tutorial_manager.start("manual_workflow_demo")
def _apply_theme(self) -> None:
# Screenshot cross-fade (THEME_FADE_MS in styles.py): freeze the old
# look in a click-through overlay, restyle beneath it in one normal
# repolish, fade the overlay out. QSS has no transitions, and
# animating the palette would re-polish every widget per frame.
old_look = self.grab() if self.isVisible() else None
# Native primitives QSS can't recolor (spin/combo arrow glyphs) draw
# from the app palette — flip its text roles with the theme.
app = QApplication.instance()
if not hasattr(self, "_default_palette"):
self._default_palette = app.palette()
if self._theme_mode == THEME_PORTRAIT:
palette = QPalette(self._default_palette)
for role in (
QPalette.ColorRole.ButtonText,
QPalette.ColorRole.Text,
QPalette.ColorRole.WindowText,
):
palette.setColor(role, qcolor(DARK_TEXT))
app.setPalette(palette)
else:
app.setPalette(self._default_palette)
self.setStyleSheet(build_app_stylesheet(self._theme_mode))
# State colors are painted in code per DAQ tick — QSS can't reach them.
self.beamline_state_panel.set_theme(self._theme_mode)
if old_look is None:
return
overlay = QLabel(self)
overlay.setAttribute(Qt.WidgetAttribute.WA_TransparentForMouseEvents, True)
overlay.setPixmap(old_look)
overlay.setGeometry(self.rect())
overlay.raise_()
overlay.show()
effect = QGraphicsOpacityEffect(overlay)
overlay.setGraphicsEffect(effect)
fade = QPropertyAnimation(effect, b"opacity", overlay)
fade.setDuration(THEME_FADE_MS)
fade.setStartValue(1.0)
fade.setEndValue(0.0)
fade.finished.connect(overlay.deleteLater)
fade.start(QPropertyAnimation.DeletionPolicy.DeleteWhenStopped)
def _restore_theme_settings(self) -> None:
settings = QSettings("PSI", "AareGUI")
self._theme_mode = settings.value("appearance/theme", THEME_ORIGINAL, type=str)
def _save_theme_settings(self) -> None:
settings = QSettings("PSI", "AareGUI")
settings.setValue("appearance/theme", self._theme_mode)
@Slot()
def use_legacy_theme(self) -> None:
self._theme_mode = THEME_ORIGINAL
self._apply_theme()
@Slot()
def use_portrait_theme(self) -> None:
self._theme_mode = THEME_PORTRAIT
self._apply_theme()
def create_menu_bar(self):
"""Create a menu bar with File->Quit and Help->About."""
menu_bar = self.menuBar()
file_menu = menu_bar.addMenu("File")
quit_action = QAction("&Quit", self)
quit_action.setShortcut(QKeySequence.StandardKey.Quit)
quit_action.triggered.connect(self.close)
file_menu.addAction(quit_action)
self._enter_automation_view_action = QAction("Automation View", self)
self._enter_automation_view_action.setShortcut(QKeySequence("Ctrl+5"))
self._enter_automation_view_action.triggered.connect(self.enter_compact_automation_view)
menu_bar.addAction(self._enter_automation_view_action)
self._return_main_view_action = QAction("Return to Main View", self)
self._return_main_view_action.setShortcut(QKeySequence("Ctrl+Shift+5"))
self._return_main_view_action.triggered.connect(self._return_from_compact_automation_view)
menu_bar.addAction(self._return_main_view_action)
self._portrait_mode_action = QAction("Portrait Mode", self)
self._portrait_mode_action.setShortcut(QKeySequence("Ctrl+6"))
self._portrait_mode_action.triggered.connect(self.enter_portrait_mode)
menu_bar.addAction(self._portrait_mode_action)
view_menu = menu_bar.addMenu("View")
self._theme_action_group = QActionGroup(self)
self._theme_action_group.setExclusive(True)
self._use_legacy_theme_action = QAction("Legacy Theme", self)
self._use_legacy_theme_action.setCheckable(True)
self._use_legacy_theme_action.setChecked(self._theme_mode == THEME_ORIGINAL)
self._use_legacy_theme_action.triggered.connect(self.use_legacy_theme)
self._theme_action_group.addAction(self._use_legacy_theme_action)
self._use_portrait_theme_action = QAction("Portrait Theme", self)
self._use_portrait_theme_action.setCheckable(True)
self._use_portrait_theme_action.setChecked(self._theme_mode == THEME_PORTRAIT)
self._use_portrait_theme_action.triggered.connect(self.use_portrait_theme)
self._theme_action_group.addAction(self._use_portrait_theme_action)
view_menu.addAction(self._use_legacy_theme_action)
view_menu.addAction(self._use_portrait_theme_action)
view_menu.addSeparator()
show_samples_action = QAction("Show Sample List", self)
show_samples_action.setCheckable(True)
show_samples_action.setChecked(True)
show_samples_action.triggered.connect(
lambda checked: self.tell_samples_dock.setVisible(checked)
)
self.tell_samples_dock.visibilityChanged.connect(show_samples_action.setChecked)
view_menu.addAction(show_samples_action)
show_face_panel_action = QAction("Show face detection", self)
show_face_panel_action.setCheckable(True)
show_face_panel_action.setChecked(False)
show_face_panel_action.triggered.connect(
lambda checked: self.face_panel_dock.setVisible(checked)
)
self.face_panel_dock.visibilityChanged.connect(show_face_panel_action.setChecked)
view_menu.addAction(show_face_panel_action)
show_fluor_panel_action = QAction("Show fluorescence", self)
show_fluor_panel_action.setCheckable(True)
show_fluor_panel_action.setChecked(False)
show_fluor_panel_action.triggered.connect(
lambda checked: self.fluor_panel_dock.setVisible(checked)
)
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(show_smargon_trace_action.setChecked)
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_target_stability_action = QAction("Show Target stability", self)
show_target_stability_action.setCheckable(True)
show_target_stability_action.setChecked(False)
show_target_stability_action.triggered.connect(
lambda checked: self.target_stability_dock.setVisible(checked)
)
self.target_stability_dock.visibilityChanged.connect(
show_target_stability_action.setChecked
)
view_menu.addAction(show_target_stability_action)
show_prediction_metrics_action = QAction("Show Prediction Metrics", self)
show_prediction_metrics_action.setCheckable(True)
show_prediction_metrics_action.setChecked(False)
show_prediction_metrics_action.triggered.connect(
lambda checked: self.prediction_metrics_dock.setVisible(checked)
)
self.prediction_metrics_dock.visibilityChanged.connect(
show_prediction_metrics_action.setChecked
)
view_menu.addAction(show_prediction_metrics_action)
show_log_action = QAction("Show Log", self)
show_log_action.setCheckable(True)
show_log_action.setChecked(False)
show_log_action.triggered.connect(lambda checked: self.log_dock.setVisible(checked))
self.log_dock.visibilityChanged.connect(show_log_action.setChecked)
view_menu.addAction(show_log_action)
view_menu.addSeparator()
sample_camera_tab_action = QAction("Sample camera tab", self)
sample_camera_tab_action.setShortcut(QKeySequence("Ctrl+1"))
sample_camera_tab_action.triggered.connect(lambda: self.video_tab.setCurrentIndex(0))
view_menu.addAction(sample_camera_tab_action)
gonio_camera_tab_action = QAction("Gonio camera tab", self)
gonio_camera_tab_action.setShortcut(QKeySequence("Ctrl+2"))
gonio_camera_tab_action.triggered.connect(lambda: self.video_tab.setCurrentIndex(1))
view_menu.addAction(gonio_camera_tab_action)
beamline_view_tab_action = QAction("Beamline view tab", self)
beamline_view_tab_action.setShortcut(QKeySequence("Ctrl+3"))
beamline_view_tab_action.triggered.connect(lambda: self.video_tab.setCurrentIndex(2))
view_menu.addAction(beamline_view_tab_action)
beamline_combined_tab_action = QAction("Beamline combined view tab", self)
beamline_combined_tab_action.setShortcut(QKeySequence("Ctrl+4"))
beamline_combined_tab_action.triggered.connect(lambda: self.video_tab.setCurrentIndex(3))
view_menu.addAction(beamline_combined_tab_action)
view_menu.addSeparator()
restore_default_view_action = QAction("Restore Default View", self)
restore_default_view_action.setShortcut(QKeySequence("Ctrl+Shift+R"))
restore_default_view_action.triggered.connect(self.restore_default_view)
view_menu.addAction(restore_default_view_action)
help_menu = menu_bar.addMenu("Help")
about_action = QAction("About", self)
about_action.triggered.connect(self.show_about_dialog)
help_menu.addAction(about_action)
controls_help_action = QAction("Mouse / Keyboard Controls", self)
controls_help_action.setShortcut(QKeySequence(Qt.Key.Key_F1))
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)
help_menu.addSeparator()
refresh_axis_cameras_action = QAction("Refresh Axis Cameras", self)
refresh_axis_cameras_action.triggered.connect(self.refresh_axis_cameras)
help_menu.addAction(refresh_axis_cameras_action)
if self._decoded_token.staff:
local_contact_action = QAction("Local Contact", self)
local_contact_action.triggered.connect(self.show_local_contact)
help_menu.addAction(local_contact_action)
help_menu.addSeparator()
start_text_tutorial_action = QAction("Start Tutorial (Text)", self)
start_text_tutorial_action.triggered.connect(self.start_text_tutorial)
help_menu.addAction(start_text_tutorial_action)
start_interactive_tutorial_action = QAction("Start Tutorial (Interactive)", self)
start_interactive_tutorial_action.triggered.connect(self.start_interactive_tutorial)
help_menu.addAction(start_interactive_tutorial_action)
def _capture_default_window_state(self) -> None:
self._default_window_state = self.saveState()
@Slot()
def restore_default_view(self) -> None:
if self._default_window_state is not None:
self.restoreState(self._default_window_state)
self.content_stack.setCurrentWidget(self._standard_main_page)
self._in_compact_automation_view = False
self._update_view_mode_actions()
self.collection_controls_scroll.setVisible(True)
self.beamline_controls_scroll.setVisible(True)
self.tell_samples_dock.setVisible(True)
self.face_panel_dock.setVisible(False)
self.fluor_panel_dock.setVisible(False)
self.smargon_trace_dock.setVisible(False)
self.target_stability_dock.setVisible(False)
self.prediction_metrics_dock.setVisible(False)
self.log_dock.setVisible(False)
self.tell_samples_dock.raise_()
self.video_tab.setCurrentIndex(0)
def show_about_dialog(self):
QMessageBox.about(self, "About", about_text(self.daq))
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(
daq=self.daq,
is_staff=bool(getattr(self._decoded_token, "staff", False)),
parent=self,
)
self._dev_help_dialog.refresh()
self._dev_help_dialog.show()
self._dev_help_dialog.raise_()
self._dev_help_dialog.activateWindow()
def _show_runtime_notification(
self,
*,
title: str,
message: str,
level: str = "error",
sticky: bool = True,
auto_clear_ms: int | None = None,
) -> None:
self.log_dock.show_notification(
title=title, message=message, level=level, sticky=sticky, auto_clear_ms=auto_clear_ms
)
def _show_runtime_waiting_notification(self, *, title: str, message: str) -> None:
self.log_dock.show_waiting_notification(title=title, message=message)
def _clear_runtime_notification(self) -> None:
self.log_dock.clear_notification()
def _clear_automation_critical_banner(self) -> None:
if not self._automation_critical_banner_active:
return
self._clear_runtime_notification()
self._automation_critical_banner_active = False
@Slot(bool)
def _on_automation_running_changed(self, running: bool) -> None:
if running:
self._clear_automation_critical_banner()
self._show_runtime_notification(
title="Automation running",
message="Automation resumed.",
level="success",
sticky=False,
auto_clear_ms=3500,
)
def _is_detector_critical_failure(self, message: str) -> bool:
text = (message or "").lower()
return "detector" in text or "jfjoch" in text
@staticmethod
def _is_detector_state_error(message: str) -> bool:
text = (message or "").lower()
return (
"daq state error" in text
or "must be idle to start measurement" in text
or "must be idle" in text
)
def _detector_error_dialog_title(self, message: str) -> str:
if self._is_detector_state_error(message):
return "Detector state error"
return "Detector error"
def _detector_error_banner_text(self, *, automation: bool, message: str) -> str:
if self._is_detector_state_error(message):
prefix = "Automation halted" if automation else "Manual collection stopped"
return f"{prefix}: detector state error."
return (
"Automation halted: detector error."
if automation
else "Manual collection stopped: detector error."
)
@Slot(str)
@Slot(str)
def _on_http_error(self, message: str) -> None:
"""Surface generic/background HTTP errors as a non-modal runtime alert.
These come from polling, resync and other background requests; a modal
pop-up would be too intrusive, so they go to the runtime dock/banner.
"""
self._portrait_alert_primary(message, True)
@Slot(bool)
def _on_pss_alarm_changed(self, active: bool) -> None:
"""Non-modal warning banner for the hutch PSS alarm (ALARM-STATE != 0).
Edge-triggered from /status: shows a sticky warning while active and a
brief confirmation when it clears.
"""
if active:
self._show_runtime_notification(
title="Hutch safety alarm",
message=(
"The hutch personnel-safety system reports an active alarm "
"(ALARM-STATE != 0). Mounting is blocked until it clears — "
"check the hutch / call your local contact."
),
level="warning",
sticky=True,
)
else:
self._show_runtime_notification(
title="Hutch safety alarm cleared",
message="The hutch safety alarm has cleared.",
level="success",
sticky=False,
auto_clear_ms=4000,
)
@Slot(str, str, bool)
def _on_operation_failed(self, title: str, message: str, critical: bool) -> None:
"""Modal pop-up for a failed user-triggered operation (mount, unmount, ...).
Critical failures (e.g. door safety could not be activated) also pause
the automation queue defensively, so we never keep dispatching samples.
"""
logger.error(f"Operation failed [{title}] (critical={critical}): {message}")
if critical:
try:
if self.job_list_panel is not None and self.job_list_panel.is_running():
self.job_list_panel.pause_automation()
except Exception:
logger.exception("Failed to pause automation after operation failure")
try:
show = QMessageBox.critical if critical else QMessageBox.warning
show(self, title, message)
except Exception:
logger.exception("Failed to show operation failure popup")
def _hutch_blocks_mount(self) -> str | None:
"""Reason the hutch PSS currently blocks a mount, or None if OK.
Uses the latest ``/status``; if no status has arrived yet we defer to
the authoritative server-side gate rather than guess.
"""
bl = getattr(self._latest_daq_status, "bl", None)
if bl is None:
return None
if getattr(bl, "pss_prohibited", True) is False:
return (
"Door safety could not be activated: close the hutch doors and "
"complete the safety search before mounting."
)
if getattr(bl, "pss_alarm", False):
return "The hutch safety alarm is active. Mounting is blocked until it clears."
return None
def _on_manual_mount_requested(self, sample, reference: bool = False) -> None:
"""Pre-check the hutch before sending a manual mount to the server.
Gives the user an immediate pop-up if the door is open / alarm active,
instead of the robot silently failing to move server-side.
"""
reason = self._hutch_blocks_mount()
if reason is not None:
logger.warning(f"Manual mount blocked by hutch PSS: {reason}")
try:
QMessageBox.critical(self, "Mounting Failed", reason)
except Exception:
logger.exception("Failed to show mount-blocked popup")
return
self.daq.mount(sample, reference)
def _on_manual_unmount_requested(self) -> None:
"""Block a manual unmount if the hutch isn't ready (robot can't move)."""
reason = self._hutch_blocks_mount()
if reason is not None:
logger.warning(f"Manual unmount blocked by hutch PSS: {reason}")
try:
QMessageBox.critical(self, "Unmounting Failed", reason)
except Exception:
logger.exception("Failed to show unmount-blocked popup")
return
self.daq.unmount()
def _precondition_ok(self) -> bool:
"""Run the shared ring-current / shutter / door 'continue?' check.
Uses the latest ``/status``; defers (returns True) if no status yet.
"""
bl = getattr(self._latest_daq_status, "bl", None)
if bl is None:
return True
return precondition_check(
self,
ring_current=getattr(bl, "ring_current_mA", None),
shutter_open=getattr(bl, "exp_shutter_open", None),
door_prohibited=getattr(bl, "pss_prohibited", None),
)
def _on_fluo_scan_requested(self, params) -> None:
if self._precondition_ok():
self.daq.fluorimeter_spectrum(params)
def _on_simple_rotation_requested(self, request) -> None:
if self._precondition_ok():
self.daq.standard_scan(request)
def _on_manual_collection_critical_failure(self, message: str) -> None:
logger.critical(f"Manual collection critical detector failure: {message}")
self._show_runtime_notification(
title="Collection paused", message=message, level="error", sticky=True
)
try:
QMessageBox.critical(
self,
self._detector_error_dialog_title(message),
(
"The manual collection has stopped because there is an error with the detector.\n\n"
"Please call your local contact.\n\n"
f"Details:\n{message}"
),
)
except Exception:
logger.exception("Failed to show manual collection detector popup")
@Slot(str)
def _on_automation_critical_failure(self, message: str) -> None:
"""
Hard-stop handler for fatal automation failures coming from the server
(transformation errors, busy-flag corruption, unhandled 500s on /scan/auto).
Stops the queue, marks the progress widget as failed and surfaces the
beamline recovery dialog (staff) or a critical error message (users).
"""
logger.critical(f"Automation critical failure: {message}")
is_detector_failure = self._is_detector_critical_failure(message)
(
self._detector_error_banner_text(automation=True, message=message)
if is_detector_failure
else f"Automation halted: {message}"
)
# 1. Stop the sample queue so we don't keep dispatching new samples
try:
if self.job_list_panel is not None and self.job_list_panel.is_running():
self.job_list_panel.pause_automation()
if self.daq is not None:
self.daq.send_status_request()
except Exception:
logger.exception("Failed to pause automation queue after critical failure")
# 2. Mark the automation progress widget as finished-with-error so
# _is_automation_active() returns False and idle/close timers behave.
try:
from aarecommon.models.automation import (
AutomationProgress,
StepState,
StepStatus,
WorkflowStateKind,
)
progress = getattr(self.automation_progress_panel, "_progress", None)
if progress is None:
progress = AutomationProgress(
current_step=None,
steps=[
StepState(step=WorkflowStateKind.MOUNT, status=StepStatus.PENDING),
StepState(step=WorkflowStateKind.LOOP_CENTRE, status=StepStatus.PENDING),
StepState(step=WorkflowStateKind.RASTER, status=StepStatus.PENDING),
StepState(
step=WorkflowStateKind.DATA_COLLECTION, status=StepStatus.PENDING
),
StepState(step=WorkflowStateKind.FINAL, status=StepStatus.PENDING),
],
finished=False,
success=None,
)
for step in progress.steps:
if step.status == StepStatus.RUNNING:
step.status = StepStatus.FAILED
step.message = message
if step.step == WorkflowStateKind.FINAL:
step.status = StepStatus.FAILED
step.message = message
progress.finished = True
progress.success = False
self.automation_progress_panel.set_progress(progress)
except Exception:
logger.exception("Failed to update automation progress after critical failure")
self._show_runtime_notification(
title="Automation paused", message=message, level="error", sticky=True
)
# 4. Surface recovery UI
try:
if bool(getattr(self._decoded_token, "staff", False)):
if is_detector_failure:
QMessageBox.critical(
self,
self._detector_error_dialog_title(message),
(
"Automation has been stopped because there is an error with the detector.\n\n"
f"Details:\n{message}"
),
)
else:
QMessageBox.critical(
self,
"Automation halted",
(
"A critical error occurred during automation and the "
"beamline could not recover automatically:\n\n"
f"{message}\n\n"
),
)
else:
if is_detector_failure:
QMessageBox.critical(
self,
self._detector_error_dialog_title(message),
(
"Automation has been stopped because there is an error with the detector.\n\n"
"Please call your local contact.\n\n"
f"Details:\n{message}"
),
)
else:
QMessageBox.critical(
self,
"Automation halted",
(
"A critical error occurred during automation and the "
"beamline could not recover automatically:\n\n"
f"{message}\n\n"
"Please contact your local contact to recover the beamline."
),
)
except Exception:
logger.exception("Failed to surface recovery UI after critical failure")
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()
def show_local_contact(self, tab_name: str = "Status") -> None:
if not bool(getattr(self._decoded_token, "staff", False)):
return
if self._local_contact_dialog is None:
self._local_contact_dialog = LocalContactDialog(daq=self.daq, parent=self)
self._local_contact_dialog.set_active_tab(tab_name)
self._local_contact_dialog.show()
self._local_contact_dialog.raise_()
self._local_contact_dialog.activateWindow()
@Slot(str)
def show_sample_missing_dialog(self, msg: str):
if self.job_list_panel.is_running():
logger.warning(f"Automation: Sample missing: {msg}")
return
QMessageBox.warning(self, "No Sample", f"{msg}")
# TODO tidy up mount and sampel view fucntions
@Slot()
def mount_view(self):
self.video_tab.setCurrentWidget(self.beamline_combined_panel)
@Slot()
def sample_view(self):
self.video_tab.setCurrentWidget(self.sample_camera)
def _apply_session_gate(self, session_state) -> None:
# No baton -> watching only: camera views stay live, every operating
# surface is greyed. The SESSION VACANT badge (and the status bar
# session menu) remain the way back in.
owned = session_state in (SessionsStateEnum.OwnedByYou, SessionsStateEnum.PendingElseToYou)
if owned == getattr(self, "_session_operations_enabled", None):
return
self._session_operations_enabled = owned
for widget in (self.left_column_tabs, self.beamline, self.tell_samples_dock.widget()):
widget.setEnabled(owned)
if owned:
# None clears the effect (Qt API contract); the PySide6 stub
# signature misses the Optional.
widget.setGraphicsEffect(None) # pyright: ignore[reportArgumentType]
else:
# Full grayscale, banners included — QSS :disabled alone
# can't reach the custom-painted TitleLabels/inline styles.
effect = QGraphicsColorizeEffect(widget)
effect.setColor(QColor("#808080"))
widget.setGraphicsEffect(effect)
self.sample_camera.set_operations_enabled(owned)
# Vacant folds every panel shut; grabbing reopens exactly the ones
# that were open before. Transient (persist=False) so the fold never
# overwrites the user's saved per-panel choices.
banners = self.left_column_tabs.findChildren(TitleLabel) + self.beamline.findChildren(
TitleLabel
)
if owned:
for banner in getattr(self, "_pre_vacancy_open_banners", []):
banner.set_collapsed(False, persist=False)
else:
self._pre_vacancy_open_banners = [b for b in banners if not b.is_collapsed()]
for banner in banners:
banner.set_collapsed(True, persist=False)
@Slot(DAQStatusModel)
def update_daq_status(self, s: DAQStatusModel):
self._latest_daq_status = s
self._apply_session_gate(getattr(getattr(s, "session", None), "session", None))
if self._is_automation_active():
self._refresh_idle_activity(report_backend=False)
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)
busy_style = build_busy_overlay_style(
is_busy=bool(s.busy),
tell_state=s.tell_state,
session_state=getattr(getattr(s, "session", None), "session", None),
)
if hasattr(self, "beamline_view_panel") and self.beamline_view_panel is not None:
self.beamline_view_panel.set_busy_style(busy_style)
if hasattr(self, "gonio_view_panel") and self.gonio_view_panel is not None:
self.gonio_view_panel.set_busy_style(busy_style)
if hasattr(self, "beamline_combined_panel") and self.beamline_combined_panel is not None:
self.beamline_combined_panel.set_busy_style(busy_style)
self.target_stability_panel.set_beam_center(
s.geom.beam_location_pxl.x, s.geom.beam_location_pxl.y
)
self._refresh_compact_queue_preview()
current_session = int(getattr(self._decoded_token, "session", -1))
for gui in getattr(s, "open_guis", []) or []:
try:
if int(gui.session) == current_session and bool(gui.close_requested):
if self._remote_close_deadline_ts is None:
self._start_remote_close_countdown(
requested_by=gui.close_requested_by,
grace_seconds=gui.close_grace_seconds,
)
break
except Exception:
logger.debug("Skipping an unreadable GUI session entry", exc_info=True)
continue
else:
if self._remote_close_deadline_ts is not None:
self._clear_remote_close_request()
if not self._mounting and s.state == BeamlineStateEnum.RobotSampleExchange:
self._mounting = True
self.video_tab.setCurrentWidget(self.beamline_combined_panel)
elif self._mounting and s.state != BeamlineStateEnum.RobotSampleExchange:
self._mounting = False
self.video_tab.setCurrentWidget(self.sample_camera)
# ========== BATON DIALOG HANDLING ==========
@Slot(dict)
def _show_baton_request_dialog(self, payload: dict):
logger.info(f"Showing baton request dialog with payload: {payload}")
requester = str(payload.get("requester") or "Another user")
timeout = int(payload.get("timeout") or 30)
if self._baton_request_dialog is not None:
if self._baton_request_dialog.isVisible():
logger.debug("Baton request dialog already visible, skipping")
return
else:
logger.debug("Baton request dialog exists but not visible, recreating")
self._baton_request_dialog.close()
self._baton_request_dialog = None
self._baton_request_dialog = BatonRequestDialog(
requester=requester, timeout_seconds=timeout, parent=self
)
self._baton_request_dialog.accepted_signal.connect(
self.status_bar._on_baton_dialog_accepted
)
self._baton_request_dialog.refused_signal.connect(self.status_bar._on_baton_dialog_refused)
self._baton_request_dialog.show()
self._baton_request_dialog.raise_()
self._baton_request_dialog.activateWindow()
logger.info("Baton request dialog shown")
@Slot()
def _accept_baton_request(self):
self.daq.respond_to_baton_request(True)
@Slot()
def _refuse_baton_request(self):
self.daq.respond_to_baton_request(False)
@Slot(BatonStatus)
def _on_baton_status_changed(self, status: BatonStatus):
"""
Update waiting UI based on SSE status.
StatusBar handles baton-granted p-group selection automatically.
"""
if self._waiting_for_baton_response and not status.you_have_pending_request:
self._waiting_for_baton_response = False
self._close_baton_pending_dialog()
if status.you_are_holder:
self.alert_banner.show_message("Baton acquired!", False, auto_clear_ms=10000)
else:
self.alert_banner.show_message(
"Request declined or cancelled", False, auto_clear_ms=10000
)
# Manage incoming request dialog (when someone requests from us)
if not status.incoming_request and self._baton_request_dialog is not None:
logger.info("Incoming baton request no longer active, closing request dialog")
self._close_baton_dialog()
@Slot(dict)
def _on_baton_request_result(self, result: dict):
"""
Handle result of our baton request.
- Show success/pending/error banners
- Manage pending dialog lifecycle
- StatusBar handles p-group selection automatically via update_baton_status
"""
if result.get("granted"):
self._waiting_for_baton_response = False
self.alert_banner.show_message("Baton acquired!", False, auto_clear_ms=10000)
logger.info("Baton acquired")
# Close pending dialog; StatusBar will trigger p-group selection via SSE
self._close_baton_pending_dialog()
elif result.get("pending"):
self._waiting_for_baton_response = True
timeout = result.get("timeout_seconds", 30)
holder = result.get("message", "Waiting for response...")
is_busy = result.get("beamline_busy", False)
if getattr(self, "_baton_pending_dialog", None) is None:
target_user = holder.replace("Request sent to ", "").replace(
" (Note: beamline is currently busy, transfer will be queued if accepted)", ""
)
self._baton_pending_dialog = BatonPendingDialog(
target_user=target_user, timeout_seconds=timeout, parent=self
)
self._baton_pending_dialog.cancelled_signal.connect(self.daq.cancel_baton_request)
self._baton_pending_dialog.show()
else:
self._baton_pending_dialog.update_remaining(timeout)
self.alert_banner.show_waiting(f"Requesting control - {holder}", timeout)
logger.info(f"Baton request pending - {timeout}s timeout (busy={is_busy})")
elif result.get("queued"):
self._waiting_for_baton_response = True
self.alert_banner.show_waiting("Control transfer queued - waiting for beamline")
logger.info("Baton transfer queued")
if getattr(self, "_baton_pending_dialog", None) is None:
self._baton_pending_dialog = BatonPendingDialog(
target_user="Current Holder", timeout_seconds=0, parent=self
)
self._baton_pending_dialog.cancelled_signal.connect(self.daq.cancel_baton_request)
self._baton_pending_dialog.show()
self._baton_pending_dialog.set_queued_state()
elif result.get("already_holder"):
self._waiting_for_baton_response = False
logger.debug("Already baton holder")
self._close_baton_pending_dialog()
elif result.get("error"):
self._waiting_for_baton_response = False
self.alert_banner.show_message(
result.get("message", "Request failed"), True, auto_clear_ms=15000
)
logger.warning(f"Baton request failed: {result.get('message')}")
self._close_baton_pending_dialog()
@Slot(dict)
def _on_baton_response_result(self, result: dict):
"""
Handle response after someone requests from us.
Just update UI banners; StatusBar handles session display.
"""
logger.debug(f"Baton response result: {result}")
if result.get("accepted"):
self._waiting_for_baton_response = False
self.alert_banner.show_message("Control transferred", False, auto_clear_ms=10000)
self._close_baton_dialog()
self.status_bar.update_baton_status(self.status_bar._baton_status)
elif result.get("refused"):
self._waiting_for_baton_response = False
self.alert_banner.show_message("Request declined", False, auto_clear_ms=10000)
self._close_baton_dialog()
self.status_bar.update_baton_status(self.status_bar._baton_status)
else:
logger.debug(f"replied with {result}")
@Slot(dict)
def _on_baton_timeout_checked(self, result: dict):
"""Refresh waiting UI when backend confirms timeout state."""
logger.debug(f"Baton timeout checked: {result}")
if result.get("pending"):
remaining = int(result.get("remaining_seconds", 0))
if self._waiting_for_baton_response:
self.alert_banner.show_waiting("Requesting control", remaining)
if getattr(self, "_baton_pending_dialog", None) is not None:
self._baton_pending_dialog.update_remaining(remaining)
elif result.get("granted"):
self._waiting_for_baton_response = False
self.alert_banner.show_message("Baton acquired!", False, auto_clear_ms=10000)
self._close_baton_pending_dialog()
elif result.get("queued"):
self._waiting_for_baton_response = True
self.alert_banner.show_waiting("Control transfer queued - waiting for beamline")
if getattr(self, "_baton_pending_dialog", None) is not None:
self._baton_pending_dialog.set_queued_state()
else:
self._baton_pending_dialog = BatonPendingDialog(
target_user="Current Holder", timeout_seconds=0, parent=self
)
self._baton_pending_dialog.cancelled_signal.connect(self.daq.cancel_baton_request)
self._baton_pending_dialog.show()
self._baton_pending_dialog.set_queued_state()
elif result.get("refused"):
self._waiting_for_baton_response = False
self.alert_banner.show_message("Request declined", False, auto_clear_ms=10000)
self._close_baton_pending_dialog()
else:
logger.debug(f"replied with {result}")
self.alert_banner.clear_message()
self._close_baton_pending_dialog()
def _close_baton_dialog(self) -> None:
"""Close the incoming-request dialog (when someone requests from us)."""
if getattr(self, "_baton_request_dialog", None) is not None:
try:
self._baton_request_dialog.close()
finally:
self._baton_request_dialog = None
def _close_baton_pending_dialog(self) -> None:
if getattr(self, "_baton_pending_dialog", None) is not None:
try:
if hasattr(self._baton_pending_dialog, "_timer"):
self._baton_pending_dialog._timer.stop()
self._baton_pending_dialog.close()
finally:
self._baton_pending_dialog = None
def _save_panel_visibility_settings(self) -> None:
"""Save visibility state for developer/diagnostic panels."""
settings = QSettings("PSI", "AareGUI")
settings.beginGroup("panel_visibility")
settings.setValue("smargon_trace", self.smargon_trace_dock.isVisible())
settings.setValue("target_stability", self.target_stability_dock.isVisible())
settings.setValue("prediction_metrics", self.prediction_metrics_dock.isVisible())
settings.setValue("face_detection", self.face_panel_dock.isVisible())
settings.setValue("fluorescence", self.fluor_panel_dock.isVisible())
settings.setValue("log", self.log_dock.isVisible())
settings.endGroup()
def _restore_panel_visibility_settings(self) -> None:
"""Restore visibility state for developer/diagnostic panels."""
settings = QSettings("PSI", "AareGUI")
settings.beginGroup("panel_visibility")
if settings.contains("smargon_trace"):
self.smargon_trace_dock.setVisible(settings.value("smargon_trace", False, type=bool))
if settings.contains("target_stability"):
self.target_stability_dock.setVisible(
settings.value("target_stability", False, type=bool)
)
if settings.contains("prediction_metrics"):
self.prediction_metrics_dock.setVisible(
settings.value("prediction_metrics", False, type=bool)
)
if settings.contains("face_detection"):
self.face_panel_dock.setVisible(settings.value("face_detection", False, type=bool))
if settings.contains("fluorescence"):
self.fluor_panel_dock.setVisible(settings.value("fluorescence", False, type=bool))
if settings.contains("log"):
self.log_dock.setVisible(settings.value("log", False, type=bool))
settings.endGroup()
def _restore_window_state(self) -> None:
# TODO put all setting related handlign into state_manager
self.state_manager.restore_window(self)
self._restore_panel_visibility_settings()
def showEvent(self, event):
super().showEvent(event)
# First show only, and only without a saved layout: re-apply the
# default dock split AFTER the real (maximized) geometry exists —
# the __init__ resizeDocks ran on the pre-show size and Qt hands the
# scale-up surplus to the sample list, skewing 50/50 into ~80/20.
if not getattr(self, "_default_dock_split_done", False):
self._default_dock_split_done = True
if not self.state_manager.settings.value("main_window/state"):
QTimer.singleShot(0, self._apply_default_dock_split)
def _apply_default_dock_split(self) -> None:
self.resizeDocks(
[self.tell_samples_dock, self.log_dock], [240, 240], Qt.Orientation.Vertical
)
self.resizeDocks(
[self.tell_samples_dock, self.automation_progress_dock],
[10000, 10000],
Qt.Orientation.Horizontal,
)
def closeEvent(self, event) -> None:
try:
self._return_to_main_view_for_shutdown()
except Exception as e:
logger.warning(f"Failed to restore main view before close: {e}", exc_info=True)
try:
# TODO put all setting related handling into state_manager
self.state_manager.save_window(self)
self._save_samcam_overlay_settings()
self._save_panel_visibility_settings()
self._save_theme_settings()
except Exception as e:
logger.warning(f"Failed to save main window state: {e}", exc_info=True)
# End session before closing so the backend removes this GUI from Redis immediately
try:
self.daq.end_session_on_close()
except Exception as e:
logger.warning(f"Failed to end session on close: {e}", exc_info=True)
try:
self.cleanup()
except Exception as e:
logger.warning(f"Cleanup during closeEvent failed: {e}", exc_info=True)
super().closeEvent(event)
def cleanup(self):
if getattr(self, "_cleanup_done", False):
return
try:
self._return_to_main_view_for_shutdown()
except Exception as e:
logger.warning(f"Failed to restore main view during cleanup: {e}", exc_info=True)
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}", exc_info=True)
try:
if hasattr(self, "_idle_timer") and self._idle_timer is not None:
self._idle_timer.stop()
except Exception as e:
logger.warning(f"Failed to stop _idle_timer: {e}", exc_info=True)
try:
if hasattr(self, "_remote_close_timer") and self._remote_close_timer is not None:
self._remote_close_timer.stop()
except Exception as e:
logger.warning(f"Failed to stop _remote_close_timer: {e}", exc_info=True)
try:
if (
hasattr(self, "_axis_camera_refresh_timer")
and self._axis_camera_refresh_timer is not None
):
self._axis_camera_refresh_timer.stop()
except Exception as e:
logger.warning(f"Failed to stop _axis_camera_refresh_timer: {e}", exc_info=True)
try:
if hasattr(self, "daq") and self.daq is not None:
self.daq.cleanup()
except Exception as e:
logger.warning(f"Failed to clean up DAQ worker: {e}", exc_info=True)
self._stop_axis_camera_threads()
for attr_name in ("prediction_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}", exc_info=True)
setattr(self, attr_name, None)
def _is_automation_active(self) -> bool:
try:
if hasattr(self, "job_list_panel") and self.job_list_panel is not None:
return bool(self.job_list_panel.is_running())
except Exception:
logger.debug("Could not read the job list panel running state", exc_info=True)
try:
progress = getattr(self.automation_progress_panel, "_progress", None)
if progress is not None and not bool(getattr(progress, "finished", False)):
return True
except Exception:
logger.debug("Could not read the automation progress state", exc_info=True)
return False
def _refresh_idle_activity(self, *, report_backend: bool = False) -> None:
self._last_user_interaction_ts = time.time()
if not report_backend:
return
now = time.monotonic()
if now - self._last_interaction_report_ts < self._interaction_report_min_interval_s:
return
self._last_interaction_report_ts = now
try:
self.daq.report_gui_interaction(int(self._decoded_token.session))
except Exception as e:
logger.debug(f"Failed to report GUI interaction: {e}", exc_info=True)
def _mark_user_interaction(self) -> None:
self._refresh_idle_activity(report_backend=True)
@Slot(bool)
def _on_dock_top_level_changed(self, floating: bool) -> None:
dock = self.sender()
if floating and isinstance(dock, QDockWidget):
# Pop-outs open enlarged instead of keeping the cramped docked
# size. Deferred: the window is mid-reparent while the signal
# fires.
QTimer.singleShot(0, lambda d=dock: self._enlarge_floating_dock(d))
def _enlarge_floating_dock(self, dock: QDockWidget) -> None:
if not dock.isFloating():
return
screen = dock.screen().availableGeometry()
dock.resize(
min(dock.width() * 2, int(screen.width() * 0.9)),
min(dock.height() * 3, int(screen.height() * 0.9)),
)
# Keep the enlarged window fully on screen.
geo = dock.frameGeometry()
dx = min(0, screen.right() - geo.right())
dy = min(0, screen.bottom() - geo.bottom())
if dx or dy:
dock.move(geo.x() + dx, geo.y() + dy)
def eventFilter(self, obj, event):
# Closing a floated (popped-out) dock re-docks it instead of hiding —
# otherwise the panel silently disappears and has to be restored via
# the View menu.
if event.type() == QEvent.Type.Close and isinstance(obj, QDockWidget) and obj.isFloating():
obj.setFloating(False)
event.ignore()
return True
try:
if event.type() in {
QEvent.Type.MouseButtonPress,
QEvent.Type.MouseButtonRelease,
QEvent.Type.MouseMove,
QEvent.Type.Wheel,
QEvent.Type.KeyPress,
QEvent.Type.KeyRelease,
QEvent.Type.FocusIn,
QEvent.Type.TouchBegin,
QEvent.Type.TouchUpdate,
}:
self._mark_user_interaction()
except Exception as e:
logger.debug(f"GUI interaction event filter error: {e}", exc_info=True)
# Non-staff click on the greyed-out Auxiliary-puck tab: only installed
# for non-staff, and tabAt() is geometric so it still sees the
# disabled tab — explain the lock instead of silently eating the click.
sample_tab_bar = self.sample_lists_tabs.tabBar()
if (
event.type() == QEvent.Type.MouseButtonPress
and sample_tab_bar is not None
and obj is sample_tab_bar
and sample_tab_bar.tabAt(event.position().toPoint()) == 1
):
self._show_reference_tools_staff_only_popup()
return True
return super().eventFilter(obj, event)
def _start_remote_close_countdown(
self, requested_by: str | None, grace_seconds: int | None
) -> None:
grace = max(1, int(grace_seconds or 60))
self._remote_close_deadline_ts = time.time() + grace
self._remote_close_reason = requested_by or "staff"
self._remote_close_timer.start()
self.alert_banner_secondary.show_message(
f"GUI close requested by {self._remote_close_reason}. Closing in {grace}s when safe.",
True,
)
self._remote_close_banner_active = True
def _clear_remote_close_request(self) -> None:
self._remote_close_deadline_ts = None
self._remote_close_reason = None
self._remote_close_timer.stop()
if self._remote_close_banner_active:
self.alert_banner_secondary.clear_message()
self._remote_close_banner_active = False
def _can_close_for_idle_or_remote(self) -> bool:
status = self._latest_daq_status
if status is not None and bool(status.busy):
return False
return not self._is_automation_active()
@Slot()
def _check_remote_close_deadline(self) -> None:
if self._remote_close_deadline_ts is None:
return
remaining = round(self._remote_close_deadline_ts - time.time())
if remaining > 0:
if self._remote_close_banner_active:
self.alert_banner_secondary.show_message(
f"GUI close requested by {self._remote_close_reason or 'staff'}. Closing in {remaining}s when safe.",
True,
)
return
if not self._can_close_for_idle_or_remote():
if self._remote_close_banner_active:
self.alert_banner_secondary.show_message(
"GUI close requested, waiting for beamline/automation to become idle.", True
)
return
logger.warning("Closing GUI due to remote close request.")
self.close()
@Slot()
def _check_idle_timeout(self) -> None:
idle_for_s = time.time() - self._last_user_interaction_ts
if idle_for_s < self._idle_close_timeout_s:
return
if not self._can_close_for_idle_or_remote():
logger.info(
"Idle timeout reached, but GUI remains open because beamline is busy or automation is active."
)
return
# TODO deny communication from GUI to DAQ while IDLE for too long rather than kill the GUI
logger.warning("Closing GUI after inactivity timeout.")
self.close()
Reference in New Issue View Git Blame Copy Permalink