appleb_mandClaude Opus 4.8
5b833bf9eb
- camera: ML detection boxes+polygons (per-class colours), state-coloured beam centre marker, focus readout - camera controls in tab strip: gain/exposure (samcam_settings), autofocus, screenshot-to-DB - Gonio/Beamline tabs wired to axis IP video via VideoThread - precondition gate (ring/shutter/hutch + 1h snooze) on scans; hutch-PSS mount block; sample-missing modal Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
648 lines
27 KiB
Python
648 lines
27 KiB
Python
"""Main window for the redesigned AareGUI.
|
|
|
|
Assembles the shared chrome (top bar / status bar) and a QStackedWidget body
|
|
(Manual / Automation), then wires everything to the existing DAQWorker and the
|
|
SAMCAM PredictionSubscriber. This is the only module that knows about the
|
|
backend; the views speak in intent signals and request objects.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from PySide6.QtCore import QTimer, Slot
|
|
from PySide6.QtWidgets import (
|
|
QMessageBox,
|
|
QStackedWidget,
|
|
QVBoxLayout,
|
|
QWidget,
|
|
)
|
|
|
|
from aare.common.beamline import cfg_get, mx_beamline
|
|
from aare.common.logger_config import setup_logger
|
|
from aare.gui.new_gui.automation_view import AutomationView
|
|
from aare.gui.new_gui.manual_view import ManualView
|
|
from aare.gui.new_gui.state import AppState
|
|
from aare.gui.new_gui.status_bar import StatusBar
|
|
from aare.gui.new_gui.theme import LIGHT, build_qss
|
|
from aare.gui.new_gui.top_bar import TopBar
|
|
from aare.gui.new_gui.widgets.alert_banner import AlertBanner
|
|
from aare.gui.new_gui.widgets import preconditions
|
|
|
|
logger = setup_logger("aareGUI.new")
|
|
|
|
_BEAMLINE_SUBTITLE = {"X06DA": "PXIII", "X10SA": "PXII", "X06SA": "PXI"}
|
|
|
|
|
|
def _collect_defaults() -> dict:
|
|
"""Scan defaults from the beamline YAML (with safe fall-backs)."""
|
|
rot = "data_collection_settings.default_rotation_settings"
|
|
ras = "data_collection_settings.default_raster_scan_settings"
|
|
return {
|
|
"exp_time_s": cfg_get(f"{rot}.exp_time_s", 0.02),
|
|
"transmission": cfg_get(f"{rot}.transmission", 1.0),
|
|
"dtz": cfg_get(f"{rot}.dtz", 200.0),
|
|
"start_omega_deg": cfg_get(f"{rot}.start_omega_deg", 0.0),
|
|
"increment_omega_deg": cfg_get(f"{rot}.increment_omega_deg", 0.1),
|
|
"steps": cfg_get(f"{rot}.steps", 1800),
|
|
"rotation_exp_time_s": cfg_get(f"{rot}.exp_time_s", 0.02),
|
|
"rotation_dtz": cfg_get(f"{rot}.dtz", 200.0),
|
|
"raster_exp_time_s": cfg_get(f"{ras}.exp_time_s", 0.02),
|
|
"raster_dtz": cfg_get(f"{ras}.dtz", 200.0),
|
|
}
|
|
|
|
|
|
class MainWindow(QWidget):
|
|
def __init__(self, base_url, token, pred_zmq_addr=None,
|
|
gonio_cam_addr=None, beamline_cam_addr=None, gonio_cam_id=1,
|
|
parent=None):
|
|
super().__init__(parent)
|
|
self._base_url = base_url
|
|
self._token = token
|
|
self._gonio_cam_addr = gonio_cam_addr
|
|
self._beamline_cam_addr = beamline_cam_addr
|
|
self._gonio_cam_id = gonio_cam_id
|
|
self._palette = LIGHT
|
|
self._defaults = _collect_defaults()
|
|
|
|
# automation run bookkeeping
|
|
self._auto_running: list = []
|
|
self._auto_index = 0
|
|
self._auto_paused = False
|
|
|
|
# baton / session bookkeeping
|
|
self._staff = False
|
|
self._baton = None
|
|
self._pending_dialog = None
|
|
self._incoming_dialog = None
|
|
self._baton_poll = QTimer(self)
|
|
self._baton_poll.setInterval(1000)
|
|
self._baton_poll.timeout.connect(self._poll_baton_timeout)
|
|
|
|
self.setObjectName("AppRoot")
|
|
self.setStyleSheet(build_qss(self._palette))
|
|
self.setWindowTitle("AareGUI")
|
|
self.resize(1400, 860)
|
|
|
|
beamline = str(mx_beamline().name if hasattr(mx_beamline(), "name")
|
|
else mx_beamline())
|
|
subtitle = _BEAMLINE_SUBTITLE.get(beamline, "")
|
|
bl_label = f"{beamline} · {subtitle}" if subtitle else beamline
|
|
|
|
self.state = AppState(self)
|
|
|
|
# chrome + body
|
|
self.top_bar = TopBar(self._palette, bl_label)
|
|
self.alert = AlertBanner(self._palette)
|
|
self.status_bar = StatusBar(self._palette)
|
|
self.body = QStackedWidget()
|
|
self.manual = ManualView(self.state, self._palette, self._defaults)
|
|
self.automation = AutomationView(self.state, self._palette)
|
|
self.body.addWidget(self.manual)
|
|
self.body.addWidget(self.automation)
|
|
|
|
layout = QVBoxLayout(self)
|
|
layout.setContentsMargins(0, 0, 0, 0)
|
|
layout.setSpacing(0)
|
|
layout.addWidget(self.top_bar)
|
|
layout.addWidget(self.alert)
|
|
layout.addWidget(self.body, 1)
|
|
layout.addWidget(self.status_bar)
|
|
|
|
# backend
|
|
self._build_backend(pred_zmq_addr)
|
|
self._wire()
|
|
|
|
# ------------------------------------------------------------- backend
|
|
def _build_backend(self, pred_zmq_addr) -> None:
|
|
from aare.gui.threads.daq_worker import DAQWorker
|
|
|
|
self.daq = DAQWorker(base_url=self._base_url, token=self._token)
|
|
|
|
self.camera_thread = None
|
|
if pred_zmq_addr:
|
|
try:
|
|
from aare.gui.threads.prediction_subscriber import PredictionSubscriber
|
|
|
|
self.camera_thread = PredictionSubscriber(
|
|
pred_zmq_url=pred_zmq_addr, topic=b""
|
|
)
|
|
self.camera_thread.start()
|
|
except Exception as exc: # pragma: no cover - hardware/runtime path
|
|
logger.warning("Could not start camera stream: %s", exc)
|
|
self.camera_thread = None
|
|
|
|
# axis IP video for Gonio / Beamline tabs
|
|
self.gonio_thread = self._start_video(
|
|
self._gonio_cam_addr, self.manual.gonio_view, camera=self._gonio_cam_id)
|
|
self.beamline_thread = self._start_video(
|
|
self._beamline_cam_addr, self.manual.beamline_view, camera=1)
|
|
|
|
def _start_video(self, addr, view, camera=1):
|
|
if not addr:
|
|
return None
|
|
try:
|
|
from aare.gui.threads.axis_video_thread import VideoThread
|
|
|
|
thread = VideoThread(ip=addr, camera=camera)
|
|
thread.frame_ready.connect(view.update_frame)
|
|
if hasattr(thread, "error_occurred"):
|
|
thread.error_occurred.connect(view.set_error)
|
|
thread.start()
|
|
return thread
|
|
except Exception as exc: # pragma: no cover - hardware/runtime path
|
|
logger.warning("Could not start video (%s): %s", addr, exc)
|
|
return None
|
|
|
|
# -------------------------------------------------------------- wiring
|
|
def _wire(self) -> None:
|
|
s = self.state
|
|
|
|
# mode toggle <-> state <-> body
|
|
self.top_bar.mode_changed.connect(s.set_mode)
|
|
s.mode_changed.connect(self._on_mode_changed)
|
|
|
|
# live status fan-out
|
|
self.daq.update.connect(self.status_bar.update_daq_status)
|
|
self.daq.update.connect(self.manual.update_daq_status)
|
|
self.daq.update.connect(self._on_status)
|
|
self.daq.baton_status_changed.connect(self.top_bar.update_baton_status)
|
|
self.daq.baton_status_changed.connect(self._on_baton_status)
|
|
|
|
# sample list -> changer + library
|
|
self.daq.spreadsheet.connect(self._on_spreadsheet)
|
|
self.daq.reference_tools.connect(self._on_reference_tools)
|
|
|
|
# baton controls
|
|
self.top_bar.request_baton.connect(self.daq.request_baton)
|
|
self.top_bar.release_baton.connect(self.daq.release_baton)
|
|
self.top_bar.cancel_baton_request.connect(self._on_cancel_baton)
|
|
self.daq.baton_request_result.connect(self._on_baton_request_result)
|
|
self.daq.baton_response_result.connect(self._on_baton_response_result)
|
|
if hasattr(self.daq, "baton_incoming_request"):
|
|
self.daq.baton_incoming_request.connect(self._on_baton_incoming)
|
|
if hasattr(self.daq, "baton_timeout_checked"):
|
|
self.daq.baton_timeout_checked.connect(self._on_baton_timeout)
|
|
|
|
# beamline state switching (clickable LED)
|
|
self.status_bar.state_change_requested.connect(self._on_state_change)
|
|
|
|
# manual intents -> DAQ
|
|
self.manual.center_loop_requested.connect(self.daq.center_loop)
|
|
self.manual.center_at_point_requested.connect(self._on_center_at_point)
|
|
# Scans are precondition-gated (ring/shutter/hutch) at the boundary.
|
|
self.manual.raster_scan_requested.connect(self._do_raster)
|
|
self.manual.rotation_scan_requested.connect(self._do_rotation)
|
|
self.manual.smart_params_requested.connect(self._on_smart_params)
|
|
self.manual.xrf_scan_requested.connect(self._do_xrf)
|
|
self.manual.raster_goto_requested.connect(self.daq.move_smargon)
|
|
self.manual.status_message.connect(self._note)
|
|
|
|
# camera controls
|
|
cc = self.manual.camera_controls
|
|
if hasattr(self.daq, "samcam_settings"):
|
|
cc.samcam_changed.connect(self.daq.samcam_settings)
|
|
if hasattr(self.daq, "autofocus"):
|
|
cc.autofocus_requested.connect(self.daq.autofocus)
|
|
if hasattr(self.daq, "send_screenshot_db"):
|
|
cc.screenshot_requested.connect(self.daq.send_screenshot_db)
|
|
if hasattr(self.daq, "raster_scan_completed"):
|
|
self.daq.raster_scan_completed.connect(self.manual.on_raster_completed)
|
|
|
|
# XRF spectrum view <-> fluorimeter
|
|
spec = self.manual.spectrum
|
|
spec.start_clicked.connect(self.daq.fluorimeter_start)
|
|
spec.stop_clicked.connect(self.daq.fluorimeter_stop)
|
|
spec.snapshot_clicked.connect(self.daq.fluorimeter_request_snapshot)
|
|
if hasattr(self.daq, "fluorimeter_spectrum_update"):
|
|
self.daq.fluorimeter_spectrum_update.connect(self._on_spectrum)
|
|
if hasattr(self.daq, "fluorimeter_update"):
|
|
self.daq.fluorimeter_update.connect(spec.update_live)
|
|
|
|
# state mount intents -> DAQ
|
|
s.mount_requested.connect(self._on_mount_requested)
|
|
s.unmount_requested.connect(self.daq.unmount)
|
|
|
|
# motors -> DAQ
|
|
self.manual.motors.omega_rel.connect(self.daq.set_omega_rel)
|
|
self.manual.motors.zoom_changed.connect(self.daq.zoom)
|
|
self.manual.motors.front_light_changed.connect(self.daq.front_light)
|
|
self.manual.motors.back_light_changed.connect(self.daq.back_light)
|
|
self.manual.motors.smargon_jog.connect(self._on_smargon_jog)
|
|
self.manual.motors.energy_changed.connect(self.daq.change_energy)
|
|
|
|
# automation
|
|
self.automation.start_requested.connect(self._start_automation)
|
|
self.automation.pause_requested.connect(self._on_auto_pause)
|
|
self.automation.stop_requested.connect(self._on_auto_stop)
|
|
self.automation.save_template_requested.connect(
|
|
lambda: self._note("Save as template is not wired yet.")
|
|
)
|
|
self.daq.automated_scan_done.connect(self._on_auto_done)
|
|
if hasattr(self.daq, "automation_progress"):
|
|
self.daq.automation_progress.connect(
|
|
self.automation.progress.update_progress
|
|
)
|
|
|
|
# errors / alerts -> banner
|
|
if hasattr(self.daq, "http_error"):
|
|
self.daq.http_error.connect(lambda msg: self._note(msg, error=True))
|
|
if hasattr(self.daq, "auth_error"):
|
|
self.daq.auth_error.connect(self._on_auth_error)
|
|
if hasattr(self.daq, "sample_missing"):
|
|
self.daq.sample_missing.connect(self._on_sample_missing)
|
|
if hasattr(self.daq, "polled_devices_status"):
|
|
self.daq.polled_devices_status.connect(self._on_device_status)
|
|
if hasattr(self.daq, "detector_error"):
|
|
self.daq.detector_error.connect(self._on_device_status)
|
|
if hasattr(self.daq, "automation_critical_failure"):
|
|
self.daq.automation_critical_failure.connect(
|
|
lambda msg: self._note(msg, error=True))
|
|
if hasattr(self.daq, "manual_collection_critical_failure"):
|
|
self.daq.manual_collection_critical_failure.connect(
|
|
lambda msg: self._note(msg, error=True))
|
|
if hasattr(self.daq, "pss_alarm_changed"):
|
|
self.daq.pss_alarm_changed.connect(
|
|
lambda *a: self.alert.show_message(
|
|
"PSS alarm — check hutch door / interlocks.", "warning"))
|
|
# completion notices
|
|
if hasattr(self.daq, "standard_scan_completed"):
|
|
self.daq.standard_scan_completed.connect(self._on_rotation_completed)
|
|
if hasattr(self.daq, "raster_scan_completed"):
|
|
self.daq.raster_scan_completed.connect(self._on_raster_banner)
|
|
|
|
# camera stream
|
|
if self.camera_thread is not None:
|
|
cam = self.manual.camera
|
|
self.camera_thread.image.connect(cam.update_pixmap)
|
|
if hasattr(self.camera_thread, "fps_measure"):
|
|
self.camera_thread.fps_measure.connect(cam.update_fps)
|
|
if hasattr(self.camera_thread, "target_point"):
|
|
self.camera_thread.target_point.connect(cam.update_target_point)
|
|
if hasattr(self.camera_thread, "prediction"):
|
|
self.camera_thread.prediction.connect(cam.update_predictions)
|
|
if hasattr(self.camera_thread, "focus_measure"):
|
|
self.camera_thread.focus_measure.connect(cam.update_focus)
|
|
if hasattr(self.camera_thread, "enable_focus_measurement"):
|
|
self.camera_thread.enable_focus_measurement(True)
|
|
if hasattr(self.camera_thread, "update_daq_status"):
|
|
self.daq.update.connect(self.camera_thread.update_daq_status)
|
|
|
|
# ------------------------------------------------------------ handlers
|
|
@Slot(str)
|
|
def _on_mode_changed(self, mode: str) -> None:
|
|
self.body.setCurrentWidget(
|
|
self.automation if mode == "automation" else self.manual
|
|
)
|
|
self.top_bar.set_mode(mode)
|
|
|
|
@Slot(object)
|
|
def _on_spreadsheet(self, payload) -> None:
|
|
samples = getattr(payload, "s", payload)
|
|
try:
|
|
samples = list(samples)
|
|
except TypeError:
|
|
samples = []
|
|
self.manual.changer.set_samples(samples)
|
|
self.automation.set_samples(samples)
|
|
|
|
# ---- precondition-gated actions ----
|
|
def _precondition_ok(self) -> bool:
|
|
bl = getattr(self.manual.last_status, "bl", None)
|
|
if bl is None:
|
|
return True
|
|
shutter = getattr(bl, "exp_shutter_open", None)
|
|
if shutter is None:
|
|
shutter = getattr(bl, "shutter_open", None)
|
|
return preconditions.precondition_check(
|
|
self,
|
|
ring_current=getattr(bl, "ring_current_mA", None),
|
|
shutter_open=shutter,
|
|
door_prohibited=getattr(bl, "pss_prohibited", None),
|
|
)
|
|
|
|
def _do_raster(self, req) -> None:
|
|
if self._precondition_ok():
|
|
self.daq.raster_scan(req)
|
|
|
|
def _do_rotation(self, req) -> None:
|
|
if self._precondition_ok():
|
|
self.daq.standard_scan(req)
|
|
|
|
def _do_xrf(self, req) -> None:
|
|
if self._precondition_ok():
|
|
self.daq.fluorimeter_spectrum(req)
|
|
|
|
@Slot(object, bool)
|
|
def _on_mount_requested(self, sample, reference: bool = False) -> None:
|
|
reason = preconditions.hutch_blocks_mount(self.manual.last_status)
|
|
if reason is not None:
|
|
QMessageBox.critical(self, "Mounting blocked", reason)
|
|
self.alert.show_message(reason, "error")
|
|
return
|
|
self.daq.mount(sample, reference)
|
|
|
|
def _on_sample_missing(self, msg: str) -> None:
|
|
self.alert.show_message(msg, "error")
|
|
if not self._auto_running: # don't interrupt automation with a modal
|
|
QMessageBox.warning(self, "No sample", msg)
|
|
|
|
@Slot(object)
|
|
def _on_reference_tools(self, payload) -> None:
|
|
tools = getattr(payload, "s", payload)
|
|
try:
|
|
tools = list(tools)
|
|
except TypeError:
|
|
tools = []
|
|
self.manual.set_reference_tools(tools)
|
|
|
|
# ---- staff + control (guest) state ----
|
|
@Slot(object)
|
|
def _on_status(self, s) -> None:
|
|
session = getattr(s, "session", None)
|
|
staff = bool(getattr(session, "staff", False)) if session else False
|
|
if staff != self._staff:
|
|
self._staff = staff
|
|
self.top_bar.set_staff(staff)
|
|
self.manual.set_staff(staff)
|
|
self._update_control()
|
|
|
|
def _on_baton_status(self, status) -> None:
|
|
self._baton = status
|
|
self._update_control()
|
|
# surface an incoming request even without the dedicated signal
|
|
if getattr(status, "incoming_request", False) and self._incoming_dialog is None:
|
|
req = getattr(status, "pending_request", None)
|
|
who = getattr(req, "requester_username", "Another user") if req else "Another user"
|
|
timeout = getattr(req, "timeout_seconds", 30) if req else 30
|
|
self._show_incoming_dialog(who, timeout)
|
|
|
|
def _update_control(self) -> None:
|
|
"""Grey the camera + show Guest/Vacant overlay unless we hold the baton."""
|
|
b = self._baton
|
|
if b is None:
|
|
self.manual.set_control_state(True)
|
|
return
|
|
if getattr(b, "you_are_holder", False):
|
|
self.manual.set_control_state(True)
|
|
return
|
|
holder = getattr(b, "holder", None)
|
|
if holder is None:
|
|
self.manual.set_control_state(
|
|
False, "VACANT", "Grab the baton to take control of the beamline.")
|
|
return
|
|
username = getattr(holder, "username", None)
|
|
holder_staff = bool(getattr(holder, "is_staff", False))
|
|
if self._staff and username:
|
|
sub = f"{username} has control of the beamline."
|
|
elif holder_staff:
|
|
sub = "Staff has control of the beamline."
|
|
else:
|
|
sub = "Another user has control of the beamline."
|
|
self.manual.set_control_state(False, "GUEST MODE", sub)
|
|
|
|
# ---- baton request / response flow ----
|
|
@Slot(dict)
|
|
def _on_baton_request_result(self, result: dict) -> None:
|
|
if result.get("granted") or result.get("already_holder"):
|
|
self._close_pending_dialog()
|
|
self._note("Baton acquired.")
|
|
elif result.get("error"):
|
|
self._close_pending_dialog()
|
|
self._note(result.get("message", "Baton request failed."), error=True)
|
|
elif result.get("pending") or result.get("queued"):
|
|
timeout = int(result.get("timeout_seconds", 30) or 30)
|
|
holder = ""
|
|
if self._baton is not None and getattr(self._baton, "holder", None):
|
|
if self._staff:
|
|
holder = getattr(self._baton.holder, "username", "") or ""
|
|
self._show_pending_dialog(holder, timeout)
|
|
if result.get("queued"):
|
|
self._pending_dialog.set_queued()
|
|
self._baton_poll.start()
|
|
|
|
@Slot(dict)
|
|
def _on_baton_response_result(self, result: dict) -> None:
|
|
if result.get("accepted"):
|
|
self._note("Control transferred.")
|
|
elif result.get("refused"):
|
|
self._note("Request declined.")
|
|
self._close_incoming_dialog()
|
|
|
|
def _on_baton_incoming(self, payload: dict) -> None:
|
|
if self._incoming_dialog is not None:
|
|
return
|
|
who = payload.get("requester", "Another user")
|
|
timeout = int(payload.get("timeout", 30) or 30)
|
|
self._show_incoming_dialog(who, timeout)
|
|
|
|
@Slot(dict)
|
|
def _on_baton_timeout(self, payload: dict) -> None:
|
|
if self._pending_dialog is None:
|
|
return
|
|
if payload.get("granted"):
|
|
self._close_pending_dialog()
|
|
self._note("Baton acquired.")
|
|
elif payload.get("refused"):
|
|
self._close_pending_dialog()
|
|
self._note("Request refused.", error=True)
|
|
elif payload.get("queued"):
|
|
self._pending_dialog.set_queued()
|
|
else:
|
|
self._pending_dialog.update_remaining(payload.get("remaining_seconds", 0))
|
|
|
|
def _poll_baton_timeout(self) -> None:
|
|
if self._pending_dialog is None:
|
|
self._baton_poll.stop()
|
|
return
|
|
if hasattr(self.daq, "check_baton_timeout"):
|
|
self.daq.check_baton_timeout()
|
|
|
|
def _on_cancel_baton(self) -> None:
|
|
self._close_pending_dialog()
|
|
if hasattr(self.daq, "cancel_baton_request"):
|
|
self.daq.cancel_baton_request()
|
|
|
|
def _show_pending_dialog(self, holder: str, timeout: int) -> None:
|
|
from aare.gui.new_gui.widgets.baton_dialogs import BatonPendingDialog
|
|
|
|
self._close_pending_dialog()
|
|
dlg = BatonPendingDialog(holder, timeout, self._palette, self)
|
|
dlg.cancelled_signal.connect(self._on_cancel_baton)
|
|
self._pending_dialog = dlg
|
|
dlg.show()
|
|
|
|
def _close_pending_dialog(self) -> None:
|
|
self._baton_poll.stop()
|
|
if self._pending_dialog is not None:
|
|
self._pending_dialog.close()
|
|
self._pending_dialog = None
|
|
|
|
def _show_incoming_dialog(self, requester: str, timeout: int) -> None:
|
|
from aare.gui.new_gui.widgets.baton_dialogs import BatonIncomingDialog
|
|
|
|
dlg = BatonIncomingDialog(requester, timeout, self._palette, self)
|
|
dlg.accepted_signal.connect(lambda: self.daq.respond_to_baton_request(True))
|
|
dlg.refused_signal.connect(lambda: self.daq.respond_to_baton_request(False))
|
|
dlg.finished.connect(self._close_incoming_dialog)
|
|
self._incoming_dialog = dlg
|
|
dlg.show()
|
|
|
|
def _close_incoming_dialog(self, *args) -> None:
|
|
if self._incoming_dialog is not None:
|
|
self._incoming_dialog = None
|
|
|
|
# ---- beamline state switching ----
|
|
@Slot(str)
|
|
def _on_state_change(self, key: str) -> None:
|
|
method = getattr(self.daq, key, None)
|
|
if callable(method):
|
|
self._note(f"Switching beamline state: {key}")
|
|
method()
|
|
else:
|
|
self._note(f"No DAQ method for state '{key}'.", error=True)
|
|
|
|
def _on_smargon_jog(self, dx_mm: float, dy_mm: float) -> None:
|
|
"""Relative sample-holder jog in beamline (screen) space.
|
|
|
|
The delta is a beamline-frame offset; ``beamline_to_smargon`` rotates it
|
|
through the current omega/phi/chi into an absolute smargon sh_mm target,
|
|
so the jog stays screen-intuitive at any goniometer angle. Arrow signs:
|
|
←/→ = -x/+x, ↑/↓ = -y/+y (camera y points down). Flip in MotorsPanel if
|
|
the beamline convention differs.
|
|
"""
|
|
from aare.common.coordinate import Coordinate, SmargonCoordinate
|
|
|
|
geom = getattr(self.manual.last_status, "geom", None)
|
|
if geom is None or not hasattr(geom, "beamline_to_smargon"):
|
|
self._note("No geometry yet; cannot jog.", error=True)
|
|
return
|
|
try:
|
|
sh = geom.beamline_to_smargon(Coordinate(x=dx_mm, y=dy_mm, z=0.0))
|
|
except Exception as exc:
|
|
self._note(f"Could not compute jog target: {exc}", error=True)
|
|
return
|
|
self.daq.move_smargon(SmargonCoordinate(sh_mm=sh))
|
|
|
|
def _on_spectrum(self, model) -> None:
|
|
self.manual.spectrum.update_spectrum(model)
|
|
self.manual.show_spectrum()
|
|
|
|
def _on_smart_params(self, sp) -> None:
|
|
"""Automate-parameters (simple strategy): forward seeds, then collect."""
|
|
from aare.gui.new_gui import requests
|
|
|
|
if not self._precondition_ok():
|
|
return
|
|
if hasattr(self.daq, "smart_params"):
|
|
self.daq.smart_params(sp)
|
|
self.daq.standard_scan(requests.rotation_from_simple(sp))
|
|
|
|
def _on_center_at_point(self, img_x: float, img_y: float) -> None:
|
|
"""Click-to-centre: move the clicked camera pixel onto the beam.
|
|
|
|
Mirrors the existing GUI: picture -> sample (beamline) -> smargon.
|
|
"""
|
|
from aare.common.coordinate import Coordinate, SmargonCoordinate
|
|
|
|
geom = getattr(self.manual.last_status, "geom", None)
|
|
if geom is None or not hasattr(geom, "picture_to_sample"):
|
|
self._note("No geometry yet; cannot centre.", error=True)
|
|
return
|
|
try:
|
|
sample = geom.picture_to_sample(Coordinate(x=img_x, y=img_y))
|
|
sh = geom.beamline_to_smargon(sample)
|
|
except Exception as exc:
|
|
self._note(f"Could not compute centre target: {exc}", error=True)
|
|
return
|
|
self.daq.move_smargon(SmargonCoordinate(sh_mm=sh))
|
|
|
|
# ---- automation loop (GUI-local; iterates automated_scan, pausable) ----
|
|
def _start_automation(self) -> None:
|
|
queue = list(self.state.queue)
|
|
if not queue:
|
|
return
|
|
self._auto_running = queue
|
|
self._auto_index = 0
|
|
self._auto_paused = False
|
|
self.automation.set_running(True, paused=False)
|
|
if hasattr(self.daq, "start_automation_progress_stream"):
|
|
self.daq.start_automation_progress_stream()
|
|
self._note(f"Starting automation: {len(queue)} samples queued.")
|
|
self._run_next_auto()
|
|
|
|
def _run_next_auto(self) -> None:
|
|
if self._auto_paused:
|
|
return
|
|
if self._auto_index >= len(self._auto_running):
|
|
self._note("Automation queue complete.")
|
|
self._auto_running = []
|
|
self.automation.set_running(False)
|
|
return
|
|
item = self._auto_running[self._auto_index]
|
|
self.daq.automated_scan(item.sample)
|
|
|
|
@Slot(int, bool, str)
|
|
def _on_auto_done(self, sample_id: int, success: bool, error: str) -> None:
|
|
if not self._auto_running:
|
|
return
|
|
if not success:
|
|
self._note(f"Automation stopped: {error}", error=True)
|
|
self._auto_running = []
|
|
self.automation.set_running(False)
|
|
return
|
|
self._auto_index += 1
|
|
self._run_next_auto() # no-op while paused; resumes on _on_auto_pause
|
|
|
|
def _on_auto_pause(self) -> None:
|
|
if not self._auto_running:
|
|
return
|
|
self._auto_paused = not self._auto_paused
|
|
self.automation.set_running(True, paused=self._auto_paused)
|
|
if self._auto_paused:
|
|
self._note("Automation will pause after the current sample.")
|
|
else:
|
|
self._note("Resuming automation.")
|
|
self._run_next_auto()
|
|
|
|
def _on_auto_stop(self) -> None:
|
|
self._auto_running = []
|
|
self._auto_paused = False
|
|
self.automation.set_running(False)
|
|
self._note("Automation stopped.")
|
|
|
|
def _on_auth_error(self) -> None:
|
|
self._note("Lost the baton / authentication error.", error=True)
|
|
|
|
def _note(self, message: str, error: bool = False, level: str | None = None) -> None:
|
|
if error:
|
|
logger.error(message)
|
|
else:
|
|
logger.info(message)
|
|
self.alert.show_message(message, level or ("error" if error else "info"))
|
|
|
|
def _on_device_status(self, message: str, is_error: bool) -> None:
|
|
if is_error:
|
|
self.alert.show_message(message, "error")
|
|
|
|
def _on_rotation_completed(self, _completed) -> None:
|
|
self.alert.show_message("Rotation collection complete.", "success")
|
|
|
|
def _on_raster_banner(self, _completed) -> None:
|
|
self.alert.show_message("Raster scan complete.", "success")
|
|
|
|
# --------------------------------------------------------------- close
|
|
def closeEvent(self, event): # noqa: N802
|
|
try:
|
|
if hasattr(self.daq, "cleanup"):
|
|
self.daq.cleanup()
|
|
except Exception:
|
|
pass
|
|
for thread in (getattr(self, "camera_thread", None),
|
|
getattr(self, "gonio_thread", None),
|
|
getattr(self, "beamline_thread", None)):
|
|
try:
|
|
if thread is not None and hasattr(thread, "stop"):
|
|
thread.stop()
|
|
except Exception:
|
|
pass
|
|
super().closeEvent(event)
|