local_contact panel: added new tabs, connected back end tools such as reinitialising bec macros and devices, opening beamline specific web helpers, restarting backend connections tod evices such as jfjoch, tell, aerotech and smargon.

This commit is contained in:
appleb_m
2026-06-05 14:10:34 +02:00
parent d9c9f4db14
commit c3f16bcdf8
11 changed files with 1454 additions and 165 deletions
+87
View File
@@ -1015,6 +1015,93 @@ class BeamlineConfig:
def increment_failed_mount_count(self) -> int:
return int(self.__client.incr(f"{self.__bl}:failed_mount_count"))
def _runtime_sim_key(self, name: str) -> str:
return f"{self.__bl}:runtime:simulate:{name}"
def get_runtime_simulated(self, name: str, default: bool = False) -> bool:
raw = self.__client.get(self._runtime_sim_key(name))
if raw is None:
return default
return str(raw).strip().lower() in {"1", "true", "yes", "on"}
def set_runtime_simulated(self, name: str, enabled: bool) -> None:
key = self._runtime_sim_key(name)
if enabled:
self.__client.set(key, "1")
else:
self.__client.delete(key)
@property
def simulate_bec(self) -> bool:
return self.get_runtime_simulated("bec", default=False)
@simulate_bec.setter
def simulate_bec(self, enabled: bool) -> None:
self.set_runtime_simulated("bec", enabled)
@property
def simulate_tell(self) -> bool:
return self.get_runtime_simulated("tell", default=False)
@simulate_tell.setter
def simulate_tell(self, enabled: bool) -> None:
self.set_runtime_simulated("tell", enabled)
@property
def simulate_aerotech(self) -> bool:
return self.get_runtime_simulated("aerotech", default=False)
@simulate_aerotech.setter
def simulate_aerotech(self, enabled: bool) -> None:
self.set_runtime_simulated("aerotech", enabled)
@property
def simulate_smargon(self) -> bool:
return self.get_runtime_simulated("smargon", default=False)
@simulate_smargon.setter
def simulate_smargon(self, enabled: bool) -> None:
self.set_runtime_simulated("smargon", enabled)
@property
def runtime_simulation_state(self) -> dict[str, bool]:
return {
"bec": self.simulate_bec,
"detector": bool(self.simulated_detector),
"tell": self.simulate_tell,
"aerotech": self.simulate_aerotech,
"smargon": self.simulate_smargon,
}
@property
def local_contact_links(self) -> dict[str, str | None]:
detector_frontend = None
smargon_frontend = cfg_get(
"daq.hardware.smargon_url",
f"http://{self.__mxb.name.lower()}-smargopolo.psi.ch:8080/",
)
aerotech_frontend = cfg_get(
"daq.hardware.aerotech_url",
f"http://mx-{self.__mxb.name.lower()}-queue-01.psi.ch:5234/",
)
tell_hint = "Please check TELL status via Remmina / VNC."
if self.__mxb is MXBeamline.X06DA:
detector_frontend = "http://sls-gpu-001:8080/frontend"
elif self.__mxb is MXBeamline.X10SA:
detector_frontend = "http://sls-gpu-002:8080/frontend"
elif self.__mxb is MXBeamline.SIMULATED:
detector_frontend = None
smargon_frontend = None
aerotech_frontend = None
return {
"aerotech": aerotech_frontend,
"detector": detector_frontend,
"smargon": smargon_frontend,
"tell_hint": tell_hint,
}
if __name__ == "__main__":
from aare.common.beamline import mx_beamline
cfg = BeamlineConfig(bl=mx_beamline())
+144
View File
@@ -218,6 +218,7 @@ class AareDAQ:
self.__mlbox = MlBox(bl)
self.__jfjoch = JFJochWrapper(bl)
self.__bl = bl.value.upper()
self._beamline = bl
self.__aare = AareWrapper(bl)
self.__saved_box = None
self._smargon_trace_path = Path("/sls/mx/applications/logs") / "smargon_trace.csv"
@@ -242,6 +243,122 @@ class AareDAQ:
self._create_mounting_service().reset_mount_failure_counter("DAQ startup")
def get_runtime_simulation_state(self) -> dict[str, bool]:
return self.__cfg.runtime_simulation_state
def get_local_contact_links(self) -> dict[str, str | None]:
return self.__cfg.local_contact_links
def get_local_contact_device_state(self) -> dict[str, dict[str, str | bool | None]]:
status = self.status
sim = self.get_runtime_simulation_state()
return {
"bec": {
"mode": "simulated" if sim.get("bec") else "live",
"error": None,
},
"detector": {
"mode": "simulated" if sim.get("detector") else "live",
"error": None,
},
"tell": {
"mode": "simulated" if sim.get("tell") else "live",
"error": getattr(status, "tell_error", None),
},
"aerotech": {
"mode": "simulated" if sim.get("aerotech") else "live",
"error": getattr(status, "aerotech_error", None),
},
"smargon": {
"mode": "simulated" if sim.get("smargon") else "live",
"error": getattr(status, "smargon_error", None),
},
}
def restart_bec_worker(self) -> dict[str, object]:
self.__cfg.try_set_busy(timeout=360)
try:
self.__devs.restart_bec_worker(simulated=self.__cfg.simulate_bec)
return {
"ok": True,
"device": "bec",
"simulated": self.__cfg.simulate_bec,
}
finally:
self.__cfg.state_busy = False
def restart_detector(self) -> dict[str, object]:
self.__cfg.try_set_busy(timeout=360)
try:
beamline = MXBeamline.SIMULATED if self.__cfg.simulated_detector else self._beamline
logger.info(f"Restarting JFJoch wrapper with simulated={self.__cfg.simulated_detector}")
self.__jfjoch = JFJochWrapper(beamline)
return {
"ok": True,
"device": "detector",
"simulated": bool(self.__cfg.simulated_detector),
}
finally:
self.__cfg.state_busy = False
def restart_tell(self) -> dict[str, object]:
self.__cfg.try_set_busy(timeout=360)
try:
self.__devs.restart_tell(simulated=self.__cfg.simulate_tell)
return {
"ok": True,
"device": "tell",
"simulated": self.__cfg.simulate_tell,
}
finally:
self.__cfg.state_busy = False
def restart_aerotech(self) -> dict[str, object]:
self.__cfg.try_set_busy(timeout=360)
try:
self.__devs.restart_aerotech(simulated=self.__cfg.simulate_aerotech)
return {
"ok": True,
"device": "aerotech",
"simulated": self.__cfg.simulate_aerotech,
}
finally:
self.__cfg.state_busy = False
def restart_smargon(self) -> dict[str, object]:
self.__cfg.try_set_busy(timeout=360)
try:
self.__devs.restart_smargon(simulated=self.__cfg.simulate_smargon)
return {
"ok": True,
"device": "smargon",
"simulated": self.__cfg.simulate_smargon,
}
finally:
self.__cfg.state_busy = False
def set_runtime_simulation(self, device: str, enabled: bool) -> dict[str, object]:
device = str(device).strip().lower()
if device == "bec":
self.__cfg.simulate_bec = enabled
return self.restart_bec_worker()
elif device == "detector":
self.__cfg.simulated_detector = enabled
return self.restart_detector()
elif device == "tell":
self.__cfg.simulate_tell = enabled
return self.restart_tell()
elif device == "aerotech":
self.__cfg.simulate_aerotech = enabled
return self.restart_aerotech()
elif device == "smargon":
self.__cfg.simulate_smargon = enabled
return self.restart_smargon()
raise ValueError("Unknown simulation device. Expected one of: bec, detector, tell, aerotech, smargon")
def _is_hardware_failure(self, error: Exception) -> bool:
return isinstance(
error,
@@ -3004,6 +3121,33 @@ class AareDAQ:
finally:
self.__cfg.state_busy = False
def bec_load_user_macros(self) -> None:
self.__cfg.try_set_busy(timeout=360)
try:
self.__devs.bec_worker.load_user_macros()
finally:
self.__cfg.state_busy = False
def bec_list_all_user_macros(self) -> list[str]:
macros = self.__devs.bec_worker.list_all_user_macros()
if macros is None:
return []
return [str(item) for item in macros]
def bec_list_all_devices(self) -> list[str]:
devices = self.__devs.bec_worker.list_position_devices()
if devices is None:
return []
return [str(item) for item in devices]
def bec_reinitialise_planner_and_position_devices(self, method: str = "auto") -> list[str]:
self.__cfg.try_set_busy(timeout=360)
try:
self.__devs.bec_worker.load_user_macros()
return self.__devs.bec_worker.reinitialise_planner_and_position_devices(method=method)
finally:
self.__cfg.state_busy = False
def fluorimeter_take_spectrum(self, fm: FluorescenceSpectrumParameterModel) -> FluorescenceSpectrumOutputModel:
self.__cfg.try_set_busy(timeout=360)
+27
View File
@@ -27,6 +27,7 @@ logger = setup_logger("aareDAQ")
class BeamlineDevices:
def __init__(self, beamline: MXBeamline):
self._beamline = beamline
self.detector_distance_minimum = 170
BEAMLINE = beamline.value.upper()
self.tell = make_tell_client(beamline)
@@ -111,6 +112,32 @@ class BeamlineDevices:
#self.magnet_position_sensor_readout = PV(f"{BEAMLINE}-ES-DFS:CBOX-REFVAL1")
self.magnet_position_sensor_state = PV(f"{BEAMLINE}-ES-DFS:CBOX-STATE")
def restart_bec_worker(self, simulated: bool = False) -> None:
try:
if getattr(self, "bec_worker", None) is not None and not simulated:
try:
self.bec_worker.shutdown_client()
except Exception as e:
logger.warning(f"Failed to shutdown previous BEC worker cleanly: {e}")
finally:
beamline = MXBeamline.SIMULATED if simulated else self._beamline
logger.info(f"Restarting BEC worker with simulated={simulated}")
self.bec_worker = BECClientWorker(beamline)
def restart_tell(self, simulated: bool = False) -> None:
beamline = MXBeamline.SIMULATED if simulated else self._beamline
logger.info(f"Restarting TELL client with simulated={simulated}")
self.tell = make_tell_client(beamline)
def restart_aerotech(self, simulated: bool = False) -> None:
beamline = MXBeamline.SIMULATED if simulated else self._beamline
logger.info(f"Restarting Aerotech controller with simulated={simulated}")
self.aerotech = aerotech.AerotechController(beamline)
def restart_smargon(self, simulated: bool = False) -> None:
beamline = MXBeamline.SIMULATED if simulated else self._beamline
logger.info(f"Restarting Smargon controller with simulated={simulated}")
self.__smargon = smargon.Smargon(beamline)
# Transmission
@property
+129 -9
View File
@@ -554,15 +554,62 @@ async def initialise_smargon(token: str = Depends(oauth2_scheme)) -> dict:
"message": "Smargon initialised.",
}
def bec_load_user_macros(self):
self.__cfg.try_set_busy(timeout=360)
try:
self.__devs.bec_worker.load_user_macros()
self.__cfg.state_busy = False
except Exception as e:
self.__cfg.state_busy = False
logger.error(f"Failed to load BEC user macros: {e}")
raise
@app.post("/bec/load_user_macros")
async def bec_load_user_macros(token: str = Depends(oauth2_scheme)) -> dict:
"""
Load BEC user macros. Staff only.
"""
data = auth.parse_token(token)
auth.check_jwt_staff_only(data)
daq.bec_load_user_macros()
return {
"ok": True,
"message": "BEC user macros loaded.",
}
@app.get("/bec/user_macros")
async def bec_list_all_user_macros(token: str = Depends(oauth2_scheme)) -> list:
"""
List all BEC user macros. Staff only.
"""
data = auth.parse_token(token)
auth.check_jwt_staff_only(data)
return daq.bec_list_all_user_macros()
@app.get("/bec/devices")
async def bec_list_all_devices(token: str = Depends(oauth2_scheme)) -> list:
"""
List all BEC position devices. Staff only.
"""
data = auth.parse_token(token)
auth.check_jwt_staff_only(data)
return daq.bec_list_all_devices()
@app.post("/bec/reinitialise_planner_and_position_devices")
async def bec_reinitialise_planner_and_position_devices(
method: str = "auto",
token: str = Depends(oauth2_scheme),
) -> dict:
"""
Reinitialise BEC planner and position devices. Staff only.
Args:
method:
"auto" - use beamline default.
"beamline" - force init_beamline_environment().
"sample" - force init_se_devices() and planner creation.
token: OAuth2 access token.
"""
data = auth.parse_token(token)
auth.check_jwt_staff_only(data)
position_devices = daq.bec_reinitialise_planner_and_position_devices(method=method)
return {
"ok": True,
"method": method,
"position_devices": position_devices,
"message": "BEC planner and position devices reinitialised.",
}
def initialise_aerotech(self):
self.__cfg.try_set_busy(timeout=360)
@@ -594,6 +641,79 @@ def initialise_detector(self):
logger.error(f"Failed to initialise detector: {e}")
raise
@app.get("/local_contact/simulation_state")
async def local_contact_simulation_state(token: str = Depends(oauth2_scheme)) -> dict:
"""
Return current runtime simulation state for Local Contact tools. Staff only.
"""
data = auth.parse_token(token)
auth.check_jwt_staff_only(data)
return daq.get_runtime_simulation_state()
@app.get("/local_contact/device_state")
async def local_contact_device_state(token: str = Depends(oauth2_scheme)) -> dict:
"""
Return Local Contact device mode/error state. Staff only.
"""
data = auth.parse_token(token)
auth.check_jwt_staff_only(data)
return daq.get_local_contact_device_state()
@app.get("/local_contact/links")
async def local_contact_links(token: str = Depends(oauth2_scheme)) -> dict:
"""
Return Local Contact web control links. Staff only.
"""
data = auth.parse_token(token)
auth.check_jwt_staff_only(data)
return daq.get_local_contact_links()
@app.post("/local_contact/simulate/{device}")
async def local_contact_set_simulation(
device: str,
enabled: bool,
token: str = Depends(oauth2_scheme),
) -> dict:
"""
Enable or disable runtime simulation for a backend device and restart its wrapper. Staff only.
"""
data = auth.parse_token(token)
auth.check_jwt_staff_only(data)
result = daq.set_runtime_simulation(device, enabled)
result["message"] = f"{device} simulation set to {enabled}."
return result
@app.post("/local_contact/restart/{device}")
async def local_contact_restart_device(
device: str,
token: str = Depends(oauth2_scheme),
) -> dict:
"""
Restart a Local Contact backend wrapper. Staff only.
"""
data = auth.parse_token(token)
auth.check_jwt_staff_only(data)
device = str(device).strip().lower()
if device == "bec":
result = daq.restart_bec_worker()
elif device == "detector":
result = daq.restart_detector()
elif device == "tell":
result = daq.restart_tell()
elif device == "aerotech":
result = daq.restart_aerotech()
elif device == "smargon":
result = daq.restart_smargon()
else:
raise HTTPException(
status_code=api_status.HTTP_400_BAD_REQUEST,
detail="Unknown restart device. Expected one of: bec, detector, tell, aerotech, smargon.",
)
result["message"] = f"{device} backend restarted."
return result
@app.post("/beamline/goto_abr_meas_pos")
async def goto_abr_meas_pos(token: str = Depends(oauth2_scheme)):
"""
+96 -16
View File
@@ -1,5 +1,5 @@
from enum import Enum
from typing import List, Optional
from typing import List, Optional, Any
from bec_ipython_client import BECIPythonClient
from bec_ipython_client.signals import OperationMode
@@ -56,7 +56,8 @@ class BeamlineState(str, Enum):
class BECClientWorker:
def __init__(self, beamline:MXBeamline, name:str = "default"):
BEAMLINE = beamline.value.lower()
if beamline is MXBeamline.SIMULATED:
self.beamline = beamline
if self.beamline is MXBeamline.SIMULATED:
self.simulated = True
else:
@@ -74,12 +75,12 @@ class BECClientWorker:
print(self.dev.keys())
self.scans = self.client.scans
self.macros = self.client.macros
self.load_user_macros()
print(self.list_all_macros())
self.__load_user_macros()
print(self.__list_all_macros())
self.helper = FrontendProcedureHelper(self.client.connector)
self.__set_scilog_tags()
try:
if beamline is MXBeamline.X06DA:
if self.beamline is MXBeamline.X06DA:
self.position_devices, self.planner = self.__init_beamline_environment()
else:
self.position_devices = self.__initialise_sample_environment_devices()
@@ -110,10 +111,10 @@ class BECClientWorker:
else:
tags += ["error", "bec"]
try:
alarms = [alarm for alarm in self.client.alarms()]
logger.debug(f"alarms: {alarms}")
if alarms:
bec_alarms = self.client.alarms()
if bec_alarms:
alarms = [alarm for alarm in bec_alarms]
logger.debug(f"alarms: {alarms}")
last_alarm = self.client.show_last_alarm()
message += f"\n\nAlarm: {last_alarm}\n\n"
logger.error(f"last_alarm: {last_alarm}")
@@ -154,10 +155,6 @@ class BECClientWorker:
except Exception as e:
self._raise_bec_error(e, operation="initialise_devices")
def list_position_devices(self):
"""List the position devices available for the BEC worker"""
return list(self.position_devices.keys())
def __get_states(self):
"""Get beamline states and modifiers from beamlien_states.yaml"""
if self.simulated:
@@ -277,6 +274,7 @@ class BECClientWorker:
logger.debug(f"Simulating move to {state.value}")
return True
try:
self.position_devices['bs_z'].set_position(23.8)
self.planner.move_to(state)
if self.planner.is_state(state):
logger.info(f"BEC move_to completed: {state.value} in {time.perf_counter() - start:.2f}s")
@@ -299,14 +297,96 @@ class BECClientWorker:
return BeamlineState.MAINTENANCE
return self.planner.current_state()
def list_position_devices(self):
"""List the position devices available for the BEC worker"""
if self.simulated or self.position_devices is None:
return []
return list(self.position_devices.keys())
def show_all_devices(self):
return self.dev.show_all
def list_all_macros(self):
return self.macros.list_user_macros()
def list_all_user_macros(self) -> list[str] | None:
if self.simulated:
logger.debug("Simulating list_all_user_macros")
return []
try:
self.__list_all_macros()
raw_macros = [name for name, _ in self.client.macros._update_handler.macros.items()]
if raw_macros is None:
logger.warning("BEC returned no user macros; treating as empty list")
return []
return [str(macro) for macro in raw_macros]
except Exception as e:
self._raise_bec_error(e, operation="list_all_user_macros")
def __list_all_macros(self):
result = self.macros.list_user_macros()
if result is None:
return []
return result
def load_user_macros(self):
self.macros.load_all_user_macros()
if self.simulated:
logger.debug("Simulating load_user_macros")
return None
try:
return self.__load_user_macros()
except Exception as e:
self._raise_bec_error(e, operation="load_user_macros")
def __load_user_macros(self):
result = self.macros.load_all_user_macros()
if result is None:
logger.warning("BEC load_all_user_macros returned None")
return result
def reinitialise_planner_and_position_devices(self, method: str = "auto"):
"""
Reinitialise BEC position devices and planner.
Args:
method:
"auto" - use the beamline-specific default.
"beamline" - force init_beamline_environment().
"sample" - force init_se_devices() + planner creation.
Returns:
List of position device names after reinitialisation.
"""
if self.simulated:
logger.debug(f"Simulating reinitialise_planner_and_position_devices(method={method})")
return []
method = str(method or "auto").strip().lower()
try:
if method == "auto":
if self.beamline is MXBeamline.X06DA:
method = "beamline"
else:
method = "sample"
if method == "beamline":
self.position_devices, self.planner = self.__init_beamline_environment()
elif method in {"sample", "sample_environment"}:
self.position_devices = self.__initialise_sample_environment_devices()
self.planner = self.__planner()
else:
raise ValueError(
"Invalid BEC reinitialisation method. "
"Expected one of: auto, beamline, sample."
)
self.__backlight_brightness = self.position_devices['bl_bright']
self.__detector_cover = self.position_devices['det_cov']
logger.info(f"Reinitialised BEC planner and position devices using method={method}")
return self.list_position_devices()
except Exception as e:
self._raise_bec_error(
e,
operation=f"reinitialise_planner_and_position_devices:{method}",
)
def shutdown_client(self):
self.client.shutdown()
+1 -5
View File
@@ -969,10 +969,6 @@ class MainWindow(QMainWindow):
help_menu.addAction(dev_help_action)
if self.__decoded_token.staff:
beamline_recovery_action = QAction("Beamline Recovery", self)
beamline_recovery_action.triggered.connect(self.show_beamline_recovery)
help_menu.addAction(beamline_recovery_action)
local_contact_action = QAction("Local Contact", self)
local_contact_action.triggered.connect(self.show_local_contact)
help_menu.addAction(local_contact_action)
@@ -1237,7 +1233,7 @@ class MainWindow(QMainWindow):
self._beamline_recovery_dialog.raise_()
self._beamline_recovery_dialog.activateWindow()
def show_local_contact(self, tab_name: str = "Recovery") -> None:
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:
+1 -39
View File
@@ -60,18 +60,6 @@ class RecoveryPanel(QWidget):
)
layout.addWidget(self._warning_secondary)
self._status = QLabel("Current status: waiting for DAQ status update…", self)
self._status.setWordWrap(True)
self._status.setStyleSheet(
"QLabel {"
" background: #fafafa;"
" border: 1px solid #d0d0d0;"
" border-radius: 6px;"
" padding: 8px;"
"}"
)
layout.addWidget(self._status)
self._last_action = QLabel("Last action: -", self)
self._last_action.setWordWrap(True)
self._last_action.setStyleSheet(
@@ -184,31 +172,6 @@ class RecoveryPanel(QWidget):
except Exception:
return False
def _status_text(self) -> str:
if self._last_status is None:
return "Current status: waiting for DAQ status update…"
state_name = getattr(self._last_status.state, "name", str(self._last_status.state))
busy = bool(getattr(self._last_status, "busy", False))
sample_mounted = self._sample_appears_mounted()
tell_connected = bool(getattr(self._last_status, "tell_connected", False))
tell_state_text = ""
if getattr(self._last_status, "tell_state", None) is not None:
tell_state = self._last_status.tell_state
tell_state_text = tell_state.activity.value
tell_message = (tell_state.message or "").strip()
if tell_message:
tell_state_text = f"{tell_state_text} ({tell_message})"
return (
f"State: {state_name}\n"
f"Busy: {busy}\n"
f"Sample mounted: {sample_mounted}\n"
f"TELL connected: {tell_connected}\n"
f"TELL state: {tell_state_text}"
)
def _refresh_buttons(self) -> None:
sample_mounted = self._sample_appears_mounted()
beamline_busy = self._beamline_appears_busy()
@@ -239,7 +202,6 @@ class RecoveryPanel(QWidget):
@Slot(DAQStatusModel)
def _set_daq_status(self, s: DAQStatusModel) -> None:
self._last_status = s
self._status.setText(self._status_text())
self._refresh_buttons()
@Slot()
@@ -333,4 +295,4 @@ class BeamlineRecoveryDialog(QDialog):
buttons = QDialogButtonBox(QDialogButtonBox.StandardButton.Close, parent=self)
buttons.rejected.connect(self.reject)
buttons.accepted.connect(self.accept)
layout.addWidget(buttons)
layout.addWidget(buttons)
+394 -96
View File
@@ -2,37 +2,64 @@ from __future__ import annotations
from collections.abc import Callable
from PySide6.QtCore import Slot
from PySide6.QtCore import Slot, QUrl
from PySide6.QtGui import QDesktopServices
from PySide6.QtWidgets import (
QCheckBox,
QDialog,
QDialogButtonBox,
QFrame,
QGridLayout,
QGroupBox,
QHBoxLayout,
QInputDialog,
QLabel,
QMessageBox,
QPushButton,
QTabWidget,
QTextEdit,
QVBoxLayout,
QWidget,
)
from aare.common.logger_config import setup_logger
from aare.common.models import DAQStatusModel
from aare.gui.panels.beamline_recovery_panel import RecoveryPanel
from aare.gui.threads.daq_worker import DAQWorker
from aare.gui.widgets.local_contact_status_widget import LocalContactStatusWidget
from aare.gui.widgets.text_list_dialog import TextListDialog
from aare.gui.widgets.title_label import TitleLabel
logger = setup_logger("aareGUI")
class LocalContactPanel(QFrame):
TAB_STATUS = "Status"
TAB_RECOVERY = "Recovery"
TAB_TELL = "TELL"
TAB_BEC = "BEC"
TAB_HARDWARE = "Hardware"
TAB_DETECTOR = "Detector"
DEVICE_TITLES = {
"bec": "BEC",
"detector": "Detector",
"tell": "TELL",
"aerotech": "Aerotech",
"smargon": "Smargon",
}
def __init__(self, *, daq: DAQWorker, parent=None):
super().__init__(parent)
self._daq = daq
self._sim_checkboxes: dict[str, QCheckBox] = {}
self._links_payload: dict = {}
self._device_state_payload: dict = {}
self._local_contact_error_message: str | None = None
self._last_status: DAQStatusModel | None = None
self._status_widgets: list[LocalContactStatusWidget] = []
self._bec_macros_dialog: TextListDialog | None = None
self._bec_devices_dialog: TextListDialog | None = None
self.setFrameShape(QFrame.Shape.StyledPanel)
self.setFrameShadow(QFrame.Shadow.Raised)
@@ -43,157 +70,312 @@ class LocalContactPanel(QFrame):
layout.addWidget(TitleLabel("Local Contact", parent=self))
self._info_label = QLabel(
"Staff tools for beamline recovery and local-contact operations.\n"
"Some buttons are placeholders until backend pathways are connected.",
"Staff tools for beamline recovery and local-contact operations.",
self,
)
self._info_label.setWordWrap(True)
layout.addWidget(self._info_label)
self._transfer_error_frame = QFrame(self)
self._transfer_error_frame.setVisible(False)
self._transfer_error_frame.setStyleSheet(
"QFrame {"
" background: #fdeaea;"
" color: #8b1e1e;"
" border: 1px solid #e6a8a8;"
" border-radius: 6px;"
"}"
)
transfer_error_layout = QVBoxLayout(self._transfer_error_frame)
transfer_error_layout.setContentsMargins(8, 8, 8, 8)
self._transfer_error_title = QLabel("Error transferring information from DAQ", self._transfer_error_frame)
self._transfer_error_title.setStyleSheet("font-weight: 700;")
transfer_error_layout.addWidget(self._transfer_error_title)
self._transfer_error_text = QTextEdit(self._transfer_error_frame)
self._transfer_error_text.setReadOnly(True)
self._transfer_error_text.setMinimumHeight(90)
transfer_error_layout.addWidget(self._transfer_error_text)
layout.addWidget(self._transfer_error_frame)
self._tabs = QTabWidget(self)
layout.addWidget(self._tabs, 1)
self._tabs.addTab(self._build_status_tab(), self.TAB_STATUS)
self._tabs.addTab(self._build_recovery_tab(), self.TAB_RECOVERY)
self._tabs.addTab(self._build_tell_tab(), self.TAB_TELL)
self._tabs.addTab(self._build_bec_tab(), self.TAB_BEC)
self._tabs.addTab(self._build_hardware_tab(), self.TAB_HARDWARE)
self._tabs.addTab(self._build_detector_tab(), self.TAB_DETECTOR)
self._daq.local_contact_simulation_state_loaded.connect(self._apply_simulation_state)
self._daq.local_contact_device_state_loaded.connect(self._apply_device_state)
self._daq.local_contact_links_loaded.connect(self._apply_links)
self._daq.local_contact_transfer_error.connect(self._show_local_contact_transfer_error)
self._daq.bec_user_macros_loaded.connect(self._show_bec_user_macros)
self._daq.bec_devices_loaded.connect(self._show_bec_devices)
self._daq.update.connect(self._update_from_status)
self._daq.load_local_contact_simulation_state()
self._daq.load_local_contact_device_state()
self._daq.load_local_contact_links()
def set_active_tab(self, tab_name: str) -> None:
for index in range(self._tabs.count()):
if self._tabs.tabText(index).strip().lower() == str(tab_name).strip().lower():
self._tabs.setCurrentIndex(index)
return
logger.warning(f"Unknown Local Contact tab requested: {tab_name}")
def _register_status_widget(self, widget: LocalContactStatusWidget) -> LocalContactStatusWidget:
self._status_widgets.append(widget)
if self._last_status is not None:
widget.set_daq_status(self._last_status)
if self._device_state_payload:
widget.set_device_state_payload(self._device_state_payload)
return widget
def _make_status_widget(
self,
*,
title: str,
summary: str,
fields: tuple[str, ...],
) -> LocalContactStatusWidget:
widget = LocalContactStatusWidget(
title=title,
visible_fields=fields,
summary=summary,
parent=self,
)
return self._register_status_widget(widget)
def _build_status_tab(self) -> QWidget:
tab = QWidget(self)
layout = QVBoxLayout(tab)
layout.setSpacing(8)
layout.addWidget(
self._make_status_widget(
title="DAQ status overview",
summary="Full DAQ and backend status summary for Local Contact.",
fields=(
"beamline_state",
"busy",
"sample",
"tell_connected",
"tell_state",
"tell_error",
"bec",
"detector",
"tell",
"aerotech",
"smargon",
"smargon_connected",
"smargon_error",
"aerotech_connected",
"aerotech_error",
"session",
"open_guis",
"box",
"last_best_res",
"last_best_b_factor",
"crystal_size",
"beamline_name",
"ring_current",
"front_light",
"back_light",
"cryojet",
"shutter",
"exposure_shutter",
"flux",
"transmission",
"zoom",
"commissioning_mode",
"omega",
"beam_size",
"smargon_position",
"aerotech_position",
"detector_description",
"detector_serial",
"dtz",
"energy",
"wavelength",
"beam_center",
),
)
)
layout.addStretch(1)
return tab
def _build_recovery_tab(self) -> QWidget:
tab = QWidget(self)
layout = QVBoxLayout(tab)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(8)
layout.addWidget(
self._make_status_widget(
title="Recovery status",
summary="Recovery-relevant DAQ state.",
fields=("beamline_state", "busy", "sample", "tell_connected", "tell_state"),
)
)
layout.addWidget(RecoveryPanel(daq=self._daq, parent=tab), 1)
return tab
def _build_tell_tab(self) -> QWidget:
tab = QWidget(self)
layout = QGridLayout(tab)
layout = QVBoxLayout(tab)
layout.setSpacing(8)
layout.addWidget(
self._make_status_widget(
title="TELL status",
summary="TELL-focused state and connection details.",
fields=("tell", "tell_connected", "tell_state", "tell_error", "busy", "beamline_state"),
)
)
grid = QGridLayout()
row = 0
layout.addWidget(QLabel("TELL operations", tab), row, 0, 1, 2)
grid.addWidget(QLabel("TELL operations", tab), row, 0, 1, 2)
row += 1
layout.addWidget(
self._make_button(
"Unmount",
self._daq.unmount,
"Requesting sample unmount.",
),
row,
0,
)
layout.addWidget(
self._make_button(
"Dry",
self._daq.tell_dry,
"Requesting TELL dry.",
),
row,
1,
)
grid.addWidget(self._make_button("Unmount", self._daq.unmount, "Requesting sample unmount."), row, 0)
grid.addWidget(self._make_button("Dry", self._daq.tell_dry, "Requesting TELL dry."), row, 1)
row += 1
layout.addWidget(
self._make_button(
"Park and dry",
self._daq.park_and_dry,
"Requesting park and dry.",
),
row,
0,
)
layout.addWidget(
self._make_button(
"Toggle blower",
self._daq.tell_toggle_blower,
"Toggling blower.",
),
row,
1,
)
grid.addWidget(self._make_button("Park and dry", self._daq.park_and_dry, "Requesting park and dry."), row, 0)
grid.addWidget(self._make_button("Toggle blower", self._daq.tell_toggle_blower, "Toggling blower."), row, 1)
row += 1
layout.addWidget(
self._make_button(
"Anneal",
self._anneal_from_dialog,
),
row,
0,
)
grid.addWidget(self._make_button("Anneal", self._anneal_from_dialog), row, 0)
grid.addWidget(self._make_button("TELL access info", self._show_tell_access_info), row, 1)
layout.setRowStretch(row + 1, 1)
wrapper = QWidget(tab)
wrapper.setLayout(grid)
layout.addWidget(wrapper)
layout.addStretch(1)
return tab
def _build_bec_tab(self) -> QWidget:
tab = QWidget(self)
layout = QVBoxLayout(tab)
layout.setSpacing(8)
label = QLabel(
"BEC tools.\n"
"For now this tab only exposes loading user macros.",
tab,
layout.addWidget(
self._make_status_widget(
title="BEC status",
summary="BEC-focused state and related backend status.",
fields=("bec", "busy", "beamline_state", "detector", "aerotech", "smargon"),
)
)
label = QLabel("BEC tools.", tab)
label.setWordWrap(True)
layout.addWidget(label)
layout.addWidget(self._make_button("Load BEC user macros", self._daq.bec_load_user_macros, "Loading BEC user macros."))
layout.addWidget(self._make_button("Show BEC user macros", self._daq.bec_list_all_user_macros, "Listing BEC user macros."))
layout.addWidget(self._make_button("Show BEC position devices", self._daq.bec_list_all_devices, "Listing BEC devices."))
layout.addWidget(
self._make_button(
"Load user macros",
self._daq.bec_load_user_macros,
"Loading BEC user macros.",
"Reinitialise BEC planner/devices",
lambda: self._daq.bec_reinitialise_planner_and_position_devices("auto"),
"Reinitialising BEC planner and position devices.",
)
)
layout.addStretch(1)
return tab
def _build_hardware_tab(self) -> QWidget:
tab = QWidget(self)
layout = QVBoxLayout(tab)
layout.setSpacing(8)
label = QLabel(
"Hardware initialisation tools.\n"
"These are placeholders for future Smargon and Aerotech initialisation support.",
tab,
layout.addWidget(
self._make_status_widget(
title="Hardware status",
summary="Hardware-related backend and beamline status.",
fields=("aerotech", "smargon", "detector", "bec", "tell", "busy", "beamline_state"),
)
)
label = QLabel("Hardware tools grouped by action.", tab)
label.setWordWrap(True)
layout.addWidget(label)
layout.addWidget(
self._make_button(
"Initialise Smargon",
self._daq.initialise_smargon,
"Initialising Smargon.",
self._build_section(
"Web panels",
[
self._build_link_button("Detector web panel", "detector"),
self._build_link_button("Smargon web panel", "smargon"),
self._build_link_button("Aerotech web panel", "aerotech"),
],
)
)
layout.addWidget(
self._make_button(
"Initialise Aerotech",
self._daq.initialise_aerotech,
"Initialising Aerotech.",
)
)
layout.addStretch(1)
layout.addWidget(
self._build_section(
"Restart Backend",
[
self._make_restart_button("tell", "Restart TELL backend"),
self._make_restart_button("bec", "Restart BEC backend"),
self._make_restart_button("detector", "Restart detector backend"),
self._make_restart_button("smargon", "Restart Smargon backend"),
self._make_restart_button("aerotech", "Restart Aerotech backend"),
],
)
)
layout.addWidget(
self._build_section(
"Simulate",
[
self._make_sim_checkbox("tell", "Simulate TELL"),
self._make_sim_checkbox("bec", "Simulate BEC"),
self._make_sim_checkbox("detector", "Simulate detector"),
self._make_sim_checkbox("smargon", "Simulate Smargon"),
self._make_sim_checkbox("aerotech", "Simulate Aerotech"),
],
)
)
layout.addWidget(
self._build_section(
"Initialise",
[
self._make_button("Initialise detector", self._daq.initialise_detector, "Initialising detector."),
self._make_button("Initialise Smargon", self._daq.initialise_smargon, "Initialising Smargon."),
self._make_button("Initialise Aerotech", self._daq.initialise_aerotech, "Initialising Aerotech."),
],
)
)
layout.addStretch(1)
return tab
def _build_detector_tab(self) -> QWidget:
tab = QWidget(self)
layout = QGridLayout(tab)
row = 0
layout.addWidget(QLabel("Detector operations", tab), row, 0, 1, 2)
row += 1
layout = QVBoxLayout(tab)
layout.setSpacing(8)
layout.addWidget(
self._make_status_widget(
title="Detector status",
summary="Detector-focused state and metadata.",
fields=("detector", "bec", "busy", "beamline_state", "dtz", "detector_description", "detector_serial"),
)
)
grid = QGridLayout()
row = 0
grid.addWidget(QLabel("Detector operations", tab), row, 0, 1, 2)
row += 1
grid.addWidget(
self._make_button(
"Cancel current state",
self._daq.cancel,
@@ -202,7 +384,7 @@ class LocalContactPanel(QFrame):
row,
0,
)
layout.addWidget(
grid.addWidget(
self._make_button(
"Take pedestal",
self._daq.detector_take_pedestal,
@@ -211,21 +393,137 @@ class LocalContactPanel(QFrame):
row,
1,
)
row += 1
layout.addWidget(
self._make_button(
"Initialise detector",
self._daq.initialise_detector,
"Initialising detector.",
),
row,
0,
)
layout.setRowStretch(row + 1, 1)
wrapper = QWidget(tab)
wrapper.setLayout(grid)
layout.addWidget(wrapper)
layout.addStretch(1)
return tab
def _build_section(self, title: str, widgets: list[QWidget]) -> QWidget:
box = QGroupBox(title, self)
box_layout = QVBoxLayout(box)
box_layout.setContentsMargins(10, 10, 10, 10)
box_layout.setSpacing(6)
for widget in widgets:
box_layout.addWidget(widget)
return box
def _make_sim_checkbox(self, device: str, text: str) -> QWidget:
row_widget = QWidget(self)
row_layout = QHBoxLayout(row_widget)
row_layout.setContentsMargins(0, 0, 0, 0)
name = QLabel(self.DEVICE_TITLES.get(device, device.title()), row_widget)
name.setMinimumWidth(90)
name.setStyleSheet("font-weight: 700;")
checkbox = QCheckBox(text, row_widget)
checkbox.toggled.connect(lambda checked, d=device: self._daq.set_local_contact_simulation(d, checked))
self._sim_checkboxes[device] = checkbox
row_layout.addWidget(name)
row_layout.addWidget(checkbox, 1)
return row_widget
def _make_restart_button(self, device: str, text: str) -> QWidget:
row_widget = QWidget(self)
row_layout = QHBoxLayout(row_widget)
row_layout.setContentsMargins(0, 0, 0, 0)
name = QLabel(self.DEVICE_TITLES.get(device, device.title()), row_widget)
name.setMinimumWidth(90)
name.setStyleSheet("font-weight: 700;")
button = QPushButton(text, row_widget)
button.clicked.connect(lambda: self._run_logged_action(
f"Restarting {device} backend.",
lambda: self._daq.restart_local_contact_device(device),
))
row_layout.addWidget(name)
row_layout.addWidget(button, 1)
return row_widget
def _build_link_button(self, text: str, key: str) -> QPushButton:
button = QPushButton(text, self)
button.clicked.connect(lambda: self._open_link(key))
return button
@Slot(dict)
def _apply_simulation_state(self, payload: dict) -> None:
for device, checkbox in self._sim_checkboxes.items():
value = bool(payload.get(device, False))
checkbox.blockSignals(True)
checkbox.setChecked(value)
checkbox.blockSignals(False)
@Slot(dict)
def _apply_device_state(self, payload: dict) -> None:
self._device_state_payload = payload or {}
for widget in self._status_widgets:
widget.set_device_state_payload(self._device_state_payload)
@Slot(dict)
def _apply_links(self, payload: dict) -> None:
self._links_payload = payload or {}
@Slot(str)
def _show_local_contact_transfer_error(self, message: str) -> None:
self._local_contact_error_message = message
self._transfer_error_text.setPlainText(message)
self._transfer_error_frame.setVisible(True)
@Slot(list)
def _show_bec_user_macros(self, items: list) -> None:
if self._bec_macros_dialog is None:
self._bec_macros_dialog = TextListDialog(title="BEC user macros", parent=self)
self._bec_macros_dialog.set_items(
[str(item) for item in items],
empty_message="No BEC user macros found.",
)
self._bec_macros_dialog.show()
self._bec_macros_dialog.raise_()
self._bec_macros_dialog.activateWindow()
@Slot(list)
def _show_bec_devices(self, items: list) -> None:
if self._bec_devices_dialog is None:
self._bec_devices_dialog = TextListDialog(title="BEC position devices", parent=self)
self._bec_devices_dialog.set_items(
[str(item) for item in items],
empty_message="No BEC devices found.",
)
self._bec_devices_dialog.show()
self._bec_devices_dialog.raise_()
self._bec_devices_dialog.activateWindow()
@Slot(DAQStatusModel)
def _update_from_status(self, status: DAQStatusModel) -> None:
self._last_status = status
for widget in self._status_widgets:
widget.set_daq_status(status)
if self._local_contact_error_message is None:
self._daq.load_local_contact_device_state()
def _open_link(self, key: str) -> None:
url = self._links_payload.get(key)
if not url:
QMessageBox.information(
self,
"Link unavailable",
f"No configured link is available for {key}.",
)
return
QDesktopServices.openUrl(QUrl(str(url)))
@Slot()
def _show_tell_access_info(self) -> None:
message = self._links_payload.get("tell_hint") or "Please check TELL status via Remmina / VNC."
QMessageBox.information(self, "TELL access", str(message))
def _make_button(
self,
text: str,
@@ -264,7 +562,7 @@ class LocalContactDialog(QDialog):
def __init__(self, *, daq: DAQWorker, parent=None):
super().__init__(parent)
self.setWindowTitle("Local Contact")
self.setMinimumSize(760, 560)
self.setMinimumSize(900, 680)
layout = QVBoxLayout(self)
layout.setContentsMargins(12, 12, 12, 12)
@@ -279,4 +577,4 @@ class LocalContactDialog(QDialog):
layout.addWidget(buttons)
def set_active_tab(self, tab_name: str) -> None:
self._panel.set_active_tab(tab_name)
self._panel.set_active_tab(tab_name)
+178
View File
@@ -70,6 +70,12 @@ class DAQWorker(QObject):
gui_sessions_loaded = Signal(list)
gui_close_requested = Signal(int, int, str)
recovery_action_completed = Signal(str)
bec_user_macros_loaded = Signal(list)
bec_devices_loaded = Signal(list)
local_contact_simulation_state_loaded = Signal(dict)
local_contact_device_state_loaded = Signal(dict)
local_contact_links_loaded = Signal(dict)
local_contact_transfer_error = Signal(str)
polled_devices_status = Signal(str, bool) # (message, is_error)
detector_error = Signal(str, bool) # (message, is_error)
@@ -175,6 +181,9 @@ class DAQWorker(QObject):
self._last_seen_automation_event_ts: dict[int | None, datetime] = {}
self._recurrence_watchers = self._load_recurrence_watchers()
self._local_contact_metadata_poll_enabled = True
self._local_contact_metadata_error: str | None = None
if self.__base_url is not None:
self.start_face_detection_stream()
self.start_baton_stream()
@@ -1330,10 +1339,179 @@ class DAQWorker(QObject):
def tell_toggle_blower(self):
self.generic_post("tell/toggle_blower")
def _disable_local_contact_metadata_polling(self, message: str) -> None:
if self._local_contact_metadata_poll_enabled is False and self._local_contact_metadata_error == message:
return
self._local_contact_metadata_poll_enabled = False
self._local_contact_metadata_error = message
logger.error(f"Disabling Local Contact metadata polling: {message}")
self.local_contact_transfer_error.emit(message)
def _build_local_contact_error_message(self, context: str, reply: QNetworkReply, exc: Exception) -> str:
status = reply.attribute(QNetworkRequest.Attribute.HttpStatusCodeAttribute)
try:
url = reply.request().url().toString()
except Exception:
url = "unknown-url"
return (
f"{context}\n\n"
f"URL: {url}\n"
f"HTTP status: {status}\n"
f"Error: {exc}"
)
@Slot()
def load_local_contact_simulation_state(self):
if not self._local_contact_metadata_poll_enabled:
return
if self.__base_url is None:
self.local_contact_simulation_state_loaded.emit(
{"bec": False, "detector": False, "tell": False, "aerotech": False, "smargon": False}
)
return
request = QNetworkRequest(QUrl(f"{self.__base_url}/local_contact/simulation_state"))
request.setRawHeader(b"Authorization", f"Bearer {self.__token}".encode("utf-8"))
reply = self.__net_manager.get(request)
reply.finished.connect(lambda: self._handle_local_contact_simulation_state_response(reply))
def _handle_local_contact_simulation_state_response(self, reply: QNetworkReply):
try:
response_data = self.handle_response(reply)
payload = json.loads(response_data) if response_data else {}
if not isinstance(payload, dict):
raise RuntimeError("Invalid local contact simulation state payload")
self.local_contact_simulation_state_loaded.emit(payload)
except Exception as e:
message = self._build_local_contact_error_message(
"Error transferring information from DAQ while loading Local Contact simulation state.",
reply,
e,
)
self._disable_local_contact_metadata_polling(message)
@Slot()
def load_local_contact_device_state(self):
if not self._local_contact_metadata_poll_enabled:
return
if self.__base_url is None:
self.local_contact_device_state_loaded.emit({})
return
request = QNetworkRequest(QUrl(f"{self.__base_url}/local_contact/device_state"))
request.setRawHeader(b"Authorization", f"Bearer {self.__token}".encode("utf-8"))
reply = self.__net_manager.get(request)
reply.finished.connect(lambda: self._handle_local_contact_device_state_response(reply))
def _handle_local_contact_device_state_response(self, reply: QNetworkReply):
try:
response_data = self.handle_response(reply)
payload = json.loads(response_data) if response_data else {}
if not isinstance(payload, dict):
raise RuntimeError("Invalid local contact device state payload")
self.local_contact_device_state_loaded.emit(payload)
except Exception as e:
message = self._build_local_contact_error_message(
"Error transferring information from DAQ while loading Local Contact device state.",
reply,
e,
)
self._disable_local_contact_metadata_polling(message)
@Slot()
def load_local_contact_links(self):
if not self._local_contact_metadata_poll_enabled:
return
if self.__base_url is None:
self.local_contact_links_loaded.emit({})
return
request = QNetworkRequest(QUrl(f"{self.__base_url}/local_contact/links"))
request.setRawHeader(b"Authorization", f"Bearer {self.__token}".encode("utf-8"))
reply = self.__net_manager.get(request)
reply.finished.connect(lambda: self._handle_local_contact_links_response(reply))
def _handle_local_contact_links_response(self, reply: QNetworkReply):
try:
response_data = self.handle_response(reply)
payload = json.loads(response_data) if response_data else {}
if not isinstance(payload, dict):
raise RuntimeError("Invalid local contact links payload")
self.local_contact_links_loaded.emit(payload)
except Exception as e:
message = self._build_local_contact_error_message(
"Error transferring information from DAQ while loading Local Contact links.",
reply,
e,
)
self._disable_local_contact_metadata_polling(message)
@Slot(str, bool)
def set_local_contact_simulation(self, device: str, enabled: bool):
self.generic_post(f"local_contact/simulate/{device}?enabled={str(enabled).lower()}")
@Slot(str)
def restart_local_contact_device(self, device: str):
self.generic_post(f"local_contact/restart/{device}")
@Slot()
def bec_load_user_macros(self):
self.generic_post("bec/load_user_macros")
@Slot()
def bec_list_all_user_macros(self):
if self.__base_url is None:
logger.info("GET /bec/user_macros")
self.bec_user_macros_loaded.emit([])
return
request = QNetworkRequest(QUrl(f"{self.__base_url}/bec/user_macros"))
request.setRawHeader(b"Authorization", f"Bearer {self.__token}".encode("utf-8"))
reply = self.__net_manager.get(request)
reply.finished.connect(lambda: self._handle_bec_user_macros_response(reply))
def _handle_bec_user_macros_response(self, reply: QNetworkReply):
try:
response_data = self.handle_response(reply)
payload = json.loads(response_data) if response_data else []
if not isinstance(payload, list):
raise RuntimeError("Invalid BEC user macros payload")
self.bec_user_macros_loaded.emit([str(item) for item in payload])
except Exception as e:
logger.error(f"Failed to list BEC user macros: {e}")
self.http_error.emit(str(e))
@Slot()
def bec_list_all_devices(self):
if self.__base_url is None:
logger.info("GET /bec/devices")
self.bec_devices_loaded.emit([])
return
request = QNetworkRequest(QUrl(f"{self.__base_url}/bec/devices"))
request.setRawHeader(b"Authorization", f"Bearer {self.__token}".encode("utf-8"))
reply = self.__net_manager.get(request)
reply.finished.connect(lambda: self._handle_bec_devices_response(reply))
def _handle_bec_devices_response(self, reply: QNetworkReply):
try:
response_data = self.handle_response(reply)
payload = json.loads(response_data) if response_data else []
if not isinstance(payload, list):
raise RuntimeError("Invalid BEC devices payload")
self.bec_devices_loaded.emit([str(item) for item in payload])
except Exception as e:
logger.error(f"Failed to list BEC devices: {e}")
self.http_error.emit(str(e))
@Slot(str)
def bec_reinitialise_planner_and_position_devices(self, method: str = "auto"):
self.generic_post(f"bec/reinitialise_planner_and_position_devices?method={method}")
@Slot()
def initialise_smargon(self):
self.generic_post("smargon/initialize")
@@ -0,0 +1,346 @@
from __future__ import annotations
from collections.abc import Iterable
from PySide6.QtCore import Qt, Slot
from PySide6.QtWidgets import (
QFrame,
QGridLayout,
QLabel,
QSizePolicy,
QVBoxLayout,
)
from aare.common.models import DAQStatusModel
from aare.gui.widgets.title_label import TitleLabel
class LocalContactStatusWidget(QFrame):
FIELD_TITLES = {
"beamline_state": "Beamline state",
"busy": "Busy",
"sample": "Sample",
"tell_connected": "TELL connected",
"tell_state": "TELL state",
"tell_error": "TELL error",
"bec": "BEC",
"detector": "Detector",
"tell": "TELL backend",
"aerotech": "Aerotech",
"smargon": "Smargon",
"smargon_connected": "Smargon connected",
"smargon_error": "Smargon error",
"aerotech_connected": "Aerotech connected",
"aerotech_error": "Aerotech error",
"session": "Session",
"open_guis": "Open GUIs",
"box": "ML box",
"last_best_res": "Last best res",
"last_best_b_factor": "Last best B factor",
"crystal_size": "Crystal size",
"beamline_name": "Beamline",
"ring_current": "Ring current",
"front_light": "Front light",
"back_light": "Back light",
"cryojet": "Cryojet",
"shutter": "Shutter",
"exposure_shutter": "Exposure shutter",
"flux": "Flux",
"transmission": "Transmission",
"zoom": "Zoom",
"commissioning_mode": "Commissioning mode",
"omega": "Omega",
"beam_size": "Beam size",
"smargon_position": "Smargon position",
"aerotech_position": "Aerotech position",
"detector_description": "Detector description",
"detector_serial": "Detector serial",
"dtz": "Detector distance",
"energy": "Energy",
"wavelength": "Wavelength",
"beam_center": "Beam centre",
}
def __init__(
self,
*,
title: str = "Status",
visible_fields: Iterable[str] | None = None,
summary: str = "DAQ and backend status overview.",
parent=None,
):
super().__init__(parent)
self._last_status: DAQStatusModel | None = None
self._device_state_payload: dict = {}
self._row_widgets: dict[str, tuple[QLabel, QLabel]] = {}
self._title = title
self._visible_fields = tuple(visible_fields or ())
self.setFrameShape(QFrame.Shape.StyledPanel)
self.setFrameShadow(QFrame.Shadow.Raised)
self.setStyleSheet(
"QFrame {"
" background: #f7f9fc;"
" border: 1px solid #c8d3e1;"
" border-radius: 8px;"
"}"
)
self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Maximum)
layout = QVBoxLayout(self)
layout.setContentsMargins(10, 10, 10, 10)
layout.setSpacing(6)
layout.addWidget(TitleLabel(self._title, self))
self._summary = QLabel(summary, self)
self._summary.setWordWrap(True)
self._summary.setStyleSheet("border: none; background: transparent; color: #334155;")
layout.addWidget(self._summary)
self._grid = QGridLayout()
self._grid.setContentsMargins(0, 0, 0, 0)
self._grid.setHorizontalSpacing(12)
self._grid.setVerticalSpacing(4)
layout.addLayout(self._grid)
self._rebuild_rows()
def set_summary_text(self, text: str) -> None:
self._summary.setText(text)
def set_visible_fields(self, fields: Iterable[str]) -> None:
self._visible_fields = tuple(fields)
self._rebuild_rows()
self._refresh()
def _clear_grid(self) -> None:
while self._grid.count():
item = self._grid.takeAt(0)
widget = item.widget()
if widget is not None:
widget.deleteLater()
def _rebuild_rows(self) -> None:
self._clear_grid()
self._row_widgets.clear()
for row, key in enumerate(self._visible_fields):
title = QLabel(self.FIELD_TITLES.get(key, key.replace("_", " ").title()), self)
title.setStyleSheet("font-weight: 700; border: none; background: transparent; color: #1e293b;")
value = QLabel(self._badge("WAITING", tone="neutral"), self)
value.setWordWrap(True)
value.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse)
value.setStyleSheet("border: none; background: transparent; color: #334155;")
self._row_widgets[key] = (title, value)
self._grid.addWidget(title, row, 0, alignment=Qt.AlignmentFlag.AlignTop)
self._grid.addWidget(value, row, 1)
def _badge(self, text: str, *, tone: str = "neutral") -> str:
palette = {
"good": ("#e7f6ea", "#1f6a3a"),
"warn": ("#fff3cd", "#7a4b00"),
"bad": ("#fdeaea", "#8b1e1e"),
"neutral": ("#e9eef5", "#475569"),
"info": ("#e8f1ff", "#12406a"),
}
background, foreground = palette.get(tone, palette["neutral"])
return (
f"<span style='background:{background};color:{foreground};"
f"padding:2px 6px;border-radius:8px;'><b>{text}</b></span>"
)
def _format_bool(self, value: bool | None, *, true_text: str = "CONNECTED", false_text: str = "DISCONNECTED") -> str:
if value is True:
return self._badge(true_text, tone="good")
if value is False:
return self._badge(false_text, tone="bad")
return self._badge("UNKNOWN", tone="neutral")
def _format_device_status(self, device: str) -> str:
payload = self._device_state_payload.get(device, {}) if isinstance(self._device_state_payload, dict) else {}
mode = str(payload.get("mode", "unknown"))
error = payload.get("error")
if error:
return (
f"{self._badge('ERROR', tone='bad')} "
f"{self._badge(mode.upper(), tone='warn')} "
f"<span style='color:#8b1e1e;'>{error}</span>"
)
if mode == "simulated":
return self._badge("SIMULATED", tone="warn")
if mode == "live":
return self._badge("LIVE", tone="good")
return self._badge("UNKNOWN", tone="neutral")
def _format_tell_state(self, status: DAQStatusModel) -> str:
tell_state = getattr(status, "tell_state", None)
if tell_state is None:
return self._badge("UNAVAILABLE", tone="neutral")
activity = getattr(tell_state.activity, "value", str(tell_state.activity))
message = str(getattr(tell_state, "message", "") or "").strip()
text = self._badge(str(activity).upper(), tone="info")
if message:
text = f"{text} {message}"
return text
def _format_sample(self, status: DAQStatusModel) -> str:
if status.sample is None:
return self._badge("NONE", tone="neutral")
sample_name = str(getattr(status.sample, "sample_name", "") or "").strip()
sample_loc = ""
if getattr(status.sample, "location", None) is not None:
try:
sample_loc = f" ({status.sample.loc_str()})"
except Exception:
sample_loc = ""
if sample_name:
return f"{self._badge('MOUNTED', tone='info')} {sample_name}{sample_loc}"
return self._badge("MOUNTED", tone="info")
def _format_state(self, status: DAQStatusModel) -> str:
state = getattr(status, "state", None)
if state is None:
return self._badge("UNKNOWN", tone="neutral")
display_name = getattr(state, "display_name", None)
if callable(display_name):
return self._badge(str(display_name()).upper(), tone="info")
return self._badge(str(getattr(state, 'name', state)).upper(), tone="info")
def _format_field_value(self, key: str, status: DAQStatusModel) -> str:
if key == "beamline_state":
return self._format_state(status)
if key == "busy":
return self._badge("BUSY", tone="warn") if bool(status.busy) else self._badge("IDLE", tone="good")
if key == "sample":
return self._format_sample(status)
if key == "tell_connected":
return self._format_bool(getattr(status, "tell_connected", None))
if key == "tell_state":
return self._format_tell_state(status)
if key == "tell_error":
err = getattr(status, "tell_error", None)
return str(err) if err else self._badge("NONE", tone="good")
if key in {"bec", "detector", "tell", "aerotech", "smargon"}:
return self._format_device_status(key)
if key == "smargon_connected":
return self._format_bool(getattr(status, "smargon_connected", None))
if key == "smargon_error":
err = getattr(status, "smargon_error", None)
return str(err) if err else self._badge("NONE", tone="good")
if key == "aerotech_connected":
return self._format_bool(getattr(status, "aerotech_connected", None))
if key == "aerotech_error":
err = getattr(status, "aerotech_error", None)
return str(err) if err else self._badge("NONE", tone="good")
if key == "session":
session = getattr(status, "session", None)
if session is None:
return self._badge("UNKNOWN", tone="neutral")
return f"{getattr(session, 'session', '-')}, pgroup={getattr(session, 'current_pgroup', '-')}, staff={getattr(session, 'staff', False)}"
if key == "open_guis":
open_guis = getattr(status, "open_guis", []) or []
if not open_guis:
return self._badge("NONE", tone="neutral")
return str(len(open_guis))
if key == "box":
box = getattr(status, "box", None)
if box is None:
return self._badge("NONE", tone="neutral")
return f"({box.top_x:.1f}, {box.top_y:.1f}) → ({box.bottom_x:.1f}, {box.bottom_y:.1f})"
if key == "last_best_res":
value = getattr(status, "last_best_res", None)
return f"{value:.3f} Å" if value is not None else self._badge("NONE", tone="neutral")
if key == "last_best_b_factor":
value = getattr(status, "last_best_b_factor", None)
return f"{value:.3f}" if value is not None else self._badge("NONE", tone="neutral")
if key == "crystal_size":
crystal = getattr(status, "crystal_size", None)
if crystal is None:
return self._badge("NONE", tone="neutral")
return f"x={crystal.x:.3f}, y={crystal.y:.3f}, z={crystal.z:.3f}"
if key == "beamline_name":
return str(getattr(status.bl, "name", "-"))
if key == "ring_current":
return f"{getattr(status.bl, 'ring_current_mA', 0.0):.2f} mA"
if key == "front_light":
return f"{getattr(status.bl, 'front_light', 0.0):.1f} %"
if key == "back_light":
return f"{getattr(status.bl, 'back_light', 0.0):.1f} %"
if key == "cryojet":
return f"{getattr(status.bl, 'cryojet_K', 0.0):.2f} K"
if key == "shutter":
return self._format_bool(getattr(status.bl, "shutter_open", None), true_text="OPEN", false_text="CLOSED")
if key == "exposure_shutter":
return self._format_bool(getattr(status.bl, "exp_shutter_open", None), true_text="OPEN", false_text="CLOSED")
if key == "flux":
return f"{getattr(status.bl, 'flux_ph_s', 0.0):.3g} ph/s"
if key == "transmission":
transmission = getattr(status.bl, "transmission", None)
return f"{100.0 * transmission:.2f} %" if transmission is not None else self._badge("NONE", tone="neutral")
if key == "zoom":
return f"{getattr(status.bl, 'zoom', 0.0):.3f}"
if key == "commissioning_mode":
return self._format_bool(getattr(status.bl, "commissioning_mode", None), true_text="ON", false_text="OFF")
if key == "omega":
return f"{getattr(status.geom, 'omega_deg', 0.0):.3f}°"
if key == "beam_size":
beam_size = getattr(status.geom, "beam_size_mm", None)
if beam_size is None:
return self._badge("UNKNOWN", tone="neutral")
return f"{beam_size.x * 1000.0:.1f} × {beam_size.y * 1000.0:.1f} µm"
if key == "smargon_position":
smargon = getattr(status.geom, "smargon", None)
if smargon is None or getattr(smargon, "sh_mm", None) is None:
return self._badge("UNKNOWN", tone="neutral")
return (
f"sh=({smargon.sh_mm.x:.4f}, {smargon.sh_mm.y:.4f}, {smargon.sh_mm.z:.4f}) mm, "
f"phi={smargon.phi_deg:.3f}°, chi={smargon.chi_deg:.3f}°"
)
if key == "aerotech_position":
pos = getattr(status.geom, "aerotech_meas", None)
if pos is None:
return self._badge("UNKNOWN", tone="neutral")
return f"({pos.x:.4f}, {pos.y:.4f}, {pos.z:.4f}) mm"
if key == "detector_description":
return str(getattr(status.diffraction, "detector_description", "-"))
if key == "detector_serial":
return str(getattr(status.diffraction, "detector_serial_number", "-"))
if key == "dtz":
return f"{getattr(status.diffraction, 'dtz_mm', 0.0):.3f} mm"
if key == "energy":
return f"{getattr(status.diffraction, 'energy_keV', 0.0):.4f} keV"
if key == "wavelength":
wavelength = getattr(status.diffraction, "wavelength_angstrom", None)
return f"{wavelength:.5f} Å" if wavelength is not None else self._badge("NONE", tone="neutral")
if key == "beam_center":
beam_center = getattr(status.diffraction, "beam_center_pxl", None)
if beam_center is None:
return self._badge("UNKNOWN", tone="neutral")
return f"({beam_center[0]:.2f}, {beam_center[1]:.2f}) px"
return self._badge("N/A", tone="neutral")
def _refresh(self) -> None:
if self._last_status is None:
for _key, (_title, value) in self._row_widgets.items():
value.setText(self._badge("WAITING", tone="neutral"))
return
for key, (_title, value) in self._row_widgets.items():
value.setText(self._format_field_value(key, self._last_status))
@Slot(dict)
def set_device_state_payload(self, payload: dict) -> None:
self._device_state_payload = payload or {}
self._refresh()
@Slot(DAQStatusModel)
def set_daq_status(self, status: DAQStatusModel) -> None:
self._last_status = status
self._refresh()
+51
View File
@@ -0,0 +1,51 @@
from __future__ import annotations
from collections.abc import Iterable
from PySide6.QtGui import QGuiApplication
from PySide6.QtWidgets import (
QDialog,
QDialogButtonBox,
QPushButton,
QTextEdit,
QVBoxLayout,
)
class TextListDialog(QDialog):
def __init__(self, *, title: str, parent=None):
super().__init__(parent)
self.setWindowTitle(title)
self.setMinimumSize(680, 460)
layout = QVBoxLayout(self)
layout.setContentsMargins(12, 12, 12, 12)
layout.setSpacing(8)
self._text = QTextEdit(self)
self._text.setReadOnly(True)
layout.addWidget(self._text, 1)
buttons = QDialogButtonBox(QDialogButtonBox.StandardButton.Close, parent=self)
self._copy_button = QPushButton("Copy to clipboard", self)
self._copy_button.clicked.connect(self._copy_to_clipboard)
buttons.addButton(self._copy_button, QDialogButtonBox.ButtonRole.ActionRole)
buttons.rejected.connect(self.reject)
buttons.accepted.connect(self.accept)
layout.addWidget(buttons)
def set_items(self, items: Iterable[str], *, empty_message: str = "No items found.") -> None:
values = [str(item) for item in items if str(item).strip()]
if not values:
self._text.setPlainText(empty_message)
return
self._text.setPlainText("\n".join(values))
def set_text(self, text: str) -> None:
self._text.setPlainText(str(text or ""))
def _copy_to_clipboard(self) -> None:
clipboard = QGuiApplication.clipboard()
if clipboard is not None:
clipboard.setText(self._text.toPlainText())