wip(digital_twin): Move core logic to ipython client

debye_bec/bec_ipython_client/plugins/digital_twin/__init__.py

@@ -0,0 +1,3 @@
+from .beamline import get_parameters
+
+parameters = get_parameters()

debye_bec/bec_ipython_client/plugins/digital_twin/beamline.py

@@ -0,0 +1,51 @@
+import socket
+
+from bec_lib import bec_logger
+
+from .types import BeamlineId
+
+logger = bec_logger.logger
+
+
+def get_beamline_id() -> BeamlineId:
+    """
+    Based on the bec servers hostname, tries to extract the beamline
+    identifier (e.g. x01da, x10da, etc).
+
+    Raises:
+        ValueError if beamline cannot be extracted from hostname or beamline not implemented.
+    """
+    bec_hostname = socket.gethostname()
+    start = bec_hostname.find("x")
+    if start != -1:
+        beamline = bec_hostname[start : start + 5]
+        match beamline:
+            case "x01da":
+                return BeamlineId.X01DA
+            case "x10da":
+                return BeamlineId.X10DA
+            case _:
+                raise ValueError(f"Not implemented beamline {beamline}")
+    else:
+        logger.warning(f"Failed to extract beamline from bec server hostname {bec_hostname}")
+        choice = input("Do you want to manually select a beamline? (yes/no): ").strip().lower()
+        if choice in ["yes", "y"]:
+            bl = input(f"Choose from: {[bl.value for bl in BeamlineId]}")
+            if bl in BeamlineId:
+                logger.info(f"Manually selected beamline {bl}")
+                return BeamlineId(bl)
+            else:
+                raise ValueError(f"Wrong selection {bl}")
+        else:
+            raise ValueError("Cannot open digital twin without a beamline")
+
+
+def get_parameters():
+    beamline = get_beamline_id()
+    if beamline == "x01da":
+        from . import x01da_parameters as parameters
+    elif beamline == "x10da":
+        from . import x10da_parameters as parameters
+    else:
+        raise ValueError(f"Unknown beamline: {beamline}")
+    return parameters

debye_bec/bec_ipython_client/plugins/digital_twin/digital_twin.py

