diff --git a/src/aare/daq/config.py b/src/aare/daq/config.py
index 107876ac..1932f5a0 100644
--- a/src/aare/daq/config.py
+++ b/src/aare/daq/config.py
@@ -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())
diff --git a/src/aare/daq/daq.py b/src/aare/daq/daq.py
index 21359715..c690734d 100644
--- a/src/aare/daq/daq.py
+++ b/src/aare/daq/daq.py
@@ -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)
diff --git a/src/aare/daq/devices.py b/src/aare/daq/devices.py
index 35906deb..33415a96 100644
--- a/src/aare/daq/devices.py
+++ b/src/aare/daq/devices.py
@@ -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
diff --git a/src/aare/daq/server.py b/src/aare/daq/server.py
index 202ef4bd..62c817b4 100644
--- a/src/aare/daq/server.py
+++ b/src/aare/daq/server.py
@@ -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)):
"""
diff --git a/src/aare/devices/bec_worker.py b/src/aare/devices/bec_worker.py
index b513c1b9..0a1e58be 100644
--- a/src/aare/devices/bec_worker.py
+++ b/src/aare/devices/bec_worker.py
@@ -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()
diff --git a/src/aare/gui/main_window.py b/src/aare/gui/main_window.py
index e42ea2d1..aa476da3 100644
--- a/src/aare/gui/main_window.py
+++ b/src/aare/gui/main_window.py
@@ -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:
diff --git a/src/aare/gui/panels/beamline_recovery_panel.py b/src/aare/gui/panels/beamline_recovery_panel.py
index 3ce97a5d..b47dbf2c 100644
--- a/src/aare/gui/panels/beamline_recovery_panel.py
+++ b/src/aare/gui/panels/beamline_recovery_panel.py
@@ -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)
\ No newline at end of file
diff --git a/src/aare/gui/panels/local_contact_panel.py b/src/aare/gui/panels/local_contact_panel.py
index 22f0b0a0..e126cadb 100644
--- a/src/aare/gui/panels/local_contact_panel.py
+++ b/src/aare/gui/panels/local_contact_panel.py
@@ -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)
\ No newline at end of file
diff --git a/src/aare/gui/threads/daq_worker.py b/src/aare/gui/threads/daq_worker.py
index cefb5db9..8387a8a6 100644
--- a/src/aare/gui/threads/daq_worker.py
+++ b/src/aare/gui/threads/daq_worker.py
@@ -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")
diff --git a/src/aare/gui/widgets/local_contact_status_widget.py b/src/aare/gui/widgets/local_contact_status_widget.py
new file mode 100644
index 00000000..21af5b60
--- /dev/null
+++ b/src/aare/gui/widgets/local_contact_status_widget.py
@@ -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"{text}"
+ )
+
+ 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"{error}"
+ )
+ 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()
\ No newline at end of file
diff --git a/src/aare/gui/widgets/text_list_dialog.py b/src/aare/gui/widgets/text_list_dialog.py
new file mode 100644
index 00000000..3a92feef
--- /dev/null
+++ b/src/aare/gui/widgets/text_list_dialog.py
@@ -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())
\ No newline at end of file