Files
AareDAQ/src/aare/gui/main_window.py
T
perl_d 83a602da73
CI / lint (push) Skipped
CI / test (3.11) (push) Skipped
CI / test (3.12) (push) Skipped
CI / test (3.13) (push) Skipped
CI / test-with-beamline-plugins (pxi_bec) (push) Skipped
CI / test-with-beamline-plugins (pxii_bec) (push) Skipped
CI / test-with-beamline-plugins (pxiii_bec) (push) Skipped
CI / test (3.11) (pull_request) Failing after 44s
CI / test (3.12) (pull_request) Failing after 41s
CI / test (3.13) (pull_request) Failing after 42s
CI / test-with-beamline-plugins (pxii_bec) (pull_request) Failing after 47s
CI / test-with-beamline-plugins (pxi_bec) (pull_request) Failing after 49s
CI / test-with-beamline-plugins (pxiii_bec) (pull_request) Failing after 1m0s
CI / test-with-coverage (pull_request) Failing after 1m1s
CI / coverage-analysis (pull_request) Skipped
CI / lint (pull_request) Failing after 2m2s
style: british->american spelling
2026-08-24 13:33:14 +02:00

3405 lines
158 KiB
Python

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 (
QByteArray,
QEvent,
QObject,
QPropertyAnimation,
QSettings,
Qt,
QTimer,
Signal,
Slot,
)
from PySide6.QtGui import (
QAction,
QActionGroup,
QColor,
QCursor,
QGuiApplication,
QKeySequence,
QPalette,
)
from PySide6.QtWidgets import (
QAbstractButton,
QApplication,
QCheckBox,
QComboBox,
QDockWidget,
QFrame,
QGraphicsColorizeEffect,
QGraphicsOpacityEffect,
QHBoxLayout,
QLabel,
QMainWindow,
QMessageBox,
QPushButton,
QScrollArea,
QSizePolicy,
QSlider,
QStackedWidget,
QTabWidget,
QToolBar,
QToolTip,
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 LogPanel
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,
SEPARATOR_HINT_DELAY_MS,
THEME_BLUEBIRD,
THEME_FADE_MS,
THEME_SUNRISE,
THEME_SUNSET,
admin_tip_qss,
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 ClickableCursorFilter(QObject):
"""App-wide pointing-hand cursor on every button/combo/slider, present
and future. QSS cannot set cursors, and per-widget setCursor calls are
forgotten whenever a new widget is added — the Polish event fires once
for each widget when its style is applied, so this catches them all.
Widgets that manage their own cursor afterwards (beamline state strip's
forbidden cursor) still win: they set it later."""
def eventFilter(self, obj, event):
if event.type() == QEvent.Type.Polish and isinstance(
obj, (QAbstractButton, QComboBox, QSlider)
):
obj.setCursor(Qt.CursorShape.PointingHandCursor)
return False
class _AlertBannerHost(QWidget):
"""Content root that repositions the floating alert banners on resize.
A plain resizeEvent override instead of event filters: filters firing
during widget teardown corrupted PySide in the test suite."""
def resizeEvent(self, event):
super().resizeEvent(event)
for banner in self.findChildren(AlertBanner):
banner.reposition()
class MainWindow(QMainWindow):
sample_geometry = Signal(SampleGeometryModel)
# Class-level defaults: Qt can call overrides (showEvent, closeEvent)
# before __init__ finishes, so these must be readable without a getattr
# guard; also typed for the basedpyright gate.
_session_operations_enabled: bool | None = None
_default_dock_split_done: bool = False
_cleanup_done: bool = False
_pre_watch_dock_state: QByteArray | None = None
_pre_watch_visibility: list[tuple[QWidget, bool]] | None = None
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_BLUEBIRD
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._watching_motion = 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._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")
# App-level, not window-level: dialogs and pop-outs get it too.
# Installed once per process and parented to the app, NOT the window:
# the old per-window copies stacked up (one per MainWindow the test
# suite builds) and kept filtering every process event after their
# parent window died — stale wrappers in the hottest Qt→Python path.
app = QApplication.instance()
assert app is not None
if not app.property("_aare_app_filters_installed"):
app.setProperty("_aare_app_filters_installed", True)
app.installEventFilter(ClickableCursorFilter(app))
# 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.
app.installEventFilter(WheelValueGuard(app))
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 initializing).\n"
"Please start/restart the server and try again.",
)
raise
self.setStyleSheet(f"background-color: {APP_BACKGROUND};")
# Resize-line hint (see SEPARATOR_HINT in styles.py): WA_Hover gives
# us HoverMove without a button held; the single-shot timer implements
# the "rested for 1s" gate. The QSS ::separator:hover rule does the
# hit-testing, so event() never needs to know where separators are.
self.setAttribute(Qt.WidgetAttribute.WA_Hover, True)
self._separator_hint_timer = QTimer(self)
self._separator_hint_timer.setSingleShot(True)
self._separator_hint_timer.setInterval(SEPARATOR_HINT_DELAY_MS)
self._separator_hint_timer.timeout.connect(lambda: self._set_separator_hint(True))
root_widget = _AlertBannerHost(parent=self)
root_widget.setObjectName("mainContentRoot")
root_layout = QVBoxLayout(root_widget)
root_layout.setContentsMargins(0, 0, 0, 0)
root_layout.setSpacing(0)
# Banners float over the content instead of sitting in root_layout, so
# messages (e.g. "Baton acquired!") no longer shift the whole UI.
self.alert_banner = AlertBanner(parent=root_widget)
self.alert_banner.float_over(root_widget)
self.alert_banner_secondary = AlertBanner(parent=root_widget)
self.alert_banner_secondary.float_over(root_widget)
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(staff=self._decoded_token.staff, parent=self)
# Anchor for _show_admin_tip: QToolTip inherits QSS from the widget
# it is shown for, so this hidden label carries the red warning wash
# without reddening any other tooltip. Eager, not lazy — basedpyright
# requires instance attributes to exist after __init__.
self._admin_tip_anchor = QLabel(self)
self._admin_tip_anchor.hide()
# 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 gray 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)
# Non-staff keep the staff-only banners visible (locked stand-ins, a
# click explains the gate) so the Beamline tab reads the same for
# every role instead of looking amputated.
self._locked_beamline_banners: list[TitleLabel] = []
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)
else:
for title in ("Beamline setup", "ABR meas. pos.", "Beam configuration"):
# Holder with default layout margins so the banner aligns with
# the real panels' banners (same trick as BeamConfigPanel).
holder = QWidget(beamline_page)
holder_layout = QVBoxLayout(holder)
holder_layout.setSpacing(0)
banner = TitleLabel(title, holder)
banner.setCursor(Qt.CursorShape.PointingHandCursor)
banner.setToolTip("Only accessible for beamline scientists")
# eventFilter install happens LATE in __init__ (next to
# installEventFilter(self)) — see the comment there.
holder_layout.addWidget(banner)
beamline_layout.addWidget(holder)
self._locked_beamline_banners.append(banner)
# 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
# Non-staff: the first locked banner's holder is the column's
# first widget now, not samcam.
else self._locked_beamline_banners[0].parentWidget()
)
assert beamline_first is not None # the locked banner always has its holder
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
)
# Baton toasts sit as a compact pill under the camera view instead of
# a full-width bar at the top of the window.
self.alert_banner.anchor_to(self.sample_camera)
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_tab.setObjectName("dewarTab")
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()
# Queue controls act on the queue, whose order only the Queued chip
# view shows — so they are live there and grayed out elsewhere. The
# pop-out clones register themselves in _clone_automation_row. "run"
# kind stays enabled while automation runs: pausing is always allowed.
self._queue_action_buttons: list[tuple[QPushButton, str]] = [
(self.job_list_panel.play_button, "run"),
(self.job_list_panel.remove_button, "queue"),
(self.job_list_panel.clear_button, "queue"),
]
self.tell_samples.status_chips.buttonClicked.connect(
lambda _chip: self._update_queue_buttons_enabled()
)
self.job_list_panel.automation_running_changed.connect(
lambda _running: self._update_queue_buttons_enabled()
)
self._update_queue_buttons_enabled()
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). Gray 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)
# Tracks beamline-state TRANSITIONS for the default sample-tab switch
# (see _apply_default_sample_tab).
self._last_beamline_state: BeamlineStateEnum | None = None
# None = not yet evaluated, so the first status tick always applies
# the gate (the GUI may start while a visitor pgroup is active).
self._beamline_tab_gated: bool | None = None
# 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()
# 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()
automation_scroll.setWidget(self.automation_progress_panel)
automation_scroll.setWidgetResizable(True)
automation_scroll.setFrameShape(QFrame.Shape.NoFrame)
# One "Information" dock with Automation progress + Console log tabs,
# mirroring the Sample List dock's Dewar/Auxiliary layout — replaces
# the two tabified docks whose switcher tabs sat at the window bottom.
self.log_panel = LogPanel()
self.log_panel.attach_logger("")
self.log_panel.attach_logger("aareGUI")
self.log_panel.reveal_requested.connect(self._reveal_console_log)
self.information_tabs = QTabWidget()
self.information_tabs.addTab(automation_scroll, "Automation progress")
self.information_tabs.addTab(self.log_panel, "Console log")
# Same wrapper trick as the Sample List dock: QTabWidget ignores its
# own contents margins for the tab bar, so the inset lives one level up.
information_wrap = QWidget()
information_wrap_layout = QVBoxLayout(information_wrap)
information_wrap_layout.setContentsMargins(DOCK_CONTENT_LEFT_PAD, 0, 0, 0)
information_wrap_layout.setSpacing(0)
information_wrap_layout.addWidget(self.information_tabs)
self.information_dock = QDockWidget("Information", self)
self.information_dock.setObjectName("information_dock")
self.information_dock.setWidget(information_wrap)
self.information_dock.setAllowedAreas(Qt.DockWidgetArea.BottomDockWidgetArea)
self.addDockWidget(Qt.DockWidgetArea.BottomDockWidgetArea, self.information_dock)
self.information_dock.setFeatures(
QDockWidget.DockWidgetFeature.DockWidgetMovable
| QDockWidget.DockWidgetFeature.DockWidgetClosable
)
# Same title-bar icons (⤢ pop-out + ✕) as Sample List.
self._information_popout: PopoutWindow | None = None
self.information_dock.setTitleBarWidget(
DockTitleBar(self.information_dock, self._open_information_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.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.information_dock], [240, 240], Qt.Orientation.Vertical
)
# Equal oversized requests -> Qt distributes proportionally = 50/50.
self.resizeDocks(
[self.tell_samples_dock, self.information_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)
# Created after the first _apply_theme, so hand it the theme directly.
self.status_bar.set_theme(self._theme_mode)
self.setStatusBar(self.status_bar)
self.daq = DAQWorker(base_url=self._base_url, token=self._token)
self.installEventFilter(self)
# These share MainWindow.eventFilter, which reads sample_lists_tabs
# and left_column_tabs on every event — so they must install only
# HERE, after every attribute the filter touches exists. Installing
# them at widget-construction time made each early event raise
# AttributeError inside the filter ("Error calling Python override
# of QObject::eventFilter()" spam that poisoned CI teardowns).
self.left_column_tabs.tabBar().installEventFilter(self)
for locked_banner in self._locked_beamline_banners:
locked_banner.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)
if hasattr(self, "monochromator_panel"):
self.daq.steer_beam_available.connect(
self.monochromator_panel.update_beam_steering_available
)
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.sample_camera.open_full_help.connect(self.show_controls_help)
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)
# Energy row inside Exp. Config. — not staff-gated like the
# Beamline setup copy; the server enforces write permission anyway.
self.data_collection.change_energy.connect(self.daq.change_energy)
if self._decoded_token.staff:
self.monochromator_panel.mono_pitch_scan.connect(self.daq.mono_pitch_scan)
self.monochromator_panel.move_beam_to_box.connect(self.daq.steer_beam)
self.monochromator_panel.change_energy.connect(self.daq.change_energy)
self.monochromator_panel.open_shutter.connect(self.daq.open_shutter)
self.monochromator_panel.close_shutter.connect(self.daq.close_shutter)
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)
self.prompt_init_state()
def prompt_init_state(self):
self.daq.update_steer_beam_available()
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("Show Console log", self)
self._shortcut_console_log.setShortcut(QKeySequence("Ctrl+Shift+L"))
self._shortcut_console_log.triggered.connect(self._reveal_console_log)
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 _update_queue_buttons_enabled(self) -> None:
queued_view = self.tell_samples.table_model.status_filter == "queued"
running = self.job_list_panel.is_running()
tip = "Active in the Queued view — click the Queued chip above the table"
for button, kind in self._queue_action_buttons:
enabled = queued_view or (kind == "run" and running)
button.setEnabled(enabled)
button.setToolTip("" if enabled else tip)
@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"))
)
# Pop-out chips drive the same shared filter — regate the queue
# buttons from here too.
dewar_panel.status_chips.buttonClicked.connect(
lambda _chip: self._update_queue_buttons_enabled()
)
dewar_panel.set_status_chip(self.tell_samples.table_model.status_filter)
dewar_tab = QWidget()
dewar_tab.setObjectName("dewarTab")
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_information_popout(self) -> None:
if self._information_popout is None:
# Automation 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())
tabs = QTabWidget()
tabs.addTab(panel, "Automation progress")
tabs.addTab(self.log_panel.make_mirror_view(), "Console log")
self._information_popout = PopoutWindow("Information", tabs, parent=self)
self._information_popout.resize(1000, 520)
self._information_popout.show()
self._information_popout.raise_()
self._information_popout.activateWindow()
@Slot()
def _reveal_console_log(self) -> None:
"""Show the Information dock with the Console log tab on top — the
one entry point for 'the user must see the log now' (notifications,
Ctrl+Shift+L)."""
self.information_dock.setVisible(True)
self.information_dock.raise_()
self.information_tabs.setCurrentWidget(self.log_panel)
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())
# Same Queued-view gating as the docked trio.
self._queue_action_buttons += [
(run_button, "run"),
(remove_button, "queue"),
(clear_button, "queue"),
]
self._update_queue_buttons_enabled()
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_admin_tip(self, message: str) -> None:
# "Not admin" denials are passive red tips, not QMessageBoxes:
# nothing to click away. Styled at show time so it always matches
# the theme.
self._admin_tip_anchor.setStyleSheet(admin_tip_qss(self._theme_mode))
QToolTip.showText(QCursor.pos(), message, self._admin_tip_anchor)
def _show_reference_tools_staff_only_popup(self) -> None:
self._show_admin_tip(
"The Auxiliary puck (reference tools) view is available to staff accounts only."
)
def _show_beamline_staff_only_popup(self) -> None:
self._show_admin_tip(
"Beamline configuration is available to staff accounts only.\n"
"Log in with a staff account or contact your local contact "
"to configure the beamline."
)
def _show_beamline_tab_gated_popup(self) -> None:
# The tab is only ever disabled by the pgroup gate; staff hitting it
# are running a visitor's pgroup, users simply lack the rights.
if self._decoded_token.staff:
QMessageBox.information(
self,
"Beamline tab disabled",
"The active pgroup is not one of yours, so the Beamline tab "
"is disabled while running a visitor's experiment. Switch "
"back to your own pgroup to configure the beamline.",
)
else:
self._show_beamline_staff_only_popup()
@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)
)
# Legend defaults off — it covers the camera image; users opt in and
# the choice persists via QSettings.
show_overlay_legend = bool(settings.value("samcam/show_overlay_legend", False, 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")
# The camera overlays show the reason too, not just the boolean.
for cam in (
self.sample_camera,
getattr(self, "compact_sample_camera", None),
getattr(self, "portrait_sample_camera", None),
):
if cam is not None:
cam.set_camera_error_message(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.information_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.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",
"information_dock",
"face_panel_dock",
"fluor_panel_dock",
"smargon_trace_dock",
"target_stability_dock",
"prediction_metrics_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", False)
)
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.information_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)
@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()
assert isinstance(app, QApplication) # palette() lives on QApplication
if not hasattr(self, "_default_palette"):
self._default_palette = app.palette()
if self._theme_mode == THEME_SUNSET:
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)
# Status bar flags likewise; it is created after the first
# _apply_theme call in __init__, hence the guard.
if hasattr(self, "status_bar"):
self.status_bar.set_theme(self._theme_mode)
# Shutter flag in Beamline setup — staff-only panel, hence the guard.
if hasattr(self, "monochromator_panel"):
self.monochromator_panel.set_theme(self._theme_mode)
self.sample_camera.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, QByteArray(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")
# str() wrap: settings.value is typed object even with type=str.
saved = str(settings.value("appearance/theme", THEME_SUNRISE, type=str))
# Migrate the pre-rename tokens so a saved theme survives the value
# change (styles.py: "original"->"sunrise", "portrait"->"sunset").
saved = {"original": THEME_SUNRISE, "portrait": THEME_SUNSET}.get(saved, saved)
self._theme_mode = saved
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_SUNRISE
self._apply_theme()
@Slot()
def use_portrait_theme(self) -> None:
self._theme_mode = THEME_SUNSET
self._apply_theme()
@Slot()
def use_bluebird_theme(self) -> None:
self._theme_mode = THEME_BLUEBIRD
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)
# Prototype views live in the View menu (below the themes), not on the
# menubar; only their in-progress labels changed, the internal
# action/slot names stay.
self._enter_automation_view_action = QAction("Automation View (prototyping)", self)
self._enter_automation_view_action.setShortcut(QKeySequence("Ctrl+5"))
self._enter_automation_view_action.triggered.connect(self.enter_compact_automation_view)
# Stays top-level: it only shows while INSIDE the automation view,
# where the way back must not hide in a menu.
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("Playlist Mode (work in progress)", self)
self._portrait_mode_action.setShortcut(QKeySequence("Ctrl+6"))
self._portrait_mode_action.triggered.connect(self.enter_portrait_mode)
view_menu = menu_bar.addMenu("View")
self._theme_action_group = QActionGroup(self)
self._theme_action_group.setExclusive(True)
self._use_legacy_theme_action = QAction("Sunrise Theme", self)
self._use_legacy_theme_action.setCheckable(True)
self._use_legacy_theme_action.setChecked(self._theme_mode == THEME_SUNRISE)
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("Sunset Theme (work in progress)", self)
self._use_portrait_theme_action.setCheckable(True)
self._use_portrait_theme_action.setChecked(self._theme_mode == THEME_SUNSET)
self._use_portrait_theme_action.triggered.connect(self.use_portrait_theme)
self._theme_action_group.addAction(self._use_portrait_theme_action)
self._use_bluebird_theme_action = QAction("Bluebird Theme (default)", self)
self._use_bluebird_theme_action.setCheckable(True)
self._use_bluebird_theme_action.setChecked(self._theme_mode == THEME_BLUEBIRD)
self._use_bluebird_theme_action.triggered.connect(self.use_bluebird_theme)
self._theme_action_group.addAction(self._use_bluebird_theme_action)
view_menu.addAction(self._use_legacy_theme_action)
view_menu.addAction(self._use_bluebird_theme_action)
view_menu.addAction(self._use_portrait_theme_action)
view_menu.addSeparator()
view_menu.addAction(self._portrait_mode_action)
view_menu.addAction(self._enter_automation_view_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_information_action = QAction("Show Information", self)
show_information_action.setCheckable(True)
show_information_action.setChecked(False)
show_information_action.triggered.connect(
lambda checked: self.information_dock.setVisible(checked)
)
self.information_dock.visibilityChanged.connect(show_information_action.setChecked)
view_menu.addAction(show_information_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)
# Tutorial entries hidden 2026-08-10: content is out of date. The
# tutorial machinery stays wired — restore the two QActions here
# (Start Tutorial (Text)/(Interactive)) once the content is refreshed.
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)
# Default look: Information dock open on the Automation progress tab
# (the old automation dock was visible by default, the log hidden).
self.information_dock.setVisible(True)
self.information_tabs.setCurrentIndex(0)
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_panel.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_panel.show_waiting_notification(title=title, message=message)
def _clear_runtime_notification(self) -> None:
self.log_panel.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_CENTER, 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 grayed. 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)
# Watch-only shows ONLY the camera stream: every other operating
# surface is hidden outright (not just grayed), and regaining the
# baton restores exactly the visibility each one had before. The
# status bar and its SESSION VACANT badge stay — they are the way
# back in. To later hide the camera streams as well, add
# self.video_tab to this list.
# findChildren instead of a hand list: hand-listing missed docks
# (Console Log, Fluorescence, ...) and would again for future ones.
# Covers the pop-out mirrors too — they are operating surfaces.
hide_in_watch_mode = [
self.left_column_tabs,
self.beamline,
self.beamline_state_panel,
*self.findChildren(PopoutWindow),
]
if owned:
for widget, was_visible in self._pre_watch_visibility or []:
widget.setVisible(was_visible)
# Docks restore via restoreState, not per-dock setVisible: a dock
# TABBED BEHIND another is isHidden() at snapshot time, so a
# visibility snapshot mistakes it for user-closed and loses it
# (the vanished Sample List). restoreState also brings back
# tab order, the active tab and dock sizes. Needs objectNames
# on every dock.
state = self._pre_watch_dock_state
if state is not None:
self.restoreState(state)
else:
# not isHidden(), NOT isVisible(): the vacant gate first runs
# before the window is shown, where isVisible() is False for
# everything and the restore would then "restore" all-hidden.
self._pre_watch_dock_state = self.saveState()
self._pre_watch_visibility = [(w, not w.isHidden()) for w in hide_in_watch_mode]
for widget in hide_in_watch_mode + self.findChildren(QDockWidget):
widget.hide()
# Alignment/maintenance states mount reference pins, so the sample dock
# defaults to the Auxiliary puck there; every other state defaults back
# to the Dewar list.
_AUX_PUCK_STATES = frozenset(
{
BeamlineStateEnum.Maintenance,
BeamlineStateEnum.BeamLocation,
BeamlineStateEnum.BeamstopAlignment,
BeamlineStateEnum.FluxMeasurement,
}
)
def _apply_default_sample_tab(self, state: BeamlineStateEnum | None) -> None:
# Index 1 = Auxiliary puck; disabled for non-staff, never force it.
if state in self._AUX_PUCK_STATES and self.sample_lists_tabs.isTabEnabled(1):
self.sample_lists_tabs.setCurrentIndex(1)
else:
self.sample_lists_tabs.setCurrentIndex(0)
def _apply_pgroup_gate(self, pgroup: str | None) -> None:
# Switching to a pgroup outside the token's own (admin) pgroups means
# running a visitor's experiment: the Beamline tab is then off-limits,
# and the Experiment tab comes up with Dataset path + Exp. Config.
# open. Applied only on gate TRANSITIONS so a manual tab choice
# survives the 1 Hz status ticks.
gated = pgroup is not None and pgroup not in (self._decoded_token.pgroups or [])
if gated == self._beamline_tab_gated:
return
self._beamline_tab_gated = gated
self.left_column_tabs.setTabEnabled(0, not gated)
self.left_column_tabs.setTabToolTip(
0, "Only accessible for beamline scientists" if gated else ""
)
if gated:
self.left_column_tabs.setCurrentIndex(1)
# persist=False: programmatic open must not overwrite the user's
# saved per-panel collapse choice (same rule as the session gate).
for banner in self.data_collection.findChildren(TitleLabel):
if banner.text() in ("Dataset path", "Experiment configuration"):
banner.set_collapsed(False, 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))
self._apply_pgroup_gate(getattr(getattr(s, "session", None), "current_pgroup", None))
# No login-default state: the GUI adopts whatever state the server
# reports and never posts a transition on startup — a state POST can
# move motors, and merely logging in must not move hardware.
# Default tab only on state TRANSITIONS — a manual tab choice
# survives while the state stays put.
new_beamline_state = getattr(s, "state", None)
if new_beamline_state != self._last_beamline_state:
self._last_beamline_state = new_beamline_state
self._apply_default_sample_tab(new_beamline_state)
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)
# Sample alignment busy is the alignment itself (same exception as
# the motion watch below) — no BEAMLINE BUSY curtain on any view.
busy_style = build_busy_overlay_style(
is_busy=bool(s.busy) and s.state != BeamlineStateEnum.SampleAlignment,
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()
# Only robot-scale motion flips to the Beamline combined view: state
# transitions (the server parks in Moving while driving motors) and
# the robot station. Busy alone no longer triggers it — Sample
# alignment, Beam location, Beamstop alignment, Flux measurement
# etc. run busy while the user watches the sample camera itself.
# Edge-triggered so a manual tab choice survives between transitions.
moving = s.state in (BeamlineStateEnum.Moving, BeamlineStateEnum.RobotSampleExchange)
if moving and not self._watching_motion:
self._watching_motion = True
self.video_tab.setCurrentWidget(self.beamline_combined_panel)
elif not moving and self._watching_motion:
self._watching_motion = 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=4000)
else:
self.alert_banner.show_message(
"Request declined or canceled", False, auto_clear_ms=4000
)
# 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=4000)
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.canceled_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.canceled_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=4000)
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=4000)
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=4000)
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.canceled_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=4000)
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("information", self.information_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))
# New key: the old "log" flag described a dock that defaulted hidden;
# the merged Information dock defaults visible, so old values would
# wrongly hide it.
if settings.contains("information"):
# bool() wrap: settings.value is typed object even with type=bool.
self.information_dock.setVisible(bool(settings.value("information", True, 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)
# Heal a poisoned layout: a state saved while watch-only mode had
# everything folded away comes back as "every dock hidden" — a user
# hides docks selectively, only the watch-mode fold hides ALL of
# them. Fall back to the default layout instead of resurrecting it
# (the vanished-Sample-List-on-restart bug).
docks = self.findChildren(QDockWidget)
if docks and all(d.isHidden() for d in docks) and self._default_window_state is not None:
self.restoreState(self._default_window_state)
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 self._default_dock_split_done:
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.information_dock], [240, 240], Qt.Orientation.Vertical
)
self.resizeDocks(
[self.tell_samples_dock, self.information_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:
# Closing while watch-only would persist the all-hidden fold and
# poison every future start — put the pre-watch layout back first.
if self._session_operations_enabled is False and self._pre_watch_dock_state is not None:
self.restoreState(self._pre_watch_dock_state)
# 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 self._cleanup_done:
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 not isinstance(dock, QDockWidget):
return
# Floating = top-level: a transparent top-level renders black on the
# container's non-composited X11, so the QDockWidget[floating="true"]
# QSS rule gives it an opaque face; docked it goes transparent again
# so the main-window gradient stays continuous.
dock.setProperty("floating", floating)
dock.style().unpolish(dock)
dock.style().polish(dock)
dock.update()
if floating:
# 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 _set_separator_hint(self, on: bool) -> None:
if self.property("separatorHint") == on:
return
self.setProperty("separatorHint", on)
# Property selectors are only re-evaluated on repolish.
self.style().unpolish(self)
self.style().polish(self)
self.update()
def event(self, event):
# Separator press/drag never reaches mousePressEvent — QMainWindow
# eats it inside event() to start the separator move — so the hint
# gate must sit here, before super() dispatches.
t = event.type()
if t in (QEvent.Type.HoverEnter, QEvent.Type.HoverMove):
# Any move restarts the 1s countdown ("hover and rest"); during a
# drag (button held) the hint stays on instead.
if QApplication.mouseButtons() == Qt.MouseButton.NoButton:
self._set_separator_hint(False)
self._separator_hint_timer.start()
elif t == QEvent.Type.HoverLeave:
self._separator_hint_timer.stop()
self._set_separator_hint(False)
elif t == QEvent.Type.MouseButtonPress:
# Press shows the line immediately, no 1s wait.
self._separator_hint_timer.stop()
self._set_separator_hint(True)
elif t == QEvent.Type.MouseButtonRelease:
self._set_separator_hint(False)
self._separator_hint_timer.start()
return super().event(event)
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
# No try/except: every attribute this touches exists before the first
# install, and the one risky call (backend report) guards itself in
# _refresh_idle_activity.
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()
# getattr defaults: this filter also runs for events delivered while
# __init__ is still building (or teardown is tearing down) the very
# widgets it inspects — a raise here spams every event and breaks
# widget cleanup, so missing attributes must mean "not my click".
if event.type() == QEvent.Type.MouseButtonPress:
# Non-staff click on the grayed-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_tabs = getattr(self, "sample_lists_tabs", None)
sample_tab_bar = sample_tabs.tabBar() if sample_tabs is not None else None
if (
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
# Non-staff click on a locked Beamline banner: same treatment as
# the Auxiliary-puck tab — explain the gate, don't eat the click.
if obj in getattr(self, "_locked_beamline_banners", ()):
self._show_beamline_staff_only_popup()
return True
# Click on the pgroup-gated (disabled) Beamline tab: tabAt() is
# geometric, so it still sees the disabled tab under the cursor.
left_tabs = getattr(self, "left_column_tabs", None)
if left_tabs is not None:
left_tab_bar = left_tabs.tabBar()
if (
left_tab_bar is not None
and obj is left_tab_bar
and left_tab_bar.tabAt(event.position().toPoint()) == 0
and not left_tabs.isTabEnabled(0)
):
self._show_beamline_tab_gated_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()