@@ -0,0 +1,389 @@
+from pathlib import Path
+
+import numpy as np
+import yaml
+from bec_lib import bec_logger
+from bec_lib.logger import bec_logger
+
+from . import parameters as bl
+from .beamline import get_beamline_id
+from .types import BeamlineId, ConfigDict
+
+OFFSET_FILE_X01DA = Path(__file__).with_name("x01da_offsets.yaml")
+OFFSET_FILE_X10DA = Path(__file__).with_name("x10da_offsets.yaml")
+
+logger = bec_logger.logger
+
+"""
+The idea is to move the core logic of digital_twin to this file. Only keep the gui elements in the widget.
+This way, the scheduler widget can access digital twin without loading the GUI (GUI is still needed for the item creation, but not the item execution)
+    Scheduler will extract assistant inputs (get_assistant_config) during item creation
+    Scheduler will use digital_twin.calculate_positons to calculate positions and digital_twin.move_all to move the motors
+"""
+
+
+class DigitalTwinCore:
+
+    def __init__(self):
+        logger.info("This is the digital twin from the ipython client!")
+        self.beamline = get_beamline_id()
+        self.offset_file = Path()
+        match self.beamline:
+            case "x01da":
+                self.offset_file = OFFSET_FILE_X01DA
+            case "x10da":
+                self.offset_file = OFFSET_FILE_X10DA
+        self.offsets = {}
+        self.load_offsets()
+
+    def move_with_config(self, config):
+        positions = self.calc_positions(self.beamline, config)
+        positions = self.apply_offsets(positions, nested_config=True)
+        logger.info(f"Would now move to these positions: {positions}")
+
+    def load_offsets(self):
+        if self.offsets == {}:
+            logger.info("Load beamline offsets")
+            if not self.offset_file.exists():
+                raise FileNotFoundError(f"Offset file not found: {self.offset_file}")
+
+            with self.offset_file.open("r", encoding="utf-8") as f:
+                data = yaml.safe_load(f)
+
+            if not isinstance(data, dict):
+                raise ValueError(f"Expected a YAML mapping, got {type(data).__name__}")
+
+            self.offsets = data
+        else:
+            logger.info("Unload beamline offsets")
+            self.offsets = {}
+
+    def apply_offsets(self, config, nested_config=False):
+        for axis, axis_data in config.items():
+            if axis in self.offsets:
+                axis_offsets = self.offsets[axis]
+                if "modifier" in axis_offsets and "offset" in axis_offsets:
+                    for idx, rng in enumerate(axis_offsets["modifier"]["range"]):
+                        modifier_axis = axis_offsets["modifier"]["axis"]
+                        modifier_value = (
+                            config[modifier_axis]["value"]
+                            if nested_config
+                            else config[modifier_axis]
+                        )
+                        if rng[0] < modifier_value < rng[1]:
+                            if nested_config:
+                                axis_data["value"] += axis_offsets["offset"][idx]
+                            else:
+                                config[axis] += axis_offsets["offset"][idx]
+                            break
+                elif "offset" in axis_offsets:
+                    if nested_config:
+                        axis_data["value"] += axis_offsets["offset"]
+                    else:
+                        config[axis] += axis_offsets["offset"]
+        return config
+
+    def remove_offsets(self, config):
+        for axis, _ in config.items():
+            if axis in self.offsets:
+                axis_offsets = self.offsets[axis]
+                if "modifier" in axis_offsets and "offset" in axis_offsets:
+                    for idx, rng in enumerate(axis_offsets["modifier"]["range"]):
+                        if rng[0] < config[axis_offsets["modifier"]["axis"]] < rng[1]:
+                            config[axis] -= axis_offsets["offset"][idx]
+                            break
+                elif "offset" in axis_offsets:
+                    config[axis] -= axis_offsets["offset"]
+        return config
+
+    @staticmethod
+    def calc_positions(beamline: BeamlineId, cfg: ConfigDict) -> dict[str, dict[str, float]]:
+        """
+        Calculates the positions of axes based on a beamline config.
+
+        Args:
+            cfg(ConfigDict): Dictionary with beamline config
+
+        Returns:
+            dict[str, dict[str, float]]: Dictionary mapping device names to dictionaries
+                containing a "value" key with the corresponding float value (position).
+        """
+
+        pos = {}
+
+        ## FE slits
+        trxr = -np.arctan(cfg["h_acc"]) * bl.feSlits.center1[1]
+        trxw = (
+            (np.arctan(cfg["h_acc"]) * bl.feSlits.center1[1])
+            / bl.feSlits.center1[1]
+            * bl.feSlits.center2[1]
+        )
+        tryb = -np.arctan(cfg["v_acc"]) * bl.feSlits.center1[1]
+        tryt = (
+            (np.arctan(cfg["v_acc"]) * bl.feSlits.center1[1])
+            / bl.feSlits.center1[1]
+            * bl.feSlits.center2[1]
+        )
+
+        xgap = trxw - trxr
+        ygap = tryt - tryb
+
+        pos["sldi_gapx"] = {"value": xgap}
+        pos["sldi_gapy"] = {"value": ygap}
+
+        ## Collimating Mirror
+        obj_dist = bl.cm.center[1]  # object distance
+        beam_vs = 2 * obj_dist * np.tan(cfg["v_acc"])  # vertical size of beam after CM
+
+        # TRX
+        if cfg["cm_stripe"] in bl.cm.surface:
+            index = bl.cm.surface.index(cfg["cm_stripe"])
+        else:
+            raise ValueError(f"Requested stripe {cfg['cm_stripe']} not found in parameters!")
+        cm_trx = -(bl.cm.limOptX[0][index] + bl.cm.limOptX[1][index]) / 2
+        pos["cm_trx"] = {"value": cm_trx}
+
+        # TRY
+        height = obj_dist * np.tan(cfg["v_acc"]) ** 2 * 1 / np.tan(cfg["cm_pitch"])
+        pos["cm_try"] = {"value": height}
+
+        # Pitch
+        pos["cm_rotx"] = {
+            "value": -cfg["cm_pitch"] * 1e3
+        }  # invert and convert to mrad (same as EGU of rotx axis)
+
+        # Bending Radius
+        radius = (
+            2.0 * obj_dist / np.sin(cfg["cm_pitch"])
+        )  # Elements of modern X-ray Physics, page 108 ff.
+        pos["cm_bnd_radius"] = {"value": radius * 1e-6}  # Convert to km
+
+        ## Monochromator
+        if cfg["mo1_mode"] == "Monochromatic":
+            # Add 2x CM pitch to the bragg angle
+            bragg = cfg["mo1_bragg"]
+        elif cfg["mo1_mode"] == "Pinkbeam":
+            # Align xtal surfaces parallel to beam
+            bragg = 0
+        else:
+            raise ValueError("Monochromator mode not supported")
+        pos["mo1_bragg_angle"] = {"value": bragg / np.pi * 180}  # Bragg angle in deg
+
+        # TRY, Height
+        l = bl.mo1.xtalGap[0] / np.sin(cfg["mo1_bragg"])
+        yhor = l * np.cos(2.0 * (cfg["mo1_bragg"] + cfg["cm_pitch"]))
+        yver = yhor * np.tan(2.0 * cfg["cm_pitch"])
+
+        if cfg["mo1_mode"] == "Monochromatic":
+            beam_offset_mo1 = (
+                l * np.sin(2.0 * (cfg["mo1_bragg"] + cfg["cm_pitch"])) - yver
+            )  # Resultat ist korrekt!
+        elif cfg["mo1_mode"] == "Pinkbeam":
+            beam_offset_mo1 = 0
+        else:
+            raise ValueError("Monochromator mode not supported")
+
+        def csc(a):
+            return 1 / np.sin(a)
+
+        def cot(a):
+            return 1 / np.tan(a)
+
+        # calculate height of center of first crystal surface
+        f = bl.mo1.rotOffset  # rotation offset, mm
+        d = bl.mo1.heightOffset  # xtal height offset, mm
+        c = d * csc(cfg["mo1_bragg"]) - f * cot(cfg["mo1_bragg"])
+
+        # Calculate height of center of rotation
+        b = np.sqrt(
+            d**2 * csc(cfg["mo1_bragg"]) ** 2
+            - 2 * d * f * cot(cfg["mo1_bragg"]) * csc(cfg["mo1_bragg"])
+            + f**2 * cot(cfg["mo1_bragg"]) ** 2
+            + f**2
+        )
+        h = np.cos(np.pi / 2 - np.arctan(f / c) - cfg["mo1_bragg"] - 2 * cfg["cm_pitch"]) * b
+        h2 = ((bl.mo1.center[1] - bl.cm.center[1]) - np.sqrt(b**2 - h**2)) * np.tan(
+            2 * cfg["cm_pitch"]
+        )
+        height_mo1_real = (
+            h + h2
+        )  # per design, the height should not change if the pitch of the CM is not changed!
+        if cfg["mo1_mode"] == "Monochromatic":
+            pass
+        elif cfg["mo1_mode"] == "Pinkbeam":
+            height_mo1_real = (
+                height_mo1_real - 13
+            )  # Move down to let beam pass between both crystal without touching copper cooler
+        else:
+            raise ValueError("Monochromator mode not supported")
+        pos["mo1_try"] = {"value": height_mo1_real}
+
+        # TRX, Crystal selection
+        if cfg["mo1_mode"] == "Monochromatic":
+            xtal = cfg["mo1_xtal"].translate(
+                str.maketrans("", "", "()")
+            )  # Remove brackets from xtal name to conform with parameters
+            if xtal in bl.mo1.xtal:
+                index = bl.mo1.xtal.index(xtal)
+            else:
+                raise ValueError(f"Requested xtal {xtal} not found in parameters!")
+            pos["mo1_trx"] = {"value": bl.mo1.xtalOffsetX[index]}
+        else:
+            pos["mo1_trx"] = {"value": 0}
+
+        diag = bl.mo1.xtalGap[0] / np.sin(cfg["mo1_bragg"])  # Calculations for Mono
+        dz = diag * np.cos(2 * (cfg["cm_pitch"] + cfg["mo1_bragg"]))
+
+        ## Slits 1
+        d = bl.opSlits1.center[1] - bl.cm.center[1] - dz
+        sl1_beam_height = d * np.tan(2 * cfg["cm_pitch"]) + beam_offset_mo1
+        pos["sl1_centery"] = {"value": sl1_beam_height}
+        pos["sl1_gapy"] = {"value": beam_vs}
+
+        ## Beam Monitor 1
+        d = bl.opBM1.center[1] - bl.cm.center[1] - dz
+        bm1_beam_height = d * np.tan(2 * cfg["cm_pitch"]) + beam_offset_mo1
+        pos["bm1_try"] = {"value": bm1_beam_height}
+
+        ## Focusing Mirror
+        p = bl.fm.center[1]
+        q = cfg["smpl"] - bl.fm.center[1]
+        f = (p * q) / (p + q)  # focal length
+
+        # Bender radius
+        if cfg["fm_qy"] is None:
+            radius = 2 * q / np.sin(cfg["fm_rotx"])  # ideal bending radius for focused beam
+        else:
+            radius = (
+                2 * cfg["fm_qy"] / np.sin(cfg["fm_rotx"])
+            )  # ideal bending radius for unfocused beam
+        pos["fm_bnd_radius"] = {"value": radius * 1e-6}  # Convert to km
+
+        # Pitch
+        d = bl.fm.center[1] - bl.cm.center[1] - dz
+        fm_rotx = (
+            2 * cfg["cm_pitch"] - cfg["fm_rotx"]
+        )  # calculate pitch in absolute values (according to horizontal plane)
+        pos["fm_rotx"] = {
+            "value": -fm_rotx * 1e3
+        }  # invert and convert to mrad (same as EGU of rotx axis)
+
+        if cfg["fm_stripe"] in ("Rh (toroid)", "Pt (toroid)"):
+
+            # TRY
+            if cfg["fm_stripe"] == "Rh (toroid)":
+                r = bl.fm.r[0]
+                h_cyl = bl.fm.hToroid[0]
+            else:  # PT toroid
+                r = bl.fm.r[1]
+                h_cyl = bl.fm.hToroid[1]
+            width_beam = 2 * bl.fm.center[1] * np.tan(cfg["h_acc"] * 1e-3)
+            alpha = np.arccos(1 - width_beam**2 / (2 * r**2))
+            h = r - (r * np.cos(alpha / 2))
+            fm_beam_height = (d * np.tan(2 * cfg["cm_pitch"]) + beam_offset_mo1) * cfg[
+                "fm_gain_height"
+            ]
+            fm_height = (d * np.tan(2 * cfg["cm_pitch"]) + beam_offset_mo1 - h_cyl + h / 2) * cfg[
+                "fm_gain_height"
+            ]
+            pos["fm_try"] = {"value": fm_height}
+
+            # TRX
+            if cfg["fm_stripe"] == "Rh (toroid)":
+                x_cyl = -bl.fm.xToroid[0]
+            else:
+                x_cyl = -bl.fm.xToroid[1]
+            pos["fm_trx"] = {"value": x_cyl}
+
+        elif cfg["fm_stripe"] in ("Rh (flat)", "Pt (flat)"):
+
+            # TRY
+            fm_height = (d * np.tan(2 * cfg["cm_pitch"]) + beam_offset_mo1) * cfg["fm_gain_height"]
+            fm_beam_height = fm_height
+            pos["fm_try"] = {"value": fm_height}
+
+            # TRX
+            if cfg["fm_stripe"] == "Rh (flat)":
+                x_flat = -bl.fm.xFlat[0]
+            else:
+                x_flat = -bl.fm.xFlat[1]
+            pos["fm_trx"] = {"value": x_flat}
+
+        else:
+            raise ValueError("FM Stripe selection not valid")
+
+        pos["fm_roty"] = {"value": 0}
+        pos["fm_rotz"] = {"value": 0}
+
+        ## Slits 2
+        if hasattr(bl, "opSlits2"):
+            d = bl.opSlits2.center[1] - bl.fm.center[1]
+            sl2_beam_height = fm_beam_height - d * np.tan(
+                -(2 * cfg["cm_pitch"] - 2 * cfg["fm_rotx"])
+            )
+            pos["sl2_centery"] = {"value": sl2_beam_height}
+            pos["sl2_gapy"] = {"value": beam_vs}
+
+        ## Beam Monitor 2
+        d = bl.opBM2.center[1] - bl.fm.center[1]
+        bm2_beam_height = fm_beam_height - d * np.tan(-(2 * cfg["cm_pitch"] - 2 * cfg["fm_rotx"]))
+        pos["bm2_try"] = {"value": bm2_beam_height}
+
+        ## Optical Table
+
+        if beamline == "x01da":
+            # TRY
+            d = bl.ehWindow.center[1] - bl.fm.center[1]
+            ot_height = fm_beam_height - d * np.tan(-(2 * cfg["cm_pitch"] - 2 * cfg["fm_rotx"]))
+            pos["ot_try"] = {"value": ot_height}
+
+            # Pitch
+            ot_pitch = -(2 * cfg["cm_pitch"] - 2 * cfg["fm_rotx"])
+            pos["ot_rotx"] = {"value": ot_pitch * 1e3}
+
+            # TRZ ES1
+            ot_es1_trz = cfg["smpl"]
+            pos["ot_es1_trz"] = {"value": ot_es1_trz}
+
+            # ES0 exit window
+            pos["es0wi_try"] = {
+                "value": 5
+            }  # At 5mm, the middle of the window is 500 mm from the table (neutral position)
+        else:
+            # Exit window height
+            d = bl.ehWindow.center[1] - bl.fm.center[1]
+            es0wi_try = fm_beam_height - d * np.tan(-(2 * cfg["cm_pitch"] - 2 * cfg["fm_rotx"]))
+            pos["es0wi_try"] = {"value": es0wi_try}
+
+            # ES1 table height
+            d = bl.es1.center[1] - bl.fm.center[1]
+            es1_try = fm_beam_height - d * np.tan(-(2 * cfg["cm_pitch"] - 2 * cfg["fm_rotx"]))
+            pos["es1_try"] = {"value": es1_try}
+
+            # IC0 height
+            d = bl.es1ic0.center[1] - bl.fm.center[1]
+            es1ic0_try = (
+                fm_beam_height - d * np.tan(-(2 * cfg["cm_pitch"] - 2 * cfg["fm_rotx"])) - es1_try
+            )
+            pos["es1ic0_try"] = {"value": es1ic0_try}
+
+            # IC1 height
+            d = bl.es1ic1.center[1] - bl.fm.center[1]
+            es1ic1_try = (
+                fm_beam_height - d * np.tan(-(2 * cfg["cm_pitch"] - 2 * cfg["fm_rotx"])) - es1_try
+            )
+            pos["es1ic1_try"] = {"value": es1ic1_try}
+
+            # IC2 height
+            d = bl.es1ic2.center[1] - bl.fm.center[1]
+            es1ic2_try = (
+                fm_beam_height - d * np.tan(-(2 * cfg["cm_pitch"] - 2 * cfg["fm_rotx"])) - es1_try
+            )
+            pos["es1ic2_try"] = {"value": es1ic2_try}
+
+            # ES2 table height
+            d = bl.es2.center[1] - bl.fm.center[1]
+            es2_try = fm_beam_height - d * np.tan(-(2 * cfg["cm_pitch"] - 2 * cfg["fm_rotx"]))
+            pos["es2_try"] = {"value": es2_try}
+
+        return pos

