wip: move components to label

2026-04-29 14:21:46 +02:00
parent f3fbdbf5f2
commit 588152871c
1 changed files with 82 additions and 0 deletions
@@ -0,0 +1,82 @@
+from __future__ import annotations
+
+import builtins
+from typing import TYPE_CHECKING
+
+from bec_lib import bec_logger
+from debye_bec.devices.absorber import STATUS as ABS_STATUS
+
+logger = bec_logger.logger
+# import builtins to avoid linter errors
+dev = builtins.__dict__.get("dev")
+
+class MoveToLabelError(Exception):
+    """Exception for the MoveToLabel function"""
+
+def move_to_label():
+    """
+    Function to move several motors to a specific position defined in the label dict.
+    """
+
+    label = get_device_conditions(label="digitalTwin")
+
+    # Get absorber status and close if open
+    logger.info("Check Frontend Absorber Status")
+    abs_was_open = dev.abs.status.get() == ABS_STATUS.OPEN
+    if abs_was_open:
+        logger.info("  Close Frontend Absorber")
+        status = dev.abs.close()
+        status.wait()
+
+    # Move Frontend Slits
+    logger.info("Move Frontend Slits into position")
+    devices = ["sldi_centerx", "sldi_centery", "sldi_gapx", "sldi_gapy"]
+    matches = {key: label[key] for key in devices if key in label}
+    statuses = []
+    for device in matches.values():
+        statuses.append(device['device'].move(device['value']))
+    for status in statuses:
+        status.wait(timeout=30)
+
+    # Move Collimating mirror
+    logger.info("Move Collimating Mirror into position")
+    if "cm_rotx" in label: # pitch
+        logger.info("  Move pitch into position")
+        surveyed_movement(
+            axis=label['cm_rotx'],
+            surveyed_axes= [
+                {'device': dev.cm_rotz, 'abs_tol': 0.1},
+            ]
+        )
+
+    # Restore absorber position
+    logger.info("Restore Frontend Absorber Status")
+    if abs_was_open:
+        status = dev.abs.open()
+        status.wait()
+
+
+def surveyed_movement(axis, surveyed_axes):
+    """
+    Moves an axis while surverying a set of axes.
+
+    Args:
+        axis (DeviceCondition): Device condition
+        surveyed_axes (list): List of dicts (same format as DeviceCondition)
+
+    Raises:
+        If during movement of axis, one of the surveyed axes moves out of tolerance.
+    """
+
+    for surv_ax in surveyed_axes:
+        surv_ax['old_value'] = surv_ax['device'].read()
+    status = axis['device'].move(axis['value'])
+    while status.status == 'RUNNING':
+        for surv_ax in surveyed_axes:
+            if abs(surv_ax['device'].read() - surv_ax['old_value']) > surv_ax['abs_tol']:
+                axis['device'].stop()
+                raise MoveToLabelError(
+                    f"During movement of {axis['device'].name}, {surv_ax['device'].name} " +
+                    f"started to move unexpectedly (old pos: {surv_ax['old_value']}, " +
+                    f"current pos: {surv_ax['device'].read()})"
+                )