debye_bec/bec_ipython_client/plugins/digital_twin/types.py

@@ -0,0 +1,83 @@
+"""Types used for the beamline config and for plotting data"""
+
+from enum import Enum
+from typing import TypedDict
+
+
+class BeamlineId(str, Enum):
+    """
+    Identifier for supported beamlines.
+    """
+
+    X01DA = "x01da"
+    X10DA = "x10da"
+
+
+class ConfigDict(TypedDict):
+    """
+    Typed dictionary representing the beamline configuration.
+
+    Attributes:
+        energy (float): Beam energy.
+        h_acc (float): Horizontal acceptance.
+        v_acc (float): Vertical acceptance.
+        cm_pitch (float): CM pitch angle.
+        cm_stripe (str): CM stripe name.
+        cm_trx (float): CM translation x.
+        mo1_mode (str): MO1 mode.
+        mo1_xtal (str): MO1 crystal.
+        mo1_bragg (float): MO1 Bragg angle.
+        fm_rotx (float): FM rotation x.
+        fm_stripe (str): FM stripe name.
+        fm_trx (float): FM translation x.
+        fm_qy (float): FM qy value.
+        fm_gain_height (int): FM gain height.
+        smpl (float): Sample value.
+    """
+
+    energy: float
+    h_acc: float
+    v_acc: float
+    cm_pitch: float
+    cm_stripe: str
+    cm_trx: float
+    mo1_mode: str
+    mo1_xtal: str
+    mo1_bragg: float
+    fm_rotx: float
+    fm_stripe: str
+    fm_trx: float
+    fm_qy: None | float
+    fm_gain_height: int
+    smpl: float
+
+
+class DataDict(TypedDict):
+    """
+    Typed dictionary representing plot data.
+
+    Attributes:
+        x (list[float]): List of x-axis values.
+        y (list[float]): List of y-axis values.
+    """
+
+    x: list
+    y: list
+
+
+class SurfaceDict(TypedDict):
+    """
+    Typed dictionary representing the surfaces of a scene,
+    grouping plot data by surface type.
+
+    Attributes:
+        cm (DataDict): Data for the cm surface.
+        mo1_1 (DataDict): Data for the mo1_1 surface.
+        mo1_2 (DataDict): Data for the mo1_2 surface.
+        fm (DataDict): Data for the fm surface.
+    """
+
+    cm: DataDict
+    mo1_1: DataDict
+    mo1_2: DataDict
+    fm: DataDict

debye_bec/bec_ipython_client/plugins/digital_twin/x01da_offsets.yaml

@@ -0,0 +1,50 @@
+cm_try:
+  offset: 0.15
+
+mo1_trx:
+  modifier:
+    axis: mo1_trx
+    range: [[-30, -0.1], [0.1, 30]]
+  offset: [-2.3, 1.31]
+
+mo1_try:
+  modifier:
+    axis: mo1_trx
+    range: [[-30, -0.1], [0.1, 30]]
+  offset: [-1.78, -1.78]
+
+sl1_centery:
+  offset: -1.2
+
+fm_trx:
+  modifier:
+    axis: fm_trx
+    range: [[-66, -31], [-24, 7], [11, 31], [38, 66]]
+  offset: [-0.61, 0, 0, -0.16]
+
+fm_try:
+  modifier:
+    axis: fm_trx
+    range: [[-66, -31], [-24, 7], [11, 31], [38, 66]]
+  offset: [0.028, 0, 0, -0.45]
+
+fm_rotx:
+  modifier:
+    axis: fm_trx
+    range: [[-66, -31], [-24, 7], [11, 31], [38, 66]]
+  offset: [0.027, 0, 0, 0.045]
+
+fm_roty:
+  modifier:
+    axis: fm_trx
+    range: [[-66, -31], [-24, 7], [11, 31], [38, 66]]
+  offset: [-0.038, 0, 0, -0.053]
+
+sl2_centery:
+  offset: -0.7
+
+ot_try:
+  offset: -0.49
+
+ot_rotx:
+  offset: 0

debye_bec/bec_ipython_client/plugins/digital_twin/x01da_parameters.py

@@ -0,0 +1,323 @@
+"""
+X01DA / Debye Beamline Parameters.
+This file describes the parameter of each component of the Debye beamline
+to be used for raytracing and geometrical calculations.
+"""
+
+from collections import namedtuple
+
+import numpy as np
+import xrt.backends.raycing.materials as rm
+
+# XRT definitions
+filterBeryl = rm.Material("Be", rho=1.85, kind="plate")  # pyright: ignore[reportArgumentType]
+filterDiamond = rm.Material("C", rho=3.52, kind="plate")  # pyright: ignore[reportArgumentType]
+filterGraphite = rm.Material("C", rho=2.266, kind="plate")  # pyright: ignore[reportArgumentType]
+
+stripeSi = rm.Material("Si", rho=2.33)  # pyright: ignore[reportArgumentType]
+stripePt = rm.Material("Pt", rho=21.45)  # pyright: ignore[reportArgumentType]
+stripeRh = rm.Material("Rh", rho=12.41)  # pyright: ignore[reportArgumentType]
+stripeCr = rm.Material("Cr", rho=7.14)  # pyright: ignore[reportArgumentType]
+stripePyrex = rm.Material(
+    "Si", rho=2.20
+)  # Use Si as bare element and the density of SiO2 # pyright: ignore[reportArgumentType]
+
+si111_1 = rm.CrystalSi(hkl=(1, 1, 1), tK=77)  # first xtal surface
+si311_1 = rm.CrystalSi(hkl=(3, 1, 1), tK=77)  # first xtal surface
+si333_1 = rm.CrystalSi(hkl=(3, 3, 3), tK=77)  # first xtal surface
+si511_1 = rm.CrystalSi(hkl=(5, 1, 1), tK=77)  # first xtal surface
+si111_2 = rm.CrystalSi(hkl=(1, 1, 1), tK=77)  # second xtal surface
+si311_2 = rm.CrystalSi(hkl=(3, 1, 1), tK=77)  # second xtal surface
+si333_2 = rm.CrystalSi(hkl=(3, 3, 3), tK=77)  # second xtal surface
+si511_2 = rm.CrystalSi(hkl=(5, 1, 1), tK=77)  # second xtal surface
+
+filterDiamond = rm.Material("C", rho=3.52, kind="plate")  # pyright: ignore[reportArgumentType]
+filterBe = rm.Material("Be", rho=1.85, kind="plate")  # pyright: ignore[reportArgumentType]
+filterSi3N4 = rm.Material(
+    ["Si", "N"], quantities=[3, 4], rho=3.44, kind="plate"
+)  # pyright: ignore[reportArgumentType]
+filterAl = rm.Material("Al", rho=2.69, kind="plate")  # pyright: ignore[reportArgumentType]
+filterGraphite = rm.Material("C", rho=2.266, kind="plate")  # pyright: ignore[reportArgumentType]
+
+# General parameters
+sourceHeight = 0
+
+# Synchrotron
+synchrotron = namedtuple(
+    "synchrotron", ["eE", "eI", "eEspread", "eEpsilonX", "eEpsilonZ", "betaX", "betaZ"]
+)
+
+sls1 = synchrotron(
+    eE=2.4, eI=0.4, eEspread=0.878e-3, eEpsilonX=5.63, eEpsilonZ=0.007, betaX=0.45, betaZ=14.4
+)
+
+sls2 = synchrotron(
+    eE=2.7, eI=0.4, eEspread=1.147e-3, eEpsilonX=0.156, eEpsilonZ=0.01, betaX=0.18, betaZ=4.6
+)
+
+# Source
+bendingMagnet = namedtuple("bendingMagnet", ["name", "center", "sync", "B0"])
+
+sls1_14t = bendingMagnet(name="FE-BM-SLS1-1.4T", center=(0, 0, 0), sync=sls1, B0=1.4)
+
+sls2_21t = bendingMagnet(name="FE-BM-SLS2-2.1T", center=(0, 0, 0), sync=sls2, B0=2.1)
+
+sls2_35t = bendingMagnet(name="FE-BM-SLS2-3.5T", center=(0, 0, 0), sync=sls2, B0=3.5)
+
+sls2_50t = bendingMagnet(name="FE-BM-SLS2-5.0T", center=(0, 0, 0), sync=sls2, B0=5.0)
+
+# FE slits
+fe_slits = namedtuple("slits", ["name", "center", "center1", "center2", "maxDivH", "maxDivV"])
+
+feSlits = fe_slits(
+    name="FE-SLITS",
+    center=(0, 6117, sourceHeight),
+    center1=(0, 5045, sourceHeight),
+    center2=(0, 5289.5, sourceHeight),
+    maxDivH=1.8e-3,
+    maxDivV=0.8e-3,
+)
+
+# FE Window
+filt = namedtuple(
+    "filt", ["name", "center", "pitch", "limPhysX", "limPhysY", "surface", "material", "thickness"]
+)
+
+feWindow = filt(
+    name="FE-WINDOW",
+    center=(0.0, 7020, sourceHeight),
+    pitch=np.pi / 2,
+    limPhysX=(-6, 6),
+    limPhysY=(-3.0, 3.0),
+    surface="None",
+    material=filterDiamond,
+    thickness=0.1,
+)
+feWindow = feWindow._replace(surface=f"CVD Diamond window {feWindow.thickness*1e3:0.0f} $\\mu$m")
+
+# Collimating mirror
+collimatingMirror = namedtuple(
+    "collimatingMirror",
+    [
+        "name",
+        "center",
+        "surface",
+        "material",
+        "limPhysX",
+        "limPhysY",
+        "limOptX",
+        "limOptY",
+        "R",
+        "pitch",
+        "jack1",
+        "jack2",
+        "jack3",
+        "tx1",
+        "tx2",
+    ],
+)
+
+cm = collimatingMirror(
+    name="FE-CM",
+    center=[0, 6890, sourceHeight],
+    surface=("Si", "Pt", "Rh"),
+    material=(stripeSi, stripePt, stripeRh),
+    limPhysX=(-34, 34),
+    limPhysY=(-600, 600),
+    limOptX=((-21, -7, 14), (-11, 11, 23)),
+    limOptY=((-500, -500, -500), (500, 500, 500)),
+    R=[3e6, 15e6],
+    pitch=[-5.0e-3, -0.0e-3],
+    jack1=[0.0, 7210.0, 0.0],  # Tripod X, Y, Z (global)
+    jack2=[-210.0, 8310.0, 0.0],
+    jack3=[210.0, 8310.0, 0.0],
+    tx1=[0.0, -575.5],  # X-Stage 1 [x, y] (local)
+    tx2=[0.0, 575],
+)  # X-Stage 2
+
+apertures = namedtuple("apertures", ["name", "center", "opening"])
+
+fePS = apertures(
+    name="FE-PS", center=[0, 8815, sourceHeight], opening=[-20.0, 20.0, -20.0 + 12.5, 20.0 + 12.5]
+)  # left, right, bottom, top
+
+opWbBsBlock = apertures(
+    name="OP-WB-BS-BLOCK", center=[0.0, 13860, sourceHeight], opening=[-18.0, 18.0, 25, 85.5]
+)  # left, right, bottom, top
+# opening=[-18., 18., 42, 76], # X10DA
+
+# Monochromator
+monochromator = namedtuple(
+    "monochromator",
+    [
+        "name",
+        "center",
+        "xtal",
+        "material1",
+        "material2",
+        "xtalWidth",
+        "xtalOffsetX",
+        "xtalLength1",
+        "xtalLength2",
+        "xtalGap",
+        "rotOffset",
+        "heightOffset",
+        "braggLim",
+        "jack1",
+        "jack2",
+        "jack3",
+        "tx",
+    ],
+)
+
+mo1 = monochromator(
+    name="OP-MO1",
+    center=[0.0, 11750, sourceHeight],
+    xtal=("Si311", "Si111"),
+    material1=(si311_1, si111_1),
+    material2=(si311_2, si111_2),
+    xtalWidth=(24, 24),
+    xtalOffsetX=(-21.2, 21.2),
+    xtalLength1=(55, 55),
+    xtalLength2=(105, 105),
+    xtalGap=(8, 8),
+    rotOffset=6,
+    heightOffset=8.5,
+    braggLim=[3.6, 33],
+    jack1=[0.0, 11350.0, 0.0],  # Tripod maybe not available!
+    jack2=[-400.0, 12350.0, 0.0],
+    jack3=[400.0, 12350.0, 0.0],
+    tx=0.0,
+)  # X-Stage [x]
+
+mo2 = monochromator(
+    name="OP-CCM2",
+    center=[0.0, 13250, sourceHeight],
+    xtal=("Si311", "Si111"),
+    material1=(si311_1, si111_1),
+    material2=(si311_2, si111_2),
+    xtalWidth=(24, 24),
+    xtalOffsetX=(-21, 21),
+    xtalLength1=(55, 55),
+    xtalLength2=(105, 105),
+    xtalGap=(8, 8),
+    rotOffset=6,
+    heightOffset=8.5,
+    braggLim=[3.6, 33],
+    jack1=[0.0, 13350.0, 0.0],  # Tripod maybe not available!
+    jack2=[-400.0, 14350.0, 0.0],
+    jack3=[400.0, 14350.0, 0.0],
+    tx=0.0,
+)  # X-Stage [x]
+
+# OP Slits
+op_slits = namedtuple("op_slits", ["name", "center"])
+
+opSlits1 = op_slits(name="OP-SLITS 1", center=(0, 14349.6, sourceHeight))
+
+opSlits2 = op_slits(name="OP-SLITS 2", center=(0, 18134.8, sourceHeight))
+
+# OP Beam Monitors
+op_bm = namedtuple("op_bm", ["name", "center"])
+
+opBM1 = op_bm(name="OP Beam Monitor 1", center=(0, 14599.6, sourceHeight))
+
+opBM2 = op_bm(name="OP Beam Monitor 2", center=(0, 18384.8, sourceHeight))
+
+# Focusing mirror
+focusingMirror = namedtuple(
+    "focusingMirror",
+    [
+        "name",
+        "center",
+        "surfaceToroid",
+        "materialToroid",
+        "surfaceFlat",
+        "materialFlat",
+        "limPhysXToroid",
+        "limPhysYToroid",
+        "limPhysXFlat",
+        "limPhysYFlat",
+        "limOptXToroid",
+        "limOptYToroid",
+        "limOptXFlat",
+        "limOptYFlat",
+        "R",
+        "pitch",
+        "r",
+        "xToroid",
+        "xFlat",
+        "hToroid",
+        "jack1",
+        "jack2",
+        "jack3",
+        "tx1",
+        "tx2",
+    ],
+)
+
+fm = focusingMirror(
+    name="OP-FM",
+    center=[0.0, 15670, sourceHeight],  # nominal height 58 mm above ring, SLS1!
+    surfaceToroid=("Rh", "Pt"),
+    materialToroid=(stripeRh, stripePt),
+    surfaceFlat=("Rh", "Pt"),
+    materialFlat=(stripeRh, stripePt),
+    limPhysXToroid=(-79.0, 79.0),
+    limPhysYToroid=(-575.0, 575.0),
+    limPhysXFlat=(-79.0, 79.0),
+    limPhysYFlat=(-575.0, 575.0),
+    limOptXToroid=((-38, 66), (-66, 31)),
+    limOptYToroid=((-500.0, -500.0), (500.0, 500.0)),
+    limOptXFlat=((-11.45, 23.55), (-30.45, -6.45)),
+    limOptYFlat=((-500.0, -500.0), (500.0, 500.0)),
+    R=[3e6, 15e6],
+    pitch=[-5.0e-3, 0e-3],
+    r=[35.510, 24.986],
+    xToroid=[-52, 48.5],  # offset in local x
+    xFlat=[-20.95, 8.55],
+    hToroid=[2.88, 7.15],  # depth of the cylinder at x = xCylinder1 and x = xCylinder2.
+    jack1=[-130.0, 15535 - 538.0, 0.0],
+    jack2=[130.0, 15535 + 538.0, 0.0],
+    jack3=[0.0, 15535 + 538.0, 0.0],
+    tx1=[0.0, -575.0],  # X-Stage 1 [x, y]
+    tx2=[0.0, 575.0],
+)  # X-Stage 2 [x, y]
+
+# EH Window
+ehWindow = filt(
+    name="EH-WINDOW",
+    center=(0.0, 19998.3, sourceHeight),
+    pitch=np.pi / 2,
+    limPhysX=(-20.0, 20.0),
+    limPhysY=(-4, 4),
+    surface="None",
+    material=filterSi3N4,
+    thickness=0.002,
+)
+ehWindow = ehWindow._replace(surface=f"Beryllium window {ehWindow.thickness*1e3:0.0f} $\\mu$m")
+
+# Sample
+sample = namedtuple("sample", ["name", "center"])
+
+smpl = sample(name="EH-SMPL", center=[0, 23365, sourceHeight])
+
+smpl2 = sample(name="EH-SMPL2", center=[0, 27500, sourceHeight])
+
+tables = {}
+
+# Vacuum pipes
+# DN40CF ID = 35 mm oder 37 mm
+# DN50CF ID = 47.5 mm
+# DN63CF ID = 60.2 mm oder 66 mm
+# DN100CF ID = 97.4 mm oder 104 mm
+pipe = namedtuple("pipes", ["center", "diameter", "start", "end"])
+vacuum_pipes = pipe(
+    center=[27.5, (37.5 + 27.5) / 2, 37.5, 62.5, 72.5],
+    diameter=[97.4, 97.4, 97.4, 97.4, 97.4],
+    start=[10952.88, 11750 + 250, mo2.center[1] + 250, 14000, fm.center[1]],
+    end=[11750 - 250, mo2.center[1] - 250, 14000, fm.center[1], ehWindow.center[1]],
+)
+
+Walls = namedtuple("walls", ["start", "end", "height"])
+walls = Walls(start=[13999.30], end=[13999 + 75.5 + 30], height=[[-20, 25]])

debye_bec/bec_ipython_client/plugins/digital_twin/x10da_offsets.yaml

@@ -0,0 +1,59 @@
+
+cm_try:
+  offset: -0.7
+
+mo1_try:
+  offset: -31.42
+
+mo1_trx:
+  modifier:
+    axis: mo1_trx
+    range: [[-30, -0.1], [0.1, 30]]
+  offset: [-4.3, 0]
+
+sl1_centery:
+  offset: -55.54
+
+bm1_try:
+  offset: 52.22
+
+fm_trx:
+  modifier:
+    axis: fm_trx
+    range: [[-100, -48], [-47, 0]]
+  offset: [-0.3, 0.52]
+
+fm_try:
+  modifier:
+    axis: fm_trx
+    range: [[-100, -48], [-47, 0]]
+  offset: [-42.56, -41.49]
+
+# pitch
+fm_rotx:
+  modifier:
+    axis: fm_trx
+    range: [[-100, -48], [-47, 0]]
+  offset: [1.30, 1.049]
+
+# yaw
+fm_roty:
+  modifier:
+    axis: fm_trx
+    range: [[-100, -48], [-47, 0]]
+  offset: [1.754, 1.924]
+
+bm2_try:
+  offset: -19
+
+es0wi_try:
+  offset: -71.98
+
+es1_try:
+  offset: -113.26
+
+es1ic1_try:
+  offset: 10.39
+
+es1ic2_try:
+  offset: 3.55

debye_bec/bec_ipython_client/plugins/digital_twin/x10da_parameters.py

@@ -0,0 +1,296 @@
+"""
+X10DA / SuperXAS Beamline Parameters.
+This file describes the parameter of each component of the SuperXAS beamline
+to be used for raytracing and geometrical calculations.
+"""
+
+from collections import namedtuple
+
+import numpy as np
+import xrt.backends.raycing.materials as rm
+
+# XRT definitions
+filterBeryl = rm.Material("Be", rho=1.85, kind="plate")  # pyright: ignore[reportArgumentType]
+filterDiamond = rm.Material("C", rho=3.52, kind="plate")  # pyright: ignore[reportArgumentType]
+filterGraphite = rm.Material("C", rho=2.266, kind="plate")  # pyright: ignore[reportArgumentType]
+
+stripeSi = rm.Material("Si", rho=2.33)  # pyright: ignore[reportArgumentType]
+stripePt = rm.Material("Pt", rho=21.45)  # pyright: ignore[reportArgumentType]
+stripeRh = rm.Material("Rh", rho=12.41)  # pyright: ignore[reportArgumentType]
+stripeCr = rm.Material("Cr", rho=7.14)  # pyright: ignore[reportArgumentType]
+stripePyrex = rm.Material(
+    "Si", rho=2.20
+)  # Use Si as bare element and the density of SiO2 # pyright: ignore[reportArgumentType]
+
+si111_1 = rm.CrystalSi(hkl=(1, 1, 1), tK=77)  # first xtal surface
+si311_1 = rm.CrystalSi(hkl=(3, 1, 1), tK=77)  # first xtal surface
+si333_1 = rm.CrystalSi(hkl=(3, 3, 3), tK=77)  # first xtal surface
+si511_1 = rm.CrystalSi(hkl=(5, 1, 1), tK=77)  # first xtal surface
+si111_2 = rm.CrystalSi(hkl=(1, 1, 1), tK=77)  # second xtal surface
+si311_2 = rm.CrystalSi(hkl=(3, 1, 1), tK=77)  # second xtal surface
+si333_2 = rm.CrystalSi(hkl=(3, 3, 3), tK=77)  # second xtal surface
+si511_2 = rm.CrystalSi(hkl=(5, 1, 1), tK=77)  # second xtal surface
+
+filterDiamond = rm.Material("C", rho=3.52, kind="plate")  # pyright: ignore[reportArgumentType]
+filterBe = rm.Material("Be", rho=1.85, kind="plate")  # pyright: ignore[reportArgumentType]
+filterSi3N4 = rm.Material(
+    ["Si", "N"], quantities=[3, 4], rho=3.44, kind="plate"
+)  # pyright: ignore[reportArgumentType]
+filterAl = rm.Material("Al", rho=2.69, kind="plate")  # pyright: ignore[reportArgumentType]
+filterGraphite = rm.Material("C", rho=2.266, kind="plate")  # pyright: ignore[reportArgumentType]
+
+# General parameters
+sourceHeight = 0
+
+# Synchrotron
+synchrotron = namedtuple(
+    "synchrotron", ["eE", "eI", "eEspread", "eEpsilonX", "eEpsilonZ", "betaX", "betaZ"]
+)
+
+sls1 = synchrotron(
+    eE=2.4, eI=0.4, eEspread=0.878e-3, eEpsilonX=5.63, eEpsilonZ=0.007, betaX=0.45, betaZ=14.4
+)
+
+sls2 = synchrotron(
+    eE=2.7, eI=0.4, eEspread=1.147e-3, eEpsilonX=0.156, eEpsilonZ=0.01, betaX=0.18, betaZ=4.6
+)
+
+# Source
+bendingMagnet = namedtuple("bendingMagnet", ["name", "center", "sync", "B0"])
+
+sls1_14t = bendingMagnet(name="FE-BM-SLS1-1.4T", center=(0, 0, 0), sync=sls1, B0=1.4)
+
+sls2_21t = bendingMagnet(name="FE-BM-SLS2-2.1T", center=(0, 0, 0), sync=sls2, B0=2.1)
+
+sls2_35t = bendingMagnet(name="FE-BM-SLS2-3.5T", center=(0, 0, 0), sync=sls2, B0=3.5)
+
+sls2_50t = bendingMagnet(name="FE-BM-SLS2-5.0T", center=(0, 0, 0), sync=sls2, B0=5.0)
+
+# FE slits
+fe_slits = namedtuple("slits", ["name", "center", "center1", "center2", "maxDivH", "maxDivV"])
+
+feSlits = fe_slits(
+    name="FE-SLITS",
+    center=(0, 6117, sourceHeight),
+    center1=(0, 5038.4, sourceHeight),
+    center2=(0, 5282.9, sourceHeight),
+    maxDivH=1.8e-3,
+    maxDivV=0.8e-3,
+)
+
+# Filters
+filt = namedtuple(
+    "filt", ["name", "center", "pitch", "limPhysX", "limPhysY", "surface", "material", "thickness"]
+)
+
+feWindow = filt(
+    name="FE-WINDOW",
+    center=(0.0, 6158, sourceHeight),
+    pitch=np.pi / 2,
+    limPhysX=(-6, 6),
+    limPhysY=(-3.0, 3.0),
+    surface="None",
+    material=filterDiamond,
+    thickness=0.1,
+)
+feWindow = feWindow._replace(
+    surface="CVD Diamond window {0:0.0f} $\\mu$m".format(feWindow.thickness * 1e3)
+)
+
+feFilt = filt(
+    name="FE-FI",
+    center=(0.0, 6590, sourceHeight),
+    pitch=np.pi / 2,
+    limPhysX=(-15, 15),
+    limPhysY=(-10, 10),
+    surface="None",
+    material=filterGraphite,
+    thickness=0.25,
+)
+feFilt = feFilt._replace(surface="Graphite filter {0:0.0f} $\\mu$m".format(feFilt.thickness * 1e3))
+
+# Collimating mirror
+collimatingMirror = namedtuple(
+    "collimatingMirror",
+    [
+        "name",
+        "center",
+        "surface",
+        "material",
+        "limPhysX",
+        "limPhysY",
+        "limOptX",
+        "limOptY",
+        "R",
+        "pitch",
+        "jack1",
+        "jack2",
+        "jack3",
+        "tx1",
+        "tx2",
+    ],
+)
+
+cm = collimatingMirror(
+    name="FE-CM",
+    center=[0, 7560.8, sourceHeight],
+    surface=("Pt", "Si", "Rh"),
+    material=(stripePt, stripeSi, stripeRh),
+    limPhysX=(-30, 30),
+    limPhysY=(-600, 600),
+    limOptX=((-21, -0.5, 11), (-4, 9.5, 23)),
+    limOptY=((-500, -500, -500), (500, 500, 500)),
+    R=[3e6, 15e6],
+    pitch=[1.4e-3, 4.5e-3],
+    jack1=[0.0, 7210.0, 0.0],  # Tripod X, Y, Z (global)
+    jack2=[-210.0, 8310.0, 0.0],
+    jack3=[210.0, 8310.0, 0.0],
+    tx1=[0.0, -575.5],  # X-Stage 1 [x, y] (local)
+    tx2=[0.0, 575],
+)  # X-Stage 2
+
+apertures = namedtuple("apertures", ["name", "center", "opening"])
+
+fePS = apertures(
+    name="FE-PS", center=[0, 8760, sourceHeight], opening=[-39 / 2, 39 / 2, -10, 29]
+)  # left, right, bottom, top
+
+opWbBsBlock = apertures(
+    name="OP-WB-BS-BLOCK", center=[0.0, 13606 - 135, sourceHeight], opening=[-18.0, 18.0, 42, 76]
+)  # left, right, bottom, top
+
+opSlits1 = apertures(
+    name="OP-SLITS 1", center=[0, 14145 - 135, sourceHeight], opening=[-35 / 2, 35 / 2, 47.5, 82.5]
+)
+
+# OP Beam Monitors
+op_bm = namedtuple("op_bm", ["name", "center"])
+
+opBM1 = op_bm(name="OP Beam Monitor 1", center=(0, 14525 - 135, sourceHeight))
+
+opBM2 = op_bm(name="OP Beam Monitor 2", center=(0, 17161.6 - 135, sourceHeight))
+
+# Monochromator
+monochromator = namedtuple(
+    "monochromator",
+    [
+        "name",
+        "center",
+        "xtal",
+        "material1",
+        "material2",
+        "xtalWidth",
+        "xtalOffsetX",
+        "xtalLength1",
+        "xtalLength2",
+        "xtalGap",
+        "rotOffset",
+        "heightOffset",
+        "braggLim",
+        "jack1",
+        "jack2",
+        "jack3",
+        "tx",
+    ],
+)
+
+mo1 = monochromator(
+    name="OP-CCM1",
+    center=[0.0, 11670 - 135, sourceHeight],
+    xtal=("Si311", "Si111"),
+    material1=(si311_1, si111_1),
+    material2=(si311_2, si111_2),
+    xtalWidth=(20, 20),
+    xtalOffsetX=(19.2, -19.2),
+    xtalLength1=(60, 60),
+    xtalLength2=(60, 60),
+    xtalGap=(8, 8),
+    rotOffset=6,  # not sure what it is
+    heightOffset=8.5,  # not sure what it is
+    braggLim=[4, 35],
+    jack1=[0.0, 11350.0, 0.0],  # Tripod not available!
+    jack2=[-400.0, 12350.0, 0.0],
+    jack3=[400.0, 12350.0, 0.0],
+    tx=0.0,
+)  # X-Stage [x]
+
+# Focusing mirror
+focusingMirror = namedtuple(
+    "focusingMirror",
+    [
+        "name",
+        "center",
+        "surfaceToroid",
+        "materialToroid",
+        "limPhysXToroid",
+        "limPhysYToroid",
+        "limOptXToroid",
+        "limOptYToroid",
+        "R",
+        "pitch",
+        "r",
+        "xToroid",
+        "hToroid",
+        "jack1",
+        "jack2",
+        "jack3",
+        "tx1",
+        "tx2",
+    ],
+)
+
+OFFSET_TRX = 46.8735
+
+fm = focusingMirror(
+    name="OP-FM",
+    center=[0.0, 15580 - 135, sourceHeight],
+    surfaceToroid=("Rh", "Pt"),
+    materialToroid=(stripeRh, stripePt),
+    limPhysXToroid=(-54.0, 54.0),
+    limPhysYToroid=(-565.0, 565.0),
+    limOptXToroid=(
+        (43.388 + OFFSET_TRX, -4.865 + OFFSET_TRX),
+        (4.865 + OFFSET_TRX, -40.882 + OFFSET_TRX),
+    ),
+    limOptYToroid=((-500.0, -500.0), (500.0, 500.0)),
+    R=[3e6, 15e6],
+    pitch=[1.4e-3, 4.5e-3],
+    r=[30, 20],
+    xToroid=[24.126 + OFFSET_TRX, -22 + OFFSET_TRX],  # offset in local x
+    hToroid=[7.0, 11.3],  # depth of the cylinder at x = xCylinder1 and x = xCylinder2.
+    jack1=[0.0, 14980.0, 0.0],
+    jack2=[-75.0, 16180.0, 0.0],
+    jack3=[75.0, 16180.0, 0.0],
+    tx1=[0.0, -575.0],  # X-Stage 1 [x, y]
+    tx2=[0.0, 575.0],
+)  # X-Stage 2 [x, y]
+
+# Entry wall experimental hutch: 21593 mm from source (SLS2)
+
+# Exit window
+ehWindow = filt(
+    name="EH-WINDOW",
+    center=(0.0, 22063, sourceHeight),
+    pitch=np.pi / 2,
+    limPhysX=(-10.0, 10.0),
+    limPhysY=(17.5, 92.5),
+    surface="None",
+    material=filterBe,
+    thickness=0.25,
+)
+ehWindow = ehWindow._replace(
+    surface="Beryllium window {0:0.0f} $\\mu$m".format(ehWindow.thickness * 1e3)
+)
+
+# Sample
+sample = namedtuple("sample", ["name", "center"])
+
+es1 = sample(name="ES1", center=[0, 23823, sourceHeight])
+es2 = sample(name="ES2", center=[0, 25843, sourceHeight])
+
+# Ionization chambers
+ic = namedtuple("sample", ["name", "center"])
+
+es1ic0 = ic(name="ES1 IC0", center=[0, 23633, sourceHeight])
+es1ic1 = ic(name="ES1 IC1", center=[0, 24383, sourceHeight])
+es1ic2 = ic(name="ES1 IC2", center=[0, 24723, sourceHeight])

debye_bec/bec_ipython_client/startup/post_startup.py

@@ -34,3 +34,14 @@ to setup the prompts.
 """
 
 # pylint: disable=invalid-name, unused-import, import-error, undefined-variable, unused-variable, unused-argument, no-name-in-module
+
+from bec_lib import bec_logger
+
+logger = bec_logger.logger
+
+logger.info("Using the Debye startup script.")
+
+from debye_bec.bec_ipython_client.plugins.digital_twin.digital_twin import DigitalTwinCore
+
+digital_twin = DigitalTwinCore()
+logger.success("Digital Twin Core loaded. Use 'digital_twin' to access it.")

debye_bec/bec_widgets/widgets/digital_twin/calculations/calc_positions.py

@@ -1,297 +0,0 @@
-"""
-Calculates the positions of axes based on a beamline config
-"""
-
-import numpy as np
-from bec_lib import bec_logger
-
-from .. import parameters as bl
-from ..types import BeamlineId, ConfigDict
-
-logger = bec_logger.logger
-
-
-def calc_positions(beamline: BeamlineId, cfg: ConfigDict) -> dict[str, dict[str, float]]:
-    """
-    Calculates the positions of axes based on a beamline config.
-
-    Args:
-        cfg(ConfigDict): Dictionary with beamline config
-
-    Returns:
-        dict[str, dict[str, float]]: Dictionary mapping device names to dictionaries
-            containing a "value" key with the corresponding float value (position).
-    """
-
-    pos = {}
-
-    ## FE slits
-    trxr = -np.arctan(cfg["h_acc"]) * bl.feSlits.center1[1]
-    trxw = (
-        (np.arctan(cfg["h_acc"]) * bl.feSlits.center1[1])
-        / bl.feSlits.center1[1]
-        * bl.feSlits.center2[1]
-    )
-    tryb = -np.arctan(cfg["v_acc"]) * bl.feSlits.center1[1]
-    tryt = (
-        (np.arctan(cfg["v_acc"]) * bl.feSlits.center1[1])
-        / bl.feSlits.center1[1]
-        * bl.feSlits.center2[1]
-    )
-
-    xgap = trxw - trxr
-    ygap = tryt - tryb
-
-    pos["sldi_gapx"] = {"value": xgap}
-    pos["sldi_gapy"] = {"value": ygap}
-
-    ## Collimating Mirror
-    obj_dist = bl.cm.center[1]  # object distance
-    beam_vs = 2 * obj_dist * np.tan(cfg["v_acc"])  # vertical size of beam after CM
-
-    # TRX
-    if cfg["cm_stripe"] in bl.cm.surface:
-        index = bl.cm.surface.index(cfg["cm_stripe"])
-    else:
-        raise ValueError(f"Requested stripe {cfg['cm_stripe']} not found in parameters!")
-    cm_trx = -(bl.cm.limOptX[0][index] + bl.cm.limOptX[1][index]) / 2
-    pos["cm_trx"] = {"value": cm_trx}
-
-    # TRY
-    height = obj_dist * np.tan(cfg["v_acc"]) ** 2 * 1 / np.tan(cfg["cm_pitch"])
-    pos["cm_try"] = {"value": height}
-
-    # Pitch
-    pos["cm_rotx"] = {
-        "value": -cfg["cm_pitch"] * 1e3
-    }  # invert and convert to mrad (same as EGU of rotx axis)
-
-    # Bending Radius
-    radius = (
-        2.0 * obj_dist / np.sin(cfg["cm_pitch"])
-    )  # Elements of modern X-ray Physics, page 108 ff.
-    pos["cm_bnd_radius"] = {"value": radius * 1e-6}  # Convert to km
-
-    ## Monochromator
-    if cfg["mo1_mode"] == "Monochromatic":
-        # Add 2x CM pitch to the bragg angle
-        bragg = cfg["mo1_bragg"]
-    elif cfg["mo1_mode"] == "Pinkbeam":
-        # Align xtal surfaces parallel to beam
-        bragg = 0
-    else:
-        raise ValueError("Monochromator mode not supported")
-    pos["mo1_bragg_angle"] = {"value": bragg / np.pi * 180}  # Bragg angle in deg
-
-    # TRY, Height
-    l = bl.mo1.xtalGap[0] / np.sin(cfg["mo1_bragg"])
-    yhor = l * np.cos(2.0 * (cfg["mo1_bragg"] + cfg["cm_pitch"]))
-    yver = yhor * np.tan(2.0 * cfg["cm_pitch"])
-
-    if cfg["mo1_mode"] == "Monochromatic":
-        beam_offset_mo1 = (
-            l * np.sin(2.0 * (cfg["mo1_bragg"] + cfg["cm_pitch"])) - yver
-        )  # Resultat ist korrekt!
-    elif cfg["mo1_mode"] == "Pinkbeam":
-        beam_offset_mo1 = 0
-    else:
-        raise ValueError("Monochromator mode not supported")
-
-    def csc(a):
-        return 1 / np.sin(a)
-
-    def cot(a):
-        return 1 / np.tan(a)
-
-    # calculate height of center of first crystal surface
-    f = bl.mo1.rotOffset  # rotation offset, mm
-    d = bl.mo1.heightOffset  # xtal height offset, mm
-    c = d * csc(cfg["mo1_bragg"]) - f * cot(cfg["mo1_bragg"])
-
-    # Calculate height of center of rotation
-    b = np.sqrt(
-        d**2 * csc(cfg["mo1_bragg"]) ** 2
-        - 2 * d * f * cot(cfg["mo1_bragg"]) * csc(cfg["mo1_bragg"])
-        + f**2 * cot(cfg["mo1_bragg"]) ** 2
-        + f**2
-    )
-    h = np.cos(np.pi / 2 - np.arctan(f / c) - cfg["mo1_bragg"] - 2 * cfg["cm_pitch"]) * b
-    h2 = ((bl.mo1.center[1] - bl.cm.center[1]) - np.sqrt(b**2 - h**2)) * np.tan(2 * cfg["cm_pitch"])
-    height_mo1_real = (
-        h + h2
-    )  # per design, the height should not change if the pitch of the CM is not changed!
-    if cfg["mo1_mode"] == "Monochromatic":
-        pass
-    elif cfg["mo1_mode"] == "Pinkbeam":
-        height_mo1_real = (
-            height_mo1_real - 13
-        )  # Move down to let beam pass between both crystal without touching copper cooler
-    else:
-        raise ValueError("Monochromator mode not supported")
-    pos["mo1_try"] = {"value": height_mo1_real}
-
-    # TRX, Crystal selection
-    if cfg["mo1_mode"] == "Monochromatic":
-        xtal = cfg["mo1_xtal"].translate(
-            str.maketrans("", "", "()")
-        )  # Remove brackets from xtal name to conform with parameters
-        if xtal in bl.mo1.xtal:
-            index = bl.mo1.xtal.index(xtal)
-        else:
-            raise ValueError(f"Requested xtal {xtal} not found in parameters!")
-        pos["mo1_trx"] = {"value": bl.mo1.xtalOffsetX[index]}
-    else:
-        pos["mo1_trx"] = {"value": 0}
-
-    diag = bl.mo1.xtalGap[0] / np.sin(cfg["mo1_bragg"])  # Calculations for Mono
-    dz = diag * np.cos(2 * (cfg["cm_pitch"] + cfg["mo1_bragg"]))
-
-    ## Slits 1
-    d = bl.opSlits1.center[1] - bl.cm.center[1] - dz
-    sl1_beam_height = d * np.tan(2 * cfg["cm_pitch"]) + beam_offset_mo1
-    pos["sl1_centery"] = {"value": sl1_beam_height}
-    pos["sl1_gapy"] = {"value": beam_vs}
-
-    ## Beam Monitor 1
-    d = bl.opBM1.center[1] - bl.cm.center[1] - dz
-    bm1_beam_height = d * np.tan(2 * cfg["cm_pitch"]) + beam_offset_mo1
-    pos["bm1_try"] = {"value": bm1_beam_height}
-
-    ## Focusing Mirror
-    p = bl.fm.center[1]
-    q = cfg["smpl"] - bl.fm.center[1]
-    f = (p * q) / (p + q)  # focal length
-
-    # Bender radius
-    if cfg["fm_qy"] is None:
-        radius = 2 * q / np.sin(cfg["fm_rotx"])  # ideal bending radius for focused beam
-    else:
-        radius = (
-            2 * cfg["fm_qy"] / np.sin(cfg["fm_rotx"])
-        )  # ideal bending radius for unfocused beam
-    pos["fm_bnd_radius"] = {"value": radius * 1e-6}  # Convert to km
-
-    # Pitch
-    d = bl.fm.center[1] - bl.cm.center[1] - dz
-    fm_rotx = (
-        2 * cfg["cm_pitch"] - cfg["fm_rotx"]
-    )  # calculate pitch in absolute values (according to horizontal plane)
-    pos["fm_rotx"] = {
-        "value": -fm_rotx * 1e3
-    }  # invert and convert to mrad (same as EGU of rotx axis)
-
-    if cfg["fm_stripe"] in ("Rh (toroid)", "Pt (toroid)"):
-
-        # TRY
-        if cfg["fm_stripe"] == "Rh (toroid)":
-            r = bl.fm.r[0]
-            h_cyl = bl.fm.hToroid[0]
-        else:  # PT toroid
-            r = bl.fm.r[1]
-            h_cyl = bl.fm.hToroid[1]
-        width_beam = 2 * bl.fm.center[1] * np.tan(cfg["h_acc"] * 1e-3)
-        alpha = np.arccos(1 - width_beam**2 / (2 * r**2))
-        h = r - (r * np.cos(alpha / 2))
-        fm_beam_height = (d * np.tan(2 * cfg["cm_pitch"]) + beam_offset_mo1) * cfg["fm_gain_height"]
-        fm_height = (d * np.tan(2 * cfg["cm_pitch"]) + beam_offset_mo1 - h_cyl + h / 2) * cfg[
-            "fm_gain_height"
-        ]
-        pos["fm_try"] = {"value": fm_height}
-
-        # TRX
-        if cfg["fm_stripe"] == "Rh (toroid)":
-            x_cyl = -bl.fm.xToroid[0]
-        else:
-            x_cyl = -bl.fm.xToroid[1]
-        pos["fm_trx"] = {"value": x_cyl}
-
-    elif cfg["fm_stripe"] in ("Rh (flat)", "Pt (flat)"):
-
-        # TRY
-        fm_height = (d * np.tan(2 * cfg["cm_pitch"]) + beam_offset_mo1) * cfg["fm_gain_height"]
-        fm_beam_height = fm_height
-        pos["fm_try"] = {"value": fm_height}
-
-        # TRX
-        if cfg["fm_stripe"] == "Rh (flat)":
-            x_flat = -bl.fm.xFlat[0]
-        else:
-            x_flat = -bl.fm.xFlat[1]
-        pos["fm_trx"] = {"value": x_flat}
-
-    else:
-        raise ValueError("FM Stripe selection not valid")
-
-    pos["fm_roty"] = {"value": 0}
-    pos["fm_rotz"] = {"value": 0}
-
-    ## Slits 2
-    if hasattr(bl, "opSlits2"):
-        d = bl.opSlits2.center[1] - bl.fm.center[1]
-        sl2_beam_height = fm_beam_height - d * np.tan(-(2 * cfg["cm_pitch"] - 2 * cfg["fm_rotx"]))
-        pos["sl2_centery"] = {"value": sl2_beam_height}
-        pos["sl2_gapy"] = {"value": beam_vs}
-
-    ## Beam Monitor 2
-    d = bl.opBM2.center[1] - bl.fm.center[1]
-    bm2_beam_height = fm_beam_height - d * np.tan(-(2 * cfg["cm_pitch"] - 2 * cfg["fm_rotx"]))
-    pos["bm2_try"] = {"value": bm2_beam_height}
-
-    ## Optical Table
-
-    if beamline == "x01da":
-        # TRY
-        d = bl.ehWindow.center[1] - bl.fm.center[1]
-        ot_height = fm_beam_height - d * np.tan(-(2 * cfg["cm_pitch"] - 2 * cfg["fm_rotx"]))
-        pos["ot_try"] = {"value": ot_height}
-
-        # Pitch
-        ot_pitch = -(2 * cfg["cm_pitch"] - 2 * cfg["fm_rotx"])
-        pos["ot_rotx"] = {"value": ot_pitch * 1e3}
-
-        # TRZ ES1
-        ot_es1_trz = cfg["smpl"]
-        pos["ot_es1_trz"] = {"value": ot_es1_trz}
-
-        # ES0 exit window
-        pos["es0wi_try"] = {
-            "value": 5
-        }  # At 5mm, the middle of the window is 500 mm from the table (neutral position)
-    else:
-        # Exit window height
-        d = bl.ehWindow.center[1] - bl.fm.center[1]
-        es0wi_try = fm_beam_height - d * np.tan(-(2 * cfg["cm_pitch"] - 2 * cfg["fm_rotx"]))
-        pos["es0wi_try"] = {"value": es0wi_try}
-
-        # ES1 table height
-        d = bl.es1.center[1] - bl.fm.center[1]
-        es1_try = fm_beam_height - d * np.tan(-(2 * cfg["cm_pitch"] - 2 * cfg["fm_rotx"]))
-        pos["es1_try"] = {"value": es1_try}
-
-        # IC0 height
-        d = bl.es1ic0.center[1] - bl.fm.center[1]
-        es1ic0_try = (
-            fm_beam_height - d * np.tan(-(2 * cfg["cm_pitch"] - 2 * cfg["fm_rotx"])) - es1_try
-        )
-        pos["es1ic0_try"] = {"value": es1ic0_try}
-
-        # IC1 height
-        d = bl.es1ic1.center[1] - bl.fm.center[1]
-        es1ic1_try = (
-            fm_beam_height - d * np.tan(-(2 * cfg["cm_pitch"] - 2 * cfg["fm_rotx"])) - es1_try
-        )
-        pos["es1ic1_try"] = {"value": es1ic1_try}
-
-        # IC2 height
-        d = bl.es1ic2.center[1] - bl.fm.center[1]
-        es1ic2_try = (
-            fm_beam_height - d * np.tan(-(2 * cfg["cm_pitch"] - 2 * cfg["fm_rotx"])) - es1_try
-        )
-        pos["es1ic2_try"] = {"value": es1ic2_try}
-
-        # ES2 table height
-        d = bl.es2.center[1] - bl.fm.center[1]
-        es2_try = fm_beam_height - d * np.tan(-(2 * cfg["cm_pitch"] - 2 * cfg["fm_rotx"]))
-        pos["es2_try"] = {"value": es2_try}
-
-    return pos

debye_bec/bec_widgets/widgets/digital_twin/digital_twin.py

@@ -3,7 +3,6 @@ Digital Twin: Custom BEC widget to support the beamline alignment.
 """
 
 import sys
-from pathlib import Path
 from typing import Literal, cast
 
 import numpy as np
@@ -35,9 +34,9 @@ from qtpy.QtWidgets import (
     QWidget,
 )
 
+from ....bec_ipython_client.plugins.digital_twin.digital_twin import DigitalTwinCore
 from ..edge_selector import EdgeSelector
 from .beamline import get_beamline_id
-from .calculations.calc_positions import calc_positions
 from .calculations.calc_sideview import calc_sideview
 from .calculations.calc_surfaces import calc_surfaces
 from .calculations.calc_varia import (
@@ -55,6 +54,7 @@ from .calculations.calc_varia import (
     sldi_gap_to_acc,
     table_to_smpl_pos,
 )
+from .offsets import Offsets
 from .panels.input_panel import InputPanel
 from .panels.mover_panel import MoverPanel
 from .panels.plots import SideviewPlot, SurfacePlots
@@ -64,9 +64,6 @@ from .widgets.qt_widgets import ComboBox, InputNumberField
 
 logger = bec_logger.logger
 
-OFFSET_FILE_X01DA = Path(__file__).with_name("x01da_offsets.yaml")
-OFFSET_FILE_X10DA = Path(__file__).with_name("x10da_offsets.yaml")
-
 X01DA_E_MIN = 4500
 X01DA_E_MAX = 60000
 X10DA_E_MIN = 4500
@@ -85,16 +82,13 @@ class DigitalTwin(BECWidget, QWidget):
         super().__init__(parent=parent, *arg, **kwargs)
         self.get_bec_shortcuts()
 
+        self.core = DigitalTwinCore()
+
         self.beamline = get_beamline_id()
         # Debugging, override beamline!
         # self.beamline = BeamlineId.X10DA
 
-        self.offset_file = Path()
-        match self.beamline:
-            case "x01da":
-                self.offset_file = OFFSET_FILE_X01DA
-            case "x10da":
-                self.offset_file = OFFSET_FILE_X10DA
+        self.offsets = Offsets()
 
         # Check if devices are all in config
         self.check_bec_config()
@@ -172,7 +166,6 @@ class DigitalTwin(BECWidget, QWidget):
         self.edge_selector_energy = 0.0
         self.bragg_angle = 0.0
         self.qy = 0.0
-        self.offsets = {}
 
         # Initialize all values
         self.load_offsets(recalculate=False)
@@ -434,16 +427,7 @@ class DigitalTwin(BECWidget, QWidget):
 
         # Apply offsets
         if apply_offset:
-            for axis, _ in config.items():
-                if axis in self.offsets:
-                    axis_offsets = self.offsets[axis]
-                    if "modifier" in axis_offsets and "offset" in axis_offsets:
-                        for idx, rng in enumerate(axis_offsets["modifier"]["range"]):
-                            if rng[0] < config[axis_offsets["modifier"]["axis"]] < rng[1]:
-                                config[axis] += axis_offsets["offset"][idx]
-                                break
-                    elif "offset" in axis_offsets:
-                        config[axis] += axis_offsets["offset"]
+            config = self.offsets.apply_offsets(config, nested_config=False)
 
         # Convert to SI units!
         config["h_acc"] *= 1e-3
@@ -592,16 +576,7 @@ class DigitalTwin(BECWidget, QWidget):
             pos["ot_es1_trz"] = self.dev.ot_es1_trz.read(cached=True)["ot_es1_trz"]["value"]
 
         # Removing offsets
-        for axis, _ in pos.items():
-            if axis in self.offsets:
-                axis_offsets = self.offsets[axis]
-                if "modifier" in axis_offsets and "offset" in axis_offsets:
-                    for idx, rng in enumerate(axis_offsets["modifier"]["range"]):
-                        if rng[0] < pos[axis_offsets["modifier"]["axis"]] < rng[1]:
-                            pos[axis] -= axis_offsets["offset"][idx]
-                            break
-                elif "offset" in axis_offsets:
-                    pos[axis] -= axis_offsets["offset"]
+        pos = self.offsets.remove_offsets(pos)
 
         self.input.energy.set_number(self.dev.mo1_bragg.read(cached=True)["mo1_bragg"]["value"])
         h_acc, v_acc = sldi_gap_to_acc(pos["sldi_gapx"], pos["sldi_gapy"])
@@ -690,19 +665,9 @@ class DigitalTwin(BECWidget, QWidget):
                                Defaults to True
         """
 
-        if self.offsets == {}:
-            # Load offsets
-            if not self.offset_file.exists():
-                raise FileNotFoundError(f"Offset file not found: {self.offset_file}")
-
-            with self.offset_file.open("r", encoding="utf-8") as f:
-                data = yaml.safe_load(f)
-
-            if not isinstance(data, dict):
-                raise ValueError(f"Expected a YAML mapping, got {type(data).__name__}")
-
-            self.offsets = data
-
+        self.offsets.load_offsets()
+        if self.offsets.offsets != {}:
+            # Offsets were loaded
             if recalculate:
                 self.calc_assistant(identifier="init")
 
@@ -711,8 +676,7 @@ class DigitalTwin(BECWidget, QWidget):
             self.settings.offsets_status.setColor(get_accent_colors().success.name())
             self.settings.show_offsets.enable_button(True)
         else:
-            # Unload offsets
-            self.offsets = {}
+            # Offsets were unloaded
             self.calc_assistant(identifier="init")
 
             self.settings.load_offsets.setText("Load")
@@ -736,7 +700,7 @@ class DigitalTwin(BECWidget, QWidget):
         intro_label.setWordWrap(True)
         layout.addWidget(intro_label)
 
-        file = QLabel(str(self.offset_file))
+        file = QLabel(str(self.offsets.offset_file))
         file.setWordWrap(True)
         font = QFont()
         font.setItalic(True)
@@ -753,7 +717,9 @@ class DigitalTwin(BECWidget, QWidget):
             def represent_sequence(self, tag, sequence, *_):
                 return super().represent_sequence(tag, sequence, flow_style=True)
 
-        text_edit.setPlainText(yaml.dump(self.offsets, Dumper=InlineListDumper, sort_keys=False))
+        text_edit.setPlainText(
+            yaml.dump(self.offsets.offsets, Dumper=InlineListDumper, sort_keys=False)
+        )
         layout.addWidget(text_edit)
 
         buttons = QDialogButtonBox(QDialogButtonBox.StandardButton.Close)
@@ -886,19 +852,13 @@ class DigitalTwin(BECWidget, QWidget):
         """
         Calculates the positions for the axes based on the assistant values
         """
-        out = calc_positions(self.beamline, self.get_assistant_config())
+        config = self.get_assistant_config()
+        out = self.core.calc_positions(self.beamline, config)
+        out = self.core.apply_offsets(out, nested_config=True)
+        # out = calc_positions(self.beamline, self.get_assistant_config())
 
         # Apply offsets
-        for axis, axis_data in out.items():
-            if axis in self.offsets:
-                axis_offsets = self.offsets[axis]
-                if "modifier" in axis_offsets and "offset" in axis_offsets:
-                    for idx, rng in enumerate(axis_offsets["modifier"]["range"]):
-                        if rng[0] < out[axis_offsets["modifier"]["axis"]]["value"] < rng[1]:
-                            axis_data["value"] += axis_offsets["offset"][idx]
-                            break
-                elif "offset" in axis_offsets:
-                    axis_data["value"] += axis_offsets["offset"]
+        # out = self.offsets.apply_offsets(out, nested_config=True)
 
         self.mover.sldi_gapx.set_target(out["sldi_gapx"]["value"])
         self.mover.sldi_gapy.set_target(out["sldi_gapy"]["value"])

debye_bec/bec_widgets/widgets/digital_twin/offsets.py

@@ -0,0 +1,83 @@
+"""
+Offset class to load or unload offsets from a file
+"""
+
+from pathlib import Path
+
+import yaml
+from bec_lib import bec_logger
+
+from .beamline import get_beamline_id
+
+OFFSET_FILE_X01DA = Path(__file__).with_name("x01da_offsets.yaml")
+OFFSET_FILE_X10DA = Path(__file__).with_name("x10da_offsets.yaml")
+
+logger = bec_logger.logger
+
+
+class Offsets:
+
+    def __init__(self, *arg, **kwargs):
+        self.beamline = get_beamline_id()
+        self.offset_file = Path()
+        match self.beamline:
+            case "x01da":
+                self.offset_file = OFFSET_FILE_X01DA
+            case "x10da":
+                self.offset_file = OFFSET_FILE_X10DA
+        self.offsets = {}
+
+    def load_offsets(self):
+        if self.offsets == {}:
+            logger.info("Load beamline offsets")
+            if not self.offset_file.exists():
+                raise FileNotFoundError(f"Offset file not found: {self.offset_file}")
+
+            with self.offset_file.open("r", encoding="utf-8") as f:
+                data = yaml.safe_load(f)
+
+            if not isinstance(data, dict):
+                raise ValueError(f"Expected a YAML mapping, got {type(data).__name__}")
+
+            self.offsets = data
+        else:
+            logger.info("Unload beamline offsets")
+            self.offsets = {}
+
+    def apply_offsets(self, config, nested_config=False):
+        for axis, axis_data in config.items():
+            if axis in self.offsets:
+                axis_offsets = self.offsets[axis]
+                if "modifier" in axis_offsets and "offset" in axis_offsets:
+                    for idx, rng in enumerate(axis_offsets["modifier"]["range"]):
+                        modifier_axis = axis_offsets["modifier"]["axis"]
+                        modifier_value = (
+                            config[modifier_axis]["value"]
+                            if nested_config
+                            else config[modifier_axis]
+                        )
+                        if rng[0] < modifier_value < rng[1]:
+                            if nested_config:
+                                axis_data["value"] += axis_offsets["offset"][idx]
+                            else:
+                                config[axis] += axis_offsets["offset"][idx]
+                            break
+                elif "offset" in axis_offsets:
+                    if nested_config:
+                        axis_data["value"] += axis_offsets["offset"]
+                    else:
+                        config[axis] += axis_offsets["offset"]
+        return config
+
+    def remove_offsets(self, config):
+        for axis, _ in config.items():
+            if axis in self.offsets:
+                axis_offsets = self.offsets[axis]
+                if "modifier" in axis_offsets and "offset" in axis_offsets:
+                    for idx, rng in enumerate(axis_offsets["modifier"]["range"]):
+                        if rng[0] < config[axis_offsets["modifier"]["axis"]] < rng[1]:
+                            config[axis] -= axis_offsets["offset"][idx]
+                            break
+                elif "offset" in axis_offsets:
+                    config[axis] -= axis_offsets["offset"]
+        return config