fix: add pytest-redis to a list of dev deps

.git_hooks/post-commit

@@ -1,3 +0,0 @@
-SCRIPT_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd )
-semantic-release changelog -D version_variable=$SCRIPT_DIR/../../semantic_release/__init__.py:__version__
-semantic-release version -D version_variable=$SCRIPT_DIR/../../semantic_release/__init__.py:__version__

.git_hooks/pre-commit

@@ -1,3 +0,0 @@
-black --line-length=100 $(git diff --cached --name-only --diff-filter=ACM -- '*.py')
-isort --line-length=100 --profile=black --multi-line=3 --trailing-comma $(git diff --cached --name-only --diff-filter=ACM -- '*.py')
-git add $(git diff --cached --name-only --diff-filter=ACM -- '*.py')

.gitignore

@@ -8,9 +8,6 @@
 **/.pytest_cache
 **/*.egg*
 
-# recovery_config files
-recovery_config_*
-
 # file writer data
 **.h5
 

.gitlab-ci.yml

@@ -1,7 +1,40 @@
-include:
-  - project: bec/awi_utils
-    file: /templates/plugin-repo-template.yml
-    inputs:
-      name: "debye"
-      target: "debye_bec"
-      branch: $CHILD_PIPELINE_BRANCH
+# This file is a template, and might need editing before it works on your project.
+# Official language image. Look for the different tagged releases at:
+# https://hub.docker.com/r/library/python/tags/
+image: $CI_DEPENDENCY_PROXY_GROUP_IMAGE_PREFIX/python:3.10
+
+#commands to run in the Docker container before starting each job.
+before_script:
+  - pip install -e .[dev]
+
+# different stages in the pipeline
+stages:
+  - Formatter
+  - Test
+  - AdditionalTests
+  - Deploy
+
+formatter:
+  stage: Formatter
+  script:
+    - pip install black
+    - black --check --diff --color --line-length=100 ./
+
+pytest:
+  stage: Test
+  script:
+    - pytest -v --random-order ./tests
+
+tests-3.11:
+  stage: AdditionalTests
+  image: $CI_DEPENDENCY_PROXY_GROUP_IMAGE_PREFIX/python:3.11
+  needs: ["pytest"]
+  script:
+    - pytest -v --random-order ./tests
+  allow_failure: true
+
+tests-3.12:
+  extends: "tests-3.11"
+  stage: AdditionalTests
+  image: $CI_DEPENDENCY_PROXY_GROUP_IMAGE_PREFIX/python:3.12
+  allow_failure: true

bec_plugins/__init__.py

@@ -0,0 +1 @@
+from .bec_client import *

bec_plugins/bec_client/__init__.py

@@ -0,0 +1 @@
+from .plugins import *

bec_plugins/bec_client/hli/spec_hli.py

@@ -0,0 +1,245 @@
+from bec_lib.devicemanager import Device
+from bec_lib.scan_report import ScanReport
+
+# pylint:disable=undefined-variable
+# pylint: disable=too-many-arguments
+
+
+def dscan(
+    motor1: Device, m1_from: float, m1_to: float, steps: int, exp_time: float, **kwargs
+) -> ScanReport:
+    """Relative line scan with one device.
+
+    Args:
+        motor1 (Device): Device that should be scanned.
+        m1_from (float): Start position relative to the current position.
+        m1_to (float): End position relative to the current position.
+        steps (int): Number of steps.
+        exp_time (float): Exposure time.
+
+    Returns:
+        ScanReport: Status object.
+
+    Examples:
+        >>> dscan(dev.motor1, -5, 5, 10, 0.1)
+    """
+    return scans.line_scan(
+        motor1, m1_from, m1_to, steps=steps, exp_time=exp_time, relative=True, **kwargs
+    )
+
+
+def d2scan(
+    motor1: Device,
+    m1_from: float,
+    m1_to: float,
+    motor2: Device,
+    m2_from: float,
+    m2_to: float,
+    steps: int,
+    exp_time: float,
+    **kwargs
+) -> ScanReport:
+    """Relative line scan with two devices.
+
+    Args:
+        motor1 (Device): First device that should be scanned.
+        m1_from (float): Start position of the first device relative to its current position.
+        m1_to (float): End position of the first device relative to its current position.
+        motor2 (Device): Second device that should be scanned.
+        m2_from (float): Start position of the second device relative to its current position.
+        m2_to (float): End position of the second device relative to its current position.
+        steps (int): Number of steps.
+        exp_time (float): Exposure time
+
+    Returns:
+        ScanReport: Status object.
+
+    Examples:
+        >>> d2scan(dev.motor1, -5, 5, dev.motor2, -8, 8, 10, 0.1)
+    """
+    return scans.line_scan(
+        motor1,
+        m1_from,
+        m1_to,
+        motor2,
+        m2_from,
+        m2_to,
+        steps=steps,
+        exp_time=exp_time,
+        relative=True,
+        **kwargs
+    )
+
+
+def ascan(motor1, m1_from, m1_to, steps, exp_time, **kwargs):
+    """Absolute line scan with one device.
+
+    Args:
+        motor1 (Device): Device that should be scanned.
+        m1_from (float): Start position.
+        m1_to (float): End position.
+        steps (int): Number of steps.
+        exp_time (float): Exposure time.
+
+    Returns:
+        ScanReport: Status object.
+
+    Examples:
+        >>> ascan(dev.motor1, -5, 5, 10, 0.1)
+    """
+    return scans.line_scan(
+        motor1, m1_from, m1_to, steps=steps, exp_time=exp_time, relative=False, **kwargs
+    )
+
+
+def a2scan(motor1, m1_from, m1_to, motor2, m2_from, m2_to, steps, exp_time, **kwargs):
+    """Absolute line scan with two devices.
+
+    Args:
+        motor1 (Device): First device that should be scanned.
+        m1_from (float): Start position of the first device.
+        m1_to (float): End position of the first device.
+        motor2 (Device): Second device that should be scanned.
+        m2_from (float): Start position of the second device.
+        m2_to (float): End position of the second device.
+        steps (int): Number of steps.
+        exp_time (float): Exposure time
+
+    Returns:
+        ScanReport: Status object.
+
+    Examples:
+        >>> a2scan(dev.motor1, -5, 5, dev.motor2, -8, 8, 10, 0.1)
+    """
+    return scans.line_scan(
+        motor1,
+        m1_from,
+        m1_to,
+        motor2,
+        m2_from,
+        m2_to,
+        steps=steps,
+        exp_time=exp_time,
+        relative=False,
+        **kwargs
+    )
+
+
+def dmesh(motor1, m1_from, m1_to, m1_steps, motor2, m2_from, m2_to, m2_steps, exp_time, **kwargs):
+    """Relative mesh scan (grid scan) with two devices.
+
+    Args:
+        motor1 (Device): First device that should be scanned.
+        m1_from (float): Start position of the first device relative to its current position.
+        m1_to (float): End position of the first device relative to its current position.
+        m1_steps (int): Number of steps for motor1.
+        motor2 (Device): Second device that should be scanned.
+        m2_from (float): Start position of the second device relative to its current position.
+        m2_to (float): End position of the second device relative to its current position.
+        m2_steps (int): Number of steps for motor2.
+        exp_time (float): Exposure time
+
+    Returns:
+        ScanReport: Status object.
+
+    Examples:
+        >>> dmesh(dev.motor1, -5, 5, 10, dev.motor2, -8, 8, 10, 0.1)
+    """
+    return scans.grid_scan(
+        motor1,
+        m1_from,
+        m1_to,
+        m1_steps,
+        motor2,
+        m2_from,
+        m2_to,
+        m2_steps,
+        exp_time=exp_time,
+        relative=True,
+    )
+
+
+def amesh(motor1, m1_from, m1_to, m1_steps, motor2, m2_from, m2_to, m2_steps, exp_time, **kwargs):
+    """Absolute mesh scan (grid scan) with two devices.
+
+    Args:
+        motor1 (Device): First device that should be scanned.
+        m1_from (float): Start position of the first device.
+        m1_to (float): End position of the first device.
+        m1_steps (int): Number of steps for motor1.
+        motor2 (Device): Second device that should be scanned.
+        m2_from (float): Start position of the second device.
+        m2_to (float): End position of the second device.
+        m2_steps (int): Number of steps for motor2.
+        exp_time (float): Exposure time
+
+    Returns:
+        ScanReport: Status object.
+
+    Examples:
+        >>> amesh(dev.motor1, -5, 5, 10, dev.motor2, -8, 8, 10, 0.1)
+    """
+    return scans.grid_scan(
+        motor1,
+        m1_from,
+        m1_to,
+        m1_steps,
+        motor2,
+        m2_from,
+        m2_to,
+        m2_steps,
+        exp_time=exp_time,
+        relative=False,
+    )
+
+
+def umv(*args) -> ScanReport:
+    """Updated absolute move (i.e. blocking) for one or more devices.
+
+    Returns:
+        ScanReport: Status object.
+
+    Examples:
+        >>> umv(dev.samx, 1)
+        >>> umv(dev.samx, 1, dev.samy, 2)
+    """
+    return scans.umv(*args, relative=False)
+
+
+def umvr(*args) -> ScanReport:
+    """Updated relative move (i.e. blocking) for one or more devices.
+
+    Returns:
+        ScanReport: Status object.
+
+    Examples:
+        >>> umvr(dev.samx, 1)
+        >>> umvr(dev.samx, 1, dev.samy, 2)
+    """
+    return scans.umv(*args, relative=True)
+
+
+def mv(*args) -> ScanReport:
+    """Absolute move for one or more devices.
+
+    Returns:
+        ScanReport: Status object.
+
+    Examples:
+        >>> mv(dev.samx, 1)
+        >>> mv(dev.samx, 1, dev.samy, 2)
+    """
+    return scans.mv(*args, relative=False)
+
+
+def mvr(*args) -> ScanReport:
+    """Relative move for one or more devices.
+
+    Returns:
+        ScanReport: Status object.
+
+    Examples:
+        >>> mvr(dev.samx, 1)
+        >>> mvr(dev.samx, 1, dev.samy, 2)
+    """
+    return scans.mv(*args, relative=True)

bec_plugins/bec_client/plugins/__init__.py

@@ -0,0 +1 @@
+

bec_plugins/bec_client/startup/post_startup.py

@@ -0,0 +1,46 @@
+"""
+Post startup script for the BEC client. This script is executed after the
+IPython shell is started. It is used to load the beamline specific
+information and to setup the prompts.
+
+The script is executed in the global namespace of the IPython shell. This
+means that all variables defined here are available in the shell.
+
+If needed, bec command-line arguments can be parsed here. For example, to
+parse the --session argument, add the following lines to the script:
+
+    import argparse
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--session", help="Session name", type=str, default="my_default_session")
+    args = parser.parse_args()
+
+    if args.session == "my_session":
+        print("Loading my_session session")
+        from bec_plugins.bec_client.plugins.my_session import *
+    else:
+        print("Loading default session")
+        from bec_plugins.bec_client.plugins.default_session import *
+"""
+
+# pylint: disable=invalid-name, unused-import, import-error, undefined-variable, unused-variable, unused-argument, no-name-in-module
+import argparse
+
+from bec_lib import bec_logger
+
+logger = bec_logger.logger
+
+logger.info("Using the Debye startup script.")
+
+parser = argparse.ArgumentParser()
+parser.add_argument("--session", help="Session name", type=str, default="Debye")
+args = parser.parse_args()
+
+# SETUP BEAMLINE INFO
+from bec_client.plugins.SLS.sls_info import OperatorInfo, SLSInfo
+
+bec._beamline_mixin._bl_info_register(SLSInfo)
+bec._beamline_mixin._bl_info_register(OperatorInfo)
+
+# SETUP PROMPTS
+bec._ip.prompts.username = "Debye"
+bec._ip.prompts.status = 1

bec_plugins/bec_client/startup/pre_startup.py

@@ -0,0 +1,25 @@
+"""
+Pre-startup script for BEC client. This script is executed before the BEC client
+is started. It can be used to set up the BEC client configuration. The script is
+executed in the global namespace of the BEC client. This means that all
+variables defined here are available in the BEC client.
+
+To set up the BEC client configuration, use the ServiceConfig class. For example,
+to set the configuration file path, add the following lines to the script:
+
+    import pathlib
+    from bec_lib import ServiceConfig
+
+    current_path = pathlib.Path(__file__).parent.resolve()
+    CONFIG_PATH = f"{current_path}/<path_to_my_config_file.yaml>"
+
+    config = ServiceConfig(CONFIG_PATH)
+
+If this startup script defined a ServiceConfig object, the BEC client will use
+it to configure itself. Otherwise, the BEC client will use the default config.
+"""
+
+# example:
+# current_path = pathlib.Path(__file__).parent.resolve()
+# CONFIG_PATH = f"{current_path}/../../../bec_config.yaml"
+# config = ServiceConfig(CONFIG_PATH)

bec_plugins/services/NIDAQ_writer/NIDAQ_writer.py

@@ -7,12 +7,10 @@ import traceback
 
 import h5py
 import numpy as np
-from bec_lib import bec_logger, messages
+from bec_lib import MessageEndpoints, RedisConnector, ServiceConfig, bec_logger, messages
 from bec_lib.bec_service import BECService
-from bec_lib.endpoints import MessageEndpoints
-from bec_lib.file_utils import FileWriter
-from bec_lib.redis_connector import MessageObject, RedisConnector
-from bec_lib.service_config import ServiceConfig
+from bec_lib.file_utils import FileWriterMixin
+from bec_lib.redis_connector import MessageObject
 
 logger = bec_logger.logger
 
@@ -29,7 +27,7 @@ class NIDAQWriterService(BECService):
         super().__init__(config=config, connector_cls=connector_cls, unique_service=True)
         self.queue = queue.Queue()
         config = self._service_config.service_config.get("file_writer")
-        self.writer_mixin = FileWriter(service_config=config)
+        self.writer_mixin = FileWriterMixin(config)
         self._scan_status_consumer = None
         self._ni_data_consumer = None
         self._ni_data_event = None
@@ -161,7 +159,7 @@ class NIDAQWriterService(BECService):
         signals = {}
 
         for key in msgs[0].content["signals"]:
-            signals[key] = np.concatenate([msg.content["signals"][key]["value"] for msg in msgs])
+            signals[key] = np.concatenate([msg.content["signals"][key] for msg in msgs])
 
         # write data to queue
         self.queue.put(signals)

bec_plugins/services/NIDAQ_writer/scan_status.py

@@ -1,6 +1,4 @@
-from bec_lib import messages
-from bec_lib.endpoints import MessageEndpoints
-from bec_lib.redis_connector import RedisConnector
+from bec_lib import MessageEndpoints, RedisConnector, messages
 
 
 def send_scan_status(scan_number, status):
@@ -26,9 +24,19 @@ if __name__ == "__main__":
     parser = argparse.ArgumentParser(description="Scan status helper")
     command = parser.add_subparsers(dest="command")
     start = command.add_parser("start", help="Start a new scan")
-    start.add_argument("--scan_number", type=int, required=True, help="Scan number")
+    start.add_argument(
+        "--scan_number",
+        type=int,
+        required=True,
+        help="Scan number",
+    )
     stop = command.add_parser("stop", help="Stop the scan")
-    stop.add_argument("--scan_number", type=int, required=True, help="Scan number")
+    stop.add_argument(
+        "--scan_number",
+        type=int,
+        required=True,
+        help="Scan number",
+    )
 
     args = parser.parse_args()
     send_scan_status(args.scan_number, args.command)

bec_plugins/services/NIDAQ_writer/utils/nidaq_sim.py

@@ -2,8 +2,7 @@ import threading
 import time
 
 import numpy as np
-from bec_lib import messages
-from bec_lib.redis_connector import RedisConnector
+from bec_lib import MessageEndpoints, RedisConnector, ServiceConfig, bec_logger, messages
 
 
 class NIDAQSim(threading.Thread):
@@ -14,7 +13,10 @@ class NIDAQSim(threading.Thread):
         index = 0
         producer = RedisConnector(["localhost:6379"]).producer()
         signal = np.asarray(range(index, index + 600000))
-        signals = {"signal1": signal, "signal2": signal}
+        signals = {
+            "signal1": signal,
+            "signal2": signal,
+        }
 
         msg = messages.DeviceMessage(signals=signals)
         msg = msg.dumps()
@@ -38,6 +40,9 @@ class NIDAQSim(threading.Thread):
 
             time.sleep(0.5)
             print(f"Elapsed time: {time.time() - start}")
+        print(f"Total time: {time.time() - total_time}")
+        print(f"FPS: {index / (time.time() - total_time)}")
+        print(f"Signal size: {signal.nbytes/1e6*2} MB")
 
 
 if __name__ == "__main__":

bec_plugins/services/launch_writer.py

@@ -1,14 +1,15 @@
 import argparse
 import threading
 
-from bec_lib import bec_logger
-from bec_lib.redis_connector import RedisConnector
-from bec_lib.service_config import ServiceConfig
-
-from debye_bec.services.NIDAQ_writer import NIDAQWriterService
+from bec_lib import RedisConnector, ServiceConfig, bec_logger
+from NIDAQ_writer import NIDAQWriterService
 
 parser = argparse.ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter)
-parser.add_argument("--config", default="", help="path to the config file")
+parser.add_argument(
+    "--config",
+    default="",
+    help="path to the config file",
+)
 clargs = parser.parse_args()
 config_path = clargs.config
 
@@ -16,7 +17,10 @@ config = ServiceConfig(config_path)
 bec_logger.level = bec_logger.LOGLEVEL.INFO
 logger = bec_logger.logger
 
-bec_server = NIDAQWriterService(config=config, connector_cls=RedisConnector)
+bec_server = NIDAQWriterService(
+    config=config,
+    connector_cls=RedisConnector,
+)
 try:
     event = threading.Event()
     # pylint: disable=E1102

debye_bec/bec_ipython_client/startup/post_startup.py

@@ -1,36 +0,0 @@
-"""
-Post startup script for the BEC client. This script is executed after the
-IPython shell is started. It is used to load the beamline specific
-information and to setup the prompts.
-
-The script is executed in the global namespace of the IPython shell. This
-means that all variables defined here are available in the shell.
-
-While command-line arguments have to be set in the pre-startup script, the
-post-startup script can be used to load beamline specific information and
-to setup the prompts.
-
-    from bec_lib import bec_logger
-
-    logger = bec_logger.logger
-
-    # pylint: disable=import-error
-    _args = _main_dict["args"]
-
-    _session_name = "cSAXS"
-    if _args.session.lower() == "lamni":
-        from csaxs_bec.bec_ipython_client.plugins.cSAXS import *
-        from csaxs_bec.bec_ipython_client.plugins.LamNI import *
-
-        _session_name = "LamNI"
-        lamni = LamNI(bec)
-        logger.success("LamNI session loaded.")
-
-    elif _args.session.lower() == "csaxs":
-        print("Loading cSAXS session")
-        from csaxs_bec.bec_ipython_client.plugins.cSAXS import *
-
-        logger.success("cSAXS session loaded.")
-"""
-
-# pylint: disable=invalid-name, unused-import, import-error, undefined-variable, unused-variable, unused-argument, no-name-in-module

debye_bec/bec_ipython_client/startup/pre_startup.py

@@ -1,23 +0,0 @@
-"""
-Pre-startup script for BEC client. This script is executed before the BEC client
-is started. It can be used to add additional command line arguments. 
-"""
-
-from bec_lib.service_config import ServiceConfig
-
-
-def extend_command_line_args(parser):
-    """
-    Extend the command line arguments of the BEC client.
-    """
-
-    # parser.add_argument("--session", help="Session name", type=str, default="cSAXS")
-
-    return parser
-
-
-def get_config() -> ServiceConfig:
-    """
-    Create and return the service configuration.
-    """
-    return ServiceConfig(redis={"host": "x01da-bec-001", "port": 6379})

debye_bec/deployments/device_server/startup.py

@@ -1,11 +0,0 @@
-import os
-
-
-def setup_epics_ca():
-    # os.environ["EPICS_CA_AUTO_ADDR_LIST"] = "NO"
-    # os.environ["EPICS_CA_ADDR_LIST"] = "129.129.122.255 sls-x12sa-cagw.psi.ch:5836"
-    os.environ["PYTHONIOENCODING"] = "latin1"
-
-
-def run():
-    setup_epics_ca()

debye_bec/device_configs/x01da_database.yaml

@@ -1,727 +0,0 @@
-###################
-#### FRONT END ####
-###################
-
-## Slit Diaphragm -- Physical positioners
-sldi_trxr:
-  readoutPriority: baseline
-  description: Front-end slit diaphragm X-translation Ring-edge
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-FE-SLDI:TRXR
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-sldi_trxw:
-  readoutPriority: baseline
-  description: Front-end slit diaphragm X-translation Wall-edge
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-FE-SLDI:TRXW
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-sldi_tryb:
-  readoutPriority: baseline
-  description: Front-end slit diaphragm Y-translation Bottom-edge
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-FE-SLDI:TRYB
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-sldi_tryt:
-  readoutPriority: baseline
-  description: Front-end slit diaphragm X-translation Top-edge
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-FE-SLDI:TRYT
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-
-## Slit Diaphragm -- Virtual positioners
-
-sldi_centerx:
-  readoutPriority: baseline
-  description: Front-end slit diaphragm X-center
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-FE-SLDI:CENTERX
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-sldi_gapx:
-  readoutPriority: baseline
-  description: Front-end slit diaphragm X-gap
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-FE-SLDI:GAPX
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-sldi_centery:
-  readoutPriority: baseline
-  description: Front-end slit diaphragm Y-center
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-FE-SLDI:CENTERY
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-sldi_gapy:
-  readoutPriority: baseline
-  description: Front-end slit diaphragm Y-gap
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-FE-SLDI:GAPY
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-
-## Collimating Mirror -- Physical Positioners
-
-cm_trxu:
-  readoutPriority: baseline
-  description: Collimating Mirror X-translation upstream
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-FE-CM:TRXU
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-cm_trxd:
-  readoutPriority: baseline
-  description: Collimating Mirror X-translation downstream
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-FE-CM:TRXD
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-cm_tryu:
-  readoutPriority: baseline
-  description: Collimating Mirror Y-translation upstream
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-FE-CM:TRYU
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-cm_trydr:
-  readoutPriority: baseline
-  description: Collimating Mirror Y-translation downstream ring
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-FE-CM:TRYDR
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-cm_trydw:
-  readoutPriority: baseline
-  description: Collimating Mirror Y-translation downstream wall
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-FE-CM:TRYDW
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-cm_bnd:
-  readoutPriority: baseline
-  description: Collimating Mirror bender
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-FE-CM:BND
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-
-## Collimating Mirror -- Virtual Positioners
-
-cm_rotx:
-  readoutPriority: baseline
-  description: Collimating Morror Pitch
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-FE-CM:ROTX
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-cm_roty:
-  readoutPriority: baseline
-  description: Collimating Morror Yaw
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-FE-CM:ROTY
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-cm_rotz:
-  readoutPriority: baseline
-  description: Collimating Morror Roll
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-FE-CM:ROTZ
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-cm_xctp:
-  readoutPriority: baseline
-  description: Collimating Morror Center Point X
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-FE-CM:XTCP
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-cm_ytcp:
-  readoutPriority: baseline
-  description: Collimating Morror Center Point Y
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-FE-CM:YTCP
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-cm_ztcp:
-  readoutPriority: baseline
-  description: Collimating Morror Center Point Z
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-FE-CM:ZTCP
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-cm_xstripe:
-  readoutPriority: baseline
-  description: Collimating Morror X Stripe
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-FE-CM:XSTRIPE
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-
-###################
-###### OPTICS #####
-###################
-
-## Bragg Monochromator
-mo1_bragg:
-  readoutPriority: baseline
-  description: Positioner for the Monochromator
-  deviceClass: debye_bec.devices.mo1_bragg.Mo1Bragg
-  deviceConfig:
-    prefix: "X01DA-OP-MO1:BRAGG:"
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-
-## Monochromator -- Physical Positioners
-
-mo_try:
-  readoutPriority: baseline
-  description: Monochromator Y Translation
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-MO1:TRY
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-mo_trx:
-  readoutPriority: baseline
-  description: Monochromator X Translation
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-MO1:TRY
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-mo_roty:
-  readoutPriority: baseline
-  description: Monochromator Yaw
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-MO1:ROTY
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-
-## Focusing Mirror -- Physical Positioners
-
-fm_trxu:
-  readoutPriority: baseline
-  description: Focusing Mirror X-translation upstream
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-FM:TRXU
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-fm_trxd:
-  readoutPriority: baseline
-  description: Focusing Mirror X-translation downstream
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-FM:TRXD
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-fm_tryd:
-  readoutPriority: baseline
-  description: Focusing Mirror Y-translation downstream
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-FM:TRYD
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-fm_tryur:
-  readoutPriority: baseline
-  description: Focusing Mirror Y-translation upstream ring
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-FM:TRYUR
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-fm_tryuw:
-  readoutPriority: baseline
-  description: Focusing Mirror Y-translation upstream wall
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-FM:TRYUW
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-fm_bnd:
-  readoutPriority: baseline
-  description: Focusing Mirror bender
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-FM:BND
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-
-## Focusing Mirror -- Virtual Positioners
-
-fm_rotx:
-  readoutPriority: baseline
-  description: Focusing Morror Pitch
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-FM:ROTX
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-fm_roty:
-  readoutPriority: baseline
-  description: Focusing Morror Yaw
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-FM:ROTY
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-fm_rotz:
-  readoutPriority: baseline
-  description: Focusing Morror Roll
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-FM:ROTZ
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-fm_xctp:
-  readoutPriority: baseline
-  description: Focusing Morror Center Point X
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-FM:XTCP
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-fm_ytcp:
-  readoutPriority: baseline
-  description: Focusing Morror Center Point Y
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-FM:YTCP
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-fm_ztcp:
-  readoutPriority: baseline
-  description: Focusing Morror Center Point Z
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-FM:ZTCP
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-fm_xstripe:
-  readoutPriority: baseline
-  description: Focusing Morror X Stripe
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-FM:XSTRIPE
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-
-## Optics Slits 1 -- Physical positioners
-
-sl1_trxr:
-  readoutPriority: baseline
-  description: Optics slits 1 X-translation Ring-edge
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-SL1:TRXR
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-sl1_trxw:
-  readoutPriority: baseline
-  description: Optics slits 1 X-translation Wall-edge
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-SL1:TRXW
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-sl1_tryb:
-  readoutPriority: baseline
-  description: Optics slits 1 Y-translation Bottom-edge
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-SL1:TRYB
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-sl1_tryt:
-  readoutPriority: baseline
-  description: Optics slits 1 X-translation Top-edge
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-SL1:TRYT
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-bm1_try:
-  readoutPriority: baseline
-  description: Beam Monitor 1 Y-translation
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-BM1:TRY
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-
-## Optics Slits 1 -- Virtual positioners
-
-sl1_centerx:
-  readoutPriority: baseline
-  description: Optics slits 1 X-center
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-SL1:CENTERX
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-sl1_gapx:
-  readoutPriority: baseline
-  description: Optics slits 1 X-gap
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-SL1:GAPX
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-sl1_centery:
-  readoutPriority: baseline
-  description: Optics slits 1 Y-center
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-SL1:CENTERY
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-sl1_gapy:
-  readoutPriority: baseline
-  description: Optics slits 1 Y-gap
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-SL1:GAPY
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-
-## Optics Slits 2 -- Physical positioners
-
-sl2_trxr:
-  readoutPriority: baseline
-  description: Optics slits 2 X-translation Ring-edge
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-SL2:TRXR
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-sl2_trxw:
-  readoutPriority: baseline
-  description: Optics slits 2 X-translation Wall-edge
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-SL2:TRXW
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-sl2_tryb:
-  readoutPriority: baseline
-  description: Optics slits 2 Y-translation Bottom-edge
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-SL2:TRYB
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-sl2_tryt:
-  readoutPriority: baseline
-  description: Optics slits 2 X-translation Top-edge
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-SL2:TRYT
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-bm2_try:
-  readoutPriority: baseline
-  description: Beam Monitor 2 Y-translation
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-BM2:TRY
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-
-## Optics Slits 2 -- Virtual positioners
-
-sl2_centerx:
-  readoutPriority: baseline
-  description: Optics slits 2 X-center
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-SL2:CENTERX
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-sl2_gapx:
-  readoutPriority: baseline
-  description: Optics slits 2 X-gap
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-SL2:GAPX
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-sl2_centery:
-  readoutPriority: baseline
-  description: Optics slits 2 Y-center
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-SL2:CENTERY
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-sl2_gapy:
-  readoutPriority: baseline
-  description: Optics slits 2 Y-gap
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-OP-SL2:GAPY
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-  
-###############################
-###### EXPERIMENTAL HUTCH #####
-###############################
-
-###########################################
-## Optical Table -- Physical Positioners ##
-###########################################
-
-ot_tryu:
-  readoutPriority: baseline
-  description: Optical Table Y-Translation Upstream
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-ES-OT:TRYU
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-ot_tryd:
-  readoutPriority: baseline
-  description: Optical Table Y-Translation Downstream
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-ES-OT:TRYD
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-
-############################################
-## Optical Table -- Virtual Positioners ###
-############################################
-
-ot_try:
-  readoutPriority: baseline
-  description: Optical Table Y-Translation
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-ES-OT:TRY
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-ot_pitch:
-  readoutPriority: baseline
-  description: Optical Table Pitch
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-ES-OT:ROTX
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-
-###############################################
-## End Station Slits -- Physical Positioners ##
-###############################################
-
-es0sl_trxr:
-  readoutPriority: baseline
-  description: End Station slits X-translation Ring-edge
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-ES0-SL:TRXR
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-es0sl_trxw:
-  readoutPriority: baseline
-  description: End Station slits X-translation Wall-edge
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-ES0-SL:TRXW
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-es0sl_tryb:
-  readoutPriority: baseline
-  description: End Station slits Y-translation Bottom-edge
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-ES0-SL:TRYB
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-es0sl_tryt:
-  readoutPriority: baseline
-  description: End Station slits X-translation Top-edge
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-ES0-SL:TRYT
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-
-##############################################
-## End Station Slits -- Virtual positioners ##
-##############################################
-
-es0sl_center:
-  readoutPriority: baseline
-  description: End Station slits X-center
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-ES0-SL:CENTERX
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-es0sl_gapx:
-  readoutPriority: baseline
-  description: End Station slits X-gap
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-ES0-SL:GAPX
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-es0sl_centery:
-  readoutPriority: baseline
-  description: End Station slits Y-center
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-ES0-SL:CENTERY
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-es0sl_gapy:
-  readoutPriority: baseline
-  description: End Station slits Y-gap
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-ES0-SL:GAPY
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-
-
-#########################################
-## Exit Window -- Physical Positioners ##
-#########################################
-
-es0wi_try:
-  readoutPriority: baseline
-  description: End Station 0 Exit Window Y-translation
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig:
-    prefix: X01DA-ES0-WI:TRY
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-
-
-##########################
-## AreaDetector Devices ##
-##########################
-
-#op_bpm1:
-#  readoutPriority: baseline
-#  description: Optics Hutch Beam Position Monitor 1
-#  deviceClass: ophyd.AreaDetector
-#  deviceConfig:
-#    prefix: X01DA-OP-GIGE01:
-#  onFailure: retry
-#  enabled: true
-#  softwareTrigger: false
-
-#op_bpm2:
-#  readoutPriority: baseline
-#  description: Optics Hutch Beam Position Monitor 2
-#  deviceClass: ophyd.AreaDetector
-#  deviceConfig:
-#    prefix: X01DA-OP-GIGE02:
-#  onFailure: retry
-#  enabled: true
-#  softwareTrigger: false
-
-#es_xryeye:
-#  readoutPriority: baseline
-#  description: X-Ray Eye
-#  deviceClass: ophyd.AreaDetector
-#  deviceConfig:
-#    prefix: X01DA-:
-#  onFailure: retry
-#  enabled: true
-#  softwareTrigger: false

debye_bec/device_configs/x01da_test_config.yaml

@@ -1,19 +0,0 @@
-## Bragg Monochromator
-mo1_bragg:
-  readoutPriority: baseline
-  description: Positioner for the Monochromator
-  deviceClass: debye_bec.devices.mo1_bragg.Mo1Bragg
-  deviceConfig:
-    prefix: "X01DA-OP-MO1:BRAGG:"
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false
-dummy_pv:
-  readoutPriority: monitored
-  description: Heartbeat of Bragg
-  deviceClass: ophyd.EpicsSignalRO
-  deviceConfig:
-    read_pv: "X01DA-OP-MO1:BRAGG:heartbeat_RBV"
-  onFailure: retry
-  enabled: true
-  softwareTrigger: false

debye_bec/devices/device_list.md

@@ -1,11 +0,0 @@
-// This file was autogenerated. Do not edit it manually.
-## Device List
-### debye_bec 
-| Device | Documentation | Module |
-| :----- | :------------- | :------ |
-| Mo1Bragg | Class for the Mo1 Bragg positioner of the Debye beamline.<br>    The prefix to connect to the soft IOC is X01DA-OP-MO1:BRAGG: which is connected to<br>    the NI motor controller via web sockets.<br>     | [debye_bec.devices.mo1_bragg](https://gitlab.psi.ch/bec/debye_bec/-/blob/main/debye_bec/devices/mo1_bragg.py) |
-| Mo1BraggCrystal | Mo1 Bragg PVs to set the crystal parameters | [debye_bec.devices.mo1_bragg](https://gitlab.psi.ch/bec/debye_bec/-/blob/main/debye_bec/devices/mo1_bragg.py) |
-| Mo1BraggEncoder | Mo1 Bragg PVs to communicate with the encoder | [debye_bec.devices.mo1_bragg](https://gitlab.psi.ch/bec/debye_bec/-/blob/main/debye_bec/devices/mo1_bragg.py) |
-| Mo1BraggScanControl | Mo1 Bragg PVs to control the scan after setting the parameters. | [debye_bec.devices.mo1_bragg](https://gitlab.psi.ch/bec/debye_bec/-/blob/main/debye_bec/devices/mo1_bragg.py) |
-| Mo1BraggScanSettings | Mo1 Bragg PVs to set the scan setttings | [debye_bec.devices.mo1_bragg](https://gitlab.psi.ch/bec/debye_bec/-/blob/main/debye_bec/devices/mo1_bragg.py) |
-| Mo1BraggStatus | Mo1 Bragg PVs for status monitoring | [debye_bec.devices.mo1_bragg](https://gitlab.psi.ch/bec/debye_bec/-/blob/main/debye_bec/devices/mo1_bragg.py) |

debye_bec/devices/mo1_bragg.py

@@ -1,891 +0,0 @@
-""" Module for the Mo1 Bragg positioner of the Debye beamline.
-The soft IOC is reachable via the EPICS prefix X01DA-OP-MO1:BRAGG: and connected
-to a motor controller via web sockets. The Mo1 Bragg positioner is not only a 
-positioner, but also a scan controller to setup XAS and XRD scans. A few scan modes
-are programmed in the controller, e.g. simple and advanced XAS scans + XRD triggering mode.
-
-Note: For some of the Epics PVs, in particular action buttons, the suffix .PROC and 
-put_complete=True is used to ensure that the action is executed completely. This is believed
-to allow for a more stable execution of the action."""
-
-import enum
-import threading
-import time
-import traceback
-from dataclasses import dataclass
-from typing import Literal
-
-from bec_lib.logger import bec_logger
-from ophyd import Component as Cpt
-from ophyd import (
-    Device,
-    DeviceStatus,
-    EpicsSignal,
-    EpicsSignalRO,
-    EpicsSignalWithRBV,
-    Kind,
-    PositionerBase,
-    Signal,
-    Staged,
-)
-from ophyd.utils import LimitError
-from ophyd_devices.utils import bec_scaninfo_mixin, bec_utils
-from ophyd_devices.utils.errors import DeviceStopError, DeviceTimeoutError
-
-logger = bec_logger.logger
-
-
-class ScanControlScanStatus(int, enum.Enum):
-    """Enum class for the scan status of the Bragg positioner"""
-
-    PARAMETER_WRONG = 0
-    VALIDATION_PENDING = 1
-    READY = 2
-    RUNNING = 3
-
-
-class ScanControlLoadMessage(int, enum.Enum):
-    """Enum for validating messages for load message of the Bragg positioner"""
-
-    PENDING = 0
-    STARTED = 1
-    SUCCESS = 2
-    ERR_XRD_MEAS_LEN_LOW = 3
-    ERR_XRD_N_TRIGGERS_LOW = 4
-    ERR_XRD_TRIGS_EVERY_N_LOW = 5
-    ERR_XRD_MEAS_LEN_HI = 6
-    ERR_XRD_N_TRIGGERS_HI = 7
-    ERR_XRD_TRIGS_EVERY_N_HI = 8
-    ERR_SCAN_HI_ANGLE_LIMIT = 9
-    ERR_SCAN_LOW_ANGLE_LIMITS = 10
-    ERR_SCAN_TIME = 11
-    ERR_SCAN_VEL_TOO_HI = 12
-    ERR_SCAN_ANGLE_OUT_OF_LIM = 13
-    ERR_SCAN_HIGH_VEL_LAR_42 = 14 
-    ERR_SCAN_MODE_INVALID = 15
-
-
-class Mo1BraggError(Exception):
-    """Mo1Bragg specific exception"""
-
-
-class MoveType(str, enum.Enum):
-    """Enum class to switch between move types energy and angle for the Bragg positioner"""
-
-    ENERGY = "energy"
-    ANGLE = "angle"
-
-
-class ScanControlMode(int, enum.Enum):
-    """Enum class for the scan control mode of the Bragg positioner"""
-
-    SIMPLE = 0
-    ADVANCED = 1
-
-
-class MoveTypeSignal(Signal):
-    """Custom Signal to set the move type of the Bragg positioner"""
-
-    # pylint: disable=arguments-differ
-    def set(self, value: str | MoveType) -> None:
-        """Returns currently active move method
-
-        Args:
-            value (str | MoveType) : Can be either 'energy' or 'angle'
-        """
-
-        value = MoveType(value.lower())
-        self._readback = value.value
-
-
-class Mo1BraggStatus(Device):
-    """Mo1 Bragg PVs for status monitoring"""
-
-    error_status = Cpt(EpicsSignalRO, suffix="error_status_RBV", kind="config", auto_monitor=True)
-    brake_enabled = Cpt(EpicsSignalRO, suffix="brake_enabled_RBV", kind="config", auto_monitor=True)
-    mot_commutated = Cpt(
-        EpicsSignalRO, suffix="mot_commutated_RBV", kind="config", auto_monitor=True
-    )
-    axis_enabled = Cpt(EpicsSignalRO, suffix="axis_enabled_RBV", kind="config", auto_monitor=True)
-    enc_initialized = Cpt(
-        EpicsSignalRO, suffix="enc_initialized_RBV", kind="config", auto_monitor=True
-    )
-    heartbeat = Cpt(EpicsSignalRO, suffix="heartbeat_RBV", kind="config", auto_monitor=True)
-
-
-class Mo1BraggEncoder(Device):
-    """Mo1 Bragg PVs to communicate with the encoder"""
-
-    enc_reinit = Cpt(EpicsSignal, suffix="enc_reinit.PROC", kind="config")
-    enc_reinit_done = Cpt(EpicsSignalRO, suffix="enc_reinit_done_RBV", kind="config")
-
-
-class Mo1BraggCrystal(Device):
-    """Mo1 Bragg PVs to set the crystal parameters"""
-
-    offset_si111 = Cpt(EpicsSignalWithRBV, suffix="offset_si111", kind="config")
-    offset_si311 = Cpt(EpicsSignalWithRBV, suffix="offset_si311", kind="config")
-    xtal_enum = Cpt(EpicsSignalWithRBV, suffix="xtal_ENUM", kind="config")
-    d_spacing_si111 = Cpt(EpicsSignalWithRBV, suffix="d_spacing_si111", kind="config")
-    d_spacing_si311 = Cpt(EpicsSignalWithRBV, suffix="d_spacing_si311", kind="config")
-    set_offset = Cpt(EpicsSignal, suffix="set_offset.PROC", kind="config", put_complete=True)
-    current_xtal = Cpt(
-        EpicsSignalRO, suffix="current_xtal_ENUM_RBV", kind="normal", auto_monitor=True
-    )
-
-
-class Mo1BraggScanSettings(Device):
-    """Mo1 Bragg PVs to set the scan setttings"""
-
-    # XRD settings
-    xrd_select_ref_enum = Cpt(EpicsSignalWithRBV, suffix="xrd_select_ref_ENUM", kind="config")
-    xrd_enable_hi_enum = Cpt(EpicsSignalWithRBV, suffix="xrd_enable_hi_ENUM", kind="config")
-
-    xrd_time_hi = Cpt(EpicsSignalWithRBV, suffix="xrd_time_hi", kind="config")
-    xrd_n_trigger_hi = Cpt(EpicsSignalWithRBV, suffix="xrd_n_trigger_hi", kind="config")
-    xrd_every_n_hi = Cpt(EpicsSignalWithRBV, suffix="xrd_every_n_hi", kind="config")
-
-    xrd_enable_lo_enum = Cpt(EpicsSignalWithRBV, suffix="xrd_enable_lo_ENUM", kind="config")
-    xrd_time_lo = Cpt(EpicsSignalWithRBV, suffix="xrd_time_lo", kind="config")
-    xrd_n_trigger_lo = Cpt(EpicsSignalWithRBV, suffix="xrd_n_trigger_lo", kind="config")
-    xrd_every_n_lo = Cpt(EpicsSignalWithRBV, suffix="xrd_every_n_lo", kind="config")
-
-    # XAS simple scan settings
-    s_scan_angle_hi = Cpt(EpicsSignalWithRBV, suffix="s_scan_angle_hi", kind="config")
-    s_scan_angle_lo = Cpt(EpicsSignalWithRBV, suffix="s_scan_angle_lo", kind="config")
-    s_scan_energy_lo = Cpt(
-        EpicsSignalWithRBV, suffix="s_scan_energy_lo", kind="config", auto_monitor=True
-    )
-    s_scan_energy_hi = Cpt(
-        EpicsSignalWithRBV, suffix="s_scan_energy_hi", kind="config", auto_monitor=True
-    )
-    s_scan_scantime = Cpt(
-        EpicsSignalWithRBV, suffix="s_scan_scantime", kind="config", auto_monitor=True
-    )
-
-    # XAS advanced scan settings
-    a_scan_pos = Cpt(EpicsSignalWithRBV, suffix="a_scan_pos", kind="config", auto_monitor=True)
-    a_scan_vel = Cpt(EpicsSignalWithRBV, suffix="a_scan_vel", kind="config", auto_monitor=True)
-    a_scan_time = Cpt(EpicsSignalWithRBV, suffix="a_scan_time", kind="config", auto_monitor=True)
-
-
-class Mo1BraggScanControl(Device):
-    """Mo1 Bragg PVs to control the scan after setting the parameters."""
-
-    scan_mode_enum = Cpt(EpicsSignalWithRBV, suffix="scan_mode_ENUM", kind="config")
-    scan_duration = Cpt(
-        EpicsSignalWithRBV, suffix="scan_duration", kind="config", auto_monitor=True
-    )
-    scan_load = Cpt(EpicsSignal, suffix="scan_load.PROC", kind="config", put_complete=True)
-    scan_msg = Cpt(EpicsSignalRO, suffix="scan_msg_ENUM_RBV", kind="config", auto_monitor=True)
-    scan_start_infinite = Cpt(
-        EpicsSignal, suffix="scan_start_infinite.PROC", kind="config", put_complete=True
-    )
-    scan_start_timer = Cpt(
-        EpicsSignal, suffix="scan_start_timer.PROC", kind="config", put_complete=True
-    )
-    scan_stop = Cpt(EpicsSignal, suffix="scan_stop.PROC", kind="config", put_complete=True)
-    scan_status = Cpt(
-        EpicsSignalRO, suffix="scan_status_ENUM_RBV", kind="config", auto_monitor=True
-    )
-    scan_time_left = Cpt(
-        EpicsSignalRO, suffix="scan_time_left_RBV", kind="config", auto_monitor=True
-    )
-    scan_done = Cpt(EpicsSignalRO, suffix="scan_done_RBV", kind="config", auto_monitor=True)
-    scan_val_reset = Cpt(
-        EpicsSignal, suffix="scan_val_reset.PROC", kind="config", put_complete=True
-    )
-    scan_progress = Cpt(EpicsSignalRO, suffix="scan_progress_RBV", kind="config", auto_monitor=True)
-    scan_spectra_done = Cpt(
-        EpicsSignalRO, suffix="scan_n_osc_RBV", kind="config", auto_monitor=True
-    )
-    scan_spectra_left = Cpt(
-        EpicsSignalRO, suffix="scan_n_osc_left_RBV", kind="config", auto_monitor=True
-    )
-
-
-@dataclass
-class ScanParameter:
-    """Dataclass to store the scan parameters for the Mo1 Bragg positioner.
-    This needs to be in sync with the kwargs of the MO1 Bragg scans from Debye, to
-    ensure that the scan parameters are correctly set. Any changes in the scan kwargs,
-    i.e. renaming or adding new parameters, need to be represented here as well."""
-
-    scan_time: float = None
-    scan_duration: float = None
-    xrd_enable_low: bool = None
-    xrd_enable_high: bool = None
-    num_trigger_low: int = None
-    num_trigger_high: int = None
-    exp_time_low: float = None
-    exp_time_high: float = None
-    cycle_low: int = None
-    cycle_high: int = None
-    start: float = None
-    stop: float = None
-
-
-class Mo1Bragg(Device, PositionerBase):
-    """Class for the Mo1 Bragg positioner of the Debye beamline.
-    The prefix to connect to the soft IOC is X01DA-OP-MO1:BRAGG: which is connected to
-    the NI motor controller via web sockets.
-    """
-
-    USER_ACCESS = ["set_xtal", "stop_scan"]
-
-    crystal = Cpt(Mo1BraggCrystal, "")
-    encoder = Cpt(Mo1BraggEncoder, "")
-    scan_settings = Cpt(Mo1BraggScanSettings, "")
-    scan_control = Cpt(Mo1BraggScanControl, "")
-    status = Cpt(Mo1BraggStatus, "")
-
-    # signal to indicate the move type 'energy' or 'angle'
-    move_type = Cpt(MoveTypeSignal, value=MoveType.ENERGY, kind="config")
-
-    # Energy PVs
-    readback = Cpt(
-        EpicsSignalRO, suffix="feedback_pos_energy_RBV", kind="hinted", auto_monitor=True
-    )
-    setpoint = Cpt(
-        EpicsSignalWithRBV, suffix="set_abs_pos_energy", kind="normal", auto_monitor=True
-    )
-    motor_is_moving = Cpt(
-        EpicsSignalRO, suffix="move_abs_done_RBV", kind="normal", auto_monitor=True
-    )
-    low_lim = Cpt(EpicsSignalRO, suffix="lo_lim_pos_energy_RBV", kind="config", auto_monitor=True)
-    high_lim = Cpt(EpicsSignalRO, suffix="hi_lim_pos_energy_RBV", kind="config", auto_monitor=True)
-    velocity = Cpt(EpicsSignalWithRBV, suffix="move_velocity", kind="config", auto_monitor=True)
-
-    # Angle PVs
-    # TODO makd angle motion a pseudo motor
-    feedback_pos_angle = Cpt(
-        EpicsSignalRO, suffix="feedback_pos_angle_RBV", kind="normal", auto_monitor=True
-    )
-    setpoint_abs_angle = Cpt(
-        EpicsSignalWithRBV, suffix="set_abs_pos_angle", kind="normal", auto_monitor=True
-    )
-    low_limit_angle = Cpt(
-        EpicsSignalRO, suffix="lo_lim_pos_angle_RBV", kind="config", auto_monitor=True
-    )
-    high_limit_angle = Cpt(
-        EpicsSignalRO, suffix="hi_lim_pos_angle_RBV", kind="config", auto_monitor=True
-    )
-
-    # Execute motion
-    move_abs = Cpt(EpicsSignal, suffix="move_abs.PROC", kind="config", put_complete=True)
-    move_stop = Cpt(EpicsSignal, suffix="move_stop.PROC", kind="config", put_complete=True)
-
-    SUB_READBACK = "readback"
-    _default_sub = SUB_READBACK
-    SUB_PROGRESS = "progress"
-
-    def __init__(
-        self, prefix="", *, name: str, kind: Kind = None, device_manager=None, parent=None, **kwargs
-    ):
-        """Initialize the Mo1 Bragg positioner.
-
-        Args:
-            prefix (str): EPICS prefix for the device
-            name (str): Name of the device
-            kind (Kind): Kind of the device
-            device_manager (DeviceManager): Device manager instance
-            parent (Device): Parent device
-            kwargs: Additional keyword arguments
-        """
-        super().__init__(prefix, name=name, kind=kind, parent=parent, **kwargs)
-        self._stopped = False
-        self.device_manager = device_manager
-        self._move_thread = None
-        self.service_cfg = None
-        self.scaninfo = None
-        # Init scan parameters
-        self.scan_parameter = ScanParameter()
-
-        self.timeout_for_pvwait = 2.5
-        self.readback.name = self.name
-        # Wait for connection on all components, ensure IOC is connected
-        self.wait_for_connection(all_signals=True, timeout=5)
-
-        if device_manager:
-            self.device_manager = device_manager
-        else:
-            self.device_manager = bec_utils.DMMock()
-
-        self.connector = self.device_manager.connector
-        self._update_scaninfo()
-        self._on_init()
-
-    def _on_init(self):
-        """Action to be executed on initialization of the device"""
-        self.scan_control.scan_progress.subscribe(self._progress_update, run=False)
-
-    def _progress_update(self, value, **kwargs) -> None:
-        """Callback method to update the scan progress, runs a callback to SUB_PROGRESS subscribers, i.e. BEC.
-
-        Args:
-            value (int) : current progress value
-        """
-        max_value = 100
-        self._run_subs(
-            sub_type=self.SUB_PROGRESS,
-            value=value,
-            max_value=max_value,
-            done=bool(max_value == value),
-        )
-
-    def _update_scaninfo(self) -> None:
-        """Connect to the ScanInfo mixin"""
-        self.scaninfo = bec_scaninfo_mixin.BecScaninfoMixin(self.device_manager)
-        self.scaninfo.load_scan_metadata()
-
-    @property
-    def stopped(self) -> bool:
-        """Return the stopped flag. If True, the motion is stopped."""
-        return self._stopped
-
-    def stop(self, *, success=False) -> None:
-        """Stop any motion on the positioner
-
-        Args:
-            success (bool) : Flag to indicate if the motion was successful
-        """
-        self.move_stop.put(1)
-        self._stopped = True
-        if self._move_thread is not None:
-            self._move_thread.join()
-            self._move_thread = None
-        super().stop(success=success)
-
-    def stop_scan(self) -> None:
-        """Stop the currently running scan gracefully, this finishes the running oscillation."""
-        self.scan_control.scan_stop.put(1)
-
-    # -------------- Positioner specific methods -----------------#
-    @property
-    def limits(self) -> tuple:
-        """Return limits of the Bragg positioner"""
-        if self.move_type.get() == MoveType.ENERGY:
-            return (self.low_lim.get(), self.high_lim.get())
-        return (self.low_limit_angle.get(), self.high_limit_angle.get())
-
-    @property
-    def low_limit(self) -> float:
-        """Return low limit of axis"""
-        return self.limits[0]
-
-    @property
-    def high_limit(self) -> float:
-        """Return high limit of axis"""
-        return self.limits[1]
-
-    @property
-    def egu(self) -> str:
-        """Return the engineering units of the positioner"""
-        if self.move_type.get() == MoveType.ENERGY:
-            return "eV"
-        return "deg"
-
-    @property
-    def position(self) -> float:
-        """Return the current position of Mo1Bragg, considering the move type"""
-        move_type = self.move_type.get()
-        move_cpt = self.readback if move_type == MoveType.ENERGY else self.feedback_pos_angle
-        return move_cpt.get()
-
-    # pylint: disable=arguments-differ
-    def check_value(self, value: float) -> None:
-        """Method to check if a value is within limits of the positioner. Called by PositionerBase.move()
-
-        Args:
-            value (float) : value to move axis to.
-        """
-        low_limit, high_limit = self.limits
-
-        if low_limit < high_limit and not low_limit <= value <= high_limit:
-            raise LimitError(f"position={value} not within limits {self.limits}")
-
-    def _move_and_finish(
-        self,
-        target_pos: float,
-        move_cpt: Cpt,
-        read_cpt: Cpt,
-        status: DeviceStatus,
-        update_frequency: float = 0.1,
-    ) -> None:
-        """Method to be called in the move thread to move the Bragg positioner to the target position.
-
-        Args:
-            target_pos (float)      : target position for the motion
-            move_cpt (Cpt)          : component to set the target position on the IOC,
-                                      either setpoint or setpoint_abs_angle depending on the move type
-            read_cpt (Cpt)          : component to read the current position of the motion, readback or feedback_pos_angle
-            status (DeviceStatus)   : status object to set the status of the motion
-            update_frequency (float): Optional, frequency to update the current position of the motion, defaults to 0.1s
-        """
-        try:
-            # Set the target position on IOC
-            move_cpt.put(target_pos)
-            self.move_abs.put(1)
-            # Currently sleep is needed due to delay in updates on PVs, maybe time can be reduced
-            time.sleep(0.5)
-            while self.motor_is_moving.get() == 0:
-                if self.stopped:
-                    raise DeviceStopError(f"Device {self.name} was stopped")
-                time.sleep(update_frequency)
-            # pylint: disable=protected-access
-            status.set_finished()
-        # pylint: disable=broad-except
-        except Exception as exc:
-            content = traceback.format_exc()
-            logger.error(f"Error in move thread of device {self.name}: {content}")
-            status.set_exception(exc=exc)
-
-    def move(self, value: float, move_type: str | MoveType = None, **kwargs) -> DeviceStatus:
-        """Move the Bragg positioner to the specified value, allows to switch between move types angle and energy.
-
-        Args:
-            value (float) : target value for the motion
-            move_type (str | MoveType) : Optional, specify the type of move, either 'energy' or 'angle'
-
-        Returns:
-            DeviceStatus : status object to track the motion
-        """
-        self._stopped = False
-        if move_type is not None:
-            self.move_type.put(move_type)
-        move_type = self.move_type.get()
-        move_cpt = self.setpoint if move_type == MoveType.ENERGY else self.setpoint_abs_angle
-        read_cpt = self.readback if move_type == MoveType.ENERGY else self.feedback_pos_angle
-
-        self.check_value(value)
-        status = DeviceStatus(device=self)
-
-        self._move_thread = threading.Thread(
-            target=self._move_and_finish, args=(value, move_cpt, read_cpt, status, 0.1)
-        )
-        self._move_thread.start()
-        return status
-
-    # -------------- End of Positioner specific methods -----------------#
-
-    # -------------- MO1 Bragg specific methods -----------------#
-
-    def set_xtal(
-        self,
-        xtal_enum: Literal["111", "311"],
-        offset_si111: float = None,
-        offset_si311: float = None,
-        d_spacing_si111: float = None,
-        d_spacing_si311: float = None,
-    ) -> None:
-        """Method to set the crystal parameters of the Bragg positioner
-
-        Args:
-            xtal_enum (Literal["111", "311"]) : Enum to set the crystal orientation
-            offset_si111 (float) : Offset for the 111 crystal
-            offset_si311 (float) : Offset for the 311 crystal
-            d_spacing_si111 (float) : d-spacing for the 111 crystal
-            d_spacing_si311 (float) : d-spacing for the 311 crystal
-        """
-        if offset_si111 is not None:
-            self.crystal.offset_si111.put(offset_si111)
-        if offset_si311 is not None:
-            self.crystal.offset_si311.put(offset_si311)
-        if d_spacing_si111 is not None:
-            self.crystal.d_spacing_si111.put(d_spacing_si111)
-        if d_spacing_si311 is not None:
-            self.crystal.d_spacing_si311.put(d_spacing_si311)
-        if xtal_enum == "111":
-            crystal_set = 0
-        elif xtal_enum == "311":
-            crystal_set = 1
-        else:
-            raise ValueError(
-                f"Invalid argument for xtal_enum : {xtal_enum}, choose from '111' or '311'"
-            )
-        self.crystal.xtal_enum.put(crystal_set)
-        self.crystal.set_offset.put(1)
-
-    def set_xas_settings(self, low: float, high: float, scan_time: float) -> None:
-        """Set XAS parameters for upcoming scan.
-
-        Args:
-            low (float): Low energy/angle value of the scan
-            high (float): High energy/angle value of the scan
-            scan_time (float): Time for a half oscillation
-        """
-        move_type = self.move_type.get()
-        if move_type == MoveType.ENERGY:
-            self.scan_settings.s_scan_energy_lo.put(low)
-            self.scan_settings.s_scan_energy_hi.put(high)
-        else:
-            self.scan_settings.s_scan_angle_lo.put(low)
-            self.scan_settings.s_scan_angle_hi.put(high)
-        self.scan_settings.s_scan_scantime.put(scan_time)
-
-    def set_xrd_settings(
-        self,
-        enable_low: bool,
-        enable_high: bool,
-        num_trigger_low: int,
-        num_trigger_high: int,
-        exp_time_low: int,
-        exp_time_high: int,
-        cycle_low: int,
-        cycle_high: int,
-    ) -> None:
-        """Set XRD settings for the upcoming scan.
-
-        Args:
-            enable_low (bool): Enable XRD for low energy/angle
-            enable_high (bool): Enable XRD for high energy/angle
-            num_trigger_low (int): Number of triggers for low energy/angle
-            num_trigger_high (int): Number of triggers for high energy/angle
-            exp_time_low (int): Exposure time for low energy/angle
-            exp_time_high (int): Exposure time for high energy/angle
-            cycle_low (int): Cycle for low energy/angle
-            cycle_high (int): Cycle for high energy/angle
-        """
-        self.scan_settings.xrd_enable_hi_enum.put(int(enable_high))
-        self.scan_settings.xrd_enable_lo_enum.put(int(enable_low))
-        self.scan_settings.xrd_n_trigger_hi.put(num_trigger_high)
-        self.scan_settings.xrd_n_trigger_lo.put(num_trigger_low)
-        self.scan_settings.xrd_time_hi.put(exp_time_high)
-        self.scan_settings.xrd_time_lo.put(exp_time_low)
-        self.scan_settings.xrd_every_n_hi.put(cycle_high)
-        self.scan_settings.xrd_every_n_lo.put(cycle_low)
-
-    def set_scan_control_settings(self, mode: ScanControlMode, scan_duration: float) -> None:
-        """Set the scan control settings for the upcoming scan.
-
-        Args:
-            mode (ScanControlMode): Mode for the scan, either simple or advanced
-            scan_duration (float): Duration of the scan
-        """
-        val = ScanControlMode(mode).value
-        self.scan_control.scan_mode_enum.put(val)
-        self.scan_control.scan_duration.put(scan_duration)
-
-    def _update_scan_parameter(self):
-        """Get the scaninfo parameters for the scan."""
-        for key, value in self.scaninfo.scan_msg.content["info"]["kwargs"].items():
-            if hasattr(self.scan_parameter, key):
-                setattr(self.scan_parameter, key, value)
-
-    # -------------- End MO1 Bragg specific methods -----------------#
-
-    # -------------- Flyer Interface methods -----------------#
-
-    def kickoff(self):
-        """Kickoff the device, called from BEC."""
-        scan_duration = self.scan_control.scan_duration.get()
-        # TODO implement better logic for infinite scans, at least bring it up with Debye
-        start_func = (
-            self.scan_control.scan_start_infinite.put
-            if scan_duration < 0.1
-            else self.scan_control.scan_start_timer.put
-        )
-        start_func(1)
-        status = self.wait_with_status(
-            signal_conditions=[(self.scan_control.scan_status.get, ScanControlScanStatus.RUNNING)],
-            timeout=self.timeout_for_pvwait,
-            check_stopped=True,
-        )
-        return status
-
-    def stage(self) -> list[object]:
-        """
-        Stage the device in preparation for a scan.
-
-        Returns:
-            list(object): list of objects that were staged
-        """
-        if self._staged != Staged.no:
-            return super().stage()
-        self._stopped = False
-        self.scaninfo.load_scan_metadata()
-        self.on_stage()
-        return super().stage()
-
-    def _check_scan_msg(self, target_state: ScanControlLoadMessage) -> None:
-        """Check if the scan message is gettting available
-
-        Args:
-            target_state (ScanControlLoadMessage): Target state to check for
-
-        Raises:
-            TimeoutError: If the scan message is not available after the timeout
-        """
-        state = self.scan_control.scan_msg.get()
-        if state != target_state:
-            logger.warning(
-                f"Resetting scan validation in stage for state: {ScanControlLoadMessage(state)}, "
-                f"retry .get() on scan_control: {ScanControlLoadMessage(self.scan_control.scan_msg.get())} and sleeping 1s"
-            )
-            self.scan_control.scan_val_reset.put(1)
-            # Sleep to ensure the reset is done
-            time.sleep(1)
-
-        if not self.wait_for_signals(
-            signal_conditions=[(self.scan_control.scan_msg.get, target_state)],
-            timeout=self.timeout_for_pvwait,
-            check_stopped=True,
-        ):
-            raise TimeoutError(
-                f"Timeout after {self.timeout_for_pvwait} while waiting for scan status,"
-                f" current state: {ScanControlScanStatus(self.scan_control.scan_msg.get())}"
-            )
-
-    def on_stage(self) -> None:
-        """Actions to be executed when the device is staged."""
-        if not self.scaninfo.scan_type == "fly":
-            return
-        self._check_scan_msg(ScanControlLoadMessage.PENDING)
-
-        scan_name = self.scaninfo.scan_msg.content["info"].get("scan_name", "")
-        self._update_scan_parameter()
-        if scan_name == "xas_simple_scan":
-            self.set_xas_settings(
-                low=self.scan_parameter.start,
-                high=self.scan_parameter.stop,
-                scan_time=self.scan_parameter.scan_time,
-            )
-            self.set_xrd_settings(
-                enable_low=False,
-                enable_high=False,
-                num_trigger_low=0,
-                num_trigger_high=0,
-                exp_time_low=0,
-                exp_time_high=0,
-                cycle_low=0,
-                cycle_high=0,
-            )
-            self.set_scan_control_settings(
-                mode=ScanControlMode.SIMPLE, scan_duration=self.scan_parameter.scan_duration
-            )
-        elif scan_name == "xas_simple_scan_with_xrd":
-            self.set_xas_settings(
-                low=self.scan_parameter.start,
-                high=self.scan_parameter.stop,
-                scan_time=self.scan_parameter.scan_time,
-            )
-            self.set_xrd_settings(
-                enable_low=self.scan_parameter.xrd_enable_low,
-                enable_high=self.scan_parameter.xrd_enable_high,
-                num_trigger_low=self.scan_parameter.num_trigger_low,
-                num_trigger_high=self.scan_parameter.num_trigger_high,
-                exp_time_low=self.scan_parameter.exp_time_low,
-                exp_time_high=self.scan_parameter.exp_time_high,
-                cycle_low=self.scan_parameter.cycle_low,
-                cycle_high=self.scan_parameter.cycle_high,
-            )
-            self.set_scan_control_settings(
-                mode=ScanControlMode.SIMPLE, scan_duration=self.scan_parameter.scan_duration
-            )
-        else:
-            raise Mo1BraggError(
-                f"Scan mode {scan_name} not implemented for scan_type={self.scaninfo.scan_type} on  device {self.name}"
-            )
-        # Load the scan parameters to the controller
-        self.scan_control.scan_load.put(1)
-        # Wait for params to be checked from controller
-        if not self.wait_for_signals(
-            signal_conditions=[(self.scan_control.scan_msg.get, ScanControlLoadMessage.SUCCESS)],
-            timeout=self.timeout_for_pvwait,
-            check_stopped=True,
-        ):
-            raise TimeoutError(
-                f"Scan parameter validation run into timeout after {self.timeout_for_pvwait} with {ScanControlLoadMessage(self.scan_control.scan_msg.get())}"
-            )
-
-    def complete(self) -> DeviceStatus:
-        """Complete the acquisition.
-
-        The method returns a DeviceStatus object that resolves to set_finished or set_exception once the acquisition is completed.
-        """
-        status = self.on_complete()
-        if isinstance(status, DeviceStatus):
-            return status
-        status = DeviceStatus(self)
-        status.set_finished()
-        return status
-
-    def on_complete(self) -> DeviceStatus:
-        """Specify actions to be performed for the completion of the acquisition."""
-        status = self.wait_with_status(
-            signal_conditions=[(self.scan_control.scan_done.get, 1)],
-            timeout=None,
-            check_stopped=True,
-        )
-        return status
-
-    def unstage(self) -> list[object]:
-        """
-        Unstage device after a scan. It has to be possible to call this multiple times.
-
-        Returns:
-            list(object): list of objects that were unstaged
-        """
-        self.check_scan_id()
-        self._stopped = False
-        self.on_unstage()
-        return super().unstage()
-
-    def on_unstage(self) -> None:
-        """Actions to be executed when the device is unstaged.
-        The checks here ensure that the controller resets the Scan_msg to PENDING state."""
-        if self.wait_for_signals(
-            signal_conditions=[(self.scan_control.scan_msg.get, ScanControlLoadMessage.PENDING)],
-            timeout=self.timeout_for_pvwait,
-            check_stopped=False,
-        ):
-            return
-
-        self.scan_control.scan_val_reset.put(1)
-        if not self.wait_for_signals(
-            signal_conditions=[(self.scan_control.scan_msg.get, ScanControlLoadMessage.PENDING)],
-            timeout=self.timeout_for_pvwait,
-            check_stopped=False,
-        ):
-            raise TimeoutError(
-                f"Timeout after {self.timeout_for_pvwait} while waiting for scan validation"
-            )
-
-    # -------------- End Flyer Interface methods -----------------#
-
-    # -------------- Utility methods -----------------#
-
-    def check_scan_id(self) -> None:
-        """Checks if scan_id has changed and set stopped flagged to True if it has."""
-        old_scan_id = self.scaninfo.scan_id
-        self.scaninfo.load_scan_metadata()
-        if self.scaninfo.scan_id != old_scan_id:
-            self._stopped = True
-
-    def wait_for_signals(
-        self,
-        signal_conditions: list[tuple],
-        timeout: float | None = None,
-        check_stopped: bool = False,
-        interval: float = 0.05,
-        all_signals: bool = False,
-    ) -> bool:
-        """Wrapper around a list of conditions that allows waiting for them to become True.
-        For EPICs PVs, an example usage is pasted at the bottom.
-
-        Args:
-            signal_conditions (list[tuple]): tuple of executable calls for conditions (get_current_state, condition) to check
-            timeout (float): timeout in seconds
-            check_stopped (bool): True if stopped flag should be checked. The function relies on the self.stopped property to be set
-            interval (float): interval in seconds
-            all_signals (bool): True if all signals should be True, False if any signal should be True
-
-        Returns:
-            bool: True if all signals are in the desired state, False if timeout is reached
-
-            >>> Example usage for EPICS PVs:
-            >>> self.wait_for_signals(signal_conditions=[(self.acquiring.get, False)], timeout=5, interval=0.05, check_stopped=True, all_signals=True)
-        """
-
-        timer = 0
-        while True:
-            checks = [
-                get_current_state() == condition
-                for get_current_state, condition in signal_conditions
-            ]
-            if check_stopped is True and self.stopped is True:
-                return False
-            if (all_signals and all(checks)) or (not all_signals and any(checks)):
-                return True
-            if timeout and timer > timeout:
-                return False
-            time.sleep(interval)
-            timer += interval
-
-    def wait_with_status(
-        self,
-        signal_conditions: list[tuple],
-        timeout: float | None = None,
-        check_stopped: bool = False,
-        interval: float = 0.05,
-        all_signals: bool = False,
-        exception_on_timeout: Exception = None,
-    ) -> DeviceStatus:
-        """Wrapper around wait_for_signals to be started in thread and attach a DeviceStatus object.
-        This allows BEC to perform actinos in parallel and not be blocked by method calls on a device.
-        Typically used for on_trigger, on_complete methods or also the kickoff.
-
-        Args:
-            signal_conditions (list[tuple]): tuple of executable calls for conditions (get_current_state, condition) to check
-            timeout (float): timeout in seconds
-            check_stopped (bool): True if stopped flag should be checked
-            interval (float): interval in seconds
-            all_signals (bool): True if all signals should be True, False if any signal should be True
-            exception_on_timeout (Exception): Exception to raise on timeout
-
-        Returns:
-            DeviceStatus: DeviceStatus object that resolves either to set_finished or set_exception
-        """
-        if exception_on_timeout is None:
-            exception_on_timeout = DeviceTimeoutError(
-                f"Timeout error for {self.name} while waiting for signals {signal_conditions}"
-            )
-
-        status = DeviceStatus(device=self)
-
-        def wait_for_signals_wrapper(
-            status: DeviceStatus,
-            signal_conditions: list[tuple],
-            timeout: float,
-            check_stopped: bool,
-            interval: float,
-            all_signals: bool,
-            exception_on_timeout: Exception = None,
-        ):
-            """Convenient wrapper around wait_for_signals to set status based on the result.
-
-            Args:
-                status (DeviceStatus): DeviceStatus object to be set
-                signal_conditions (list[tuple]): tuple of executable calls for conditions (get_current_state, condition) to check
-                timeout (float): timeout in seconds
-                check_stopped (bool): True if stopped flag should be checked
-                interval (float): interval in seconds
-                all_signals (bool): True if all signals should be True, False if any signal should be True
-                exception_on_timeout (Exception): Exception to raise on timeout
-            """
-            try:
-                result = self.wait_for_signals(
-                    signal_conditions, timeout, check_stopped, interval, all_signals
-                )
-                if result is True:
-                    # pylint: disable=protected-access
-                    status.set_finished()
-                else:
-                    if self.stopped:
-                        # INFO This will execute a callback to the parent device.stop() method
-                        status.set_exception(exc=DeviceStopError(f"{self.name} was stopped"))
-                    else:
-                        # INFO This will execute a callback to the parent device.stop() method
-                        status.set_exception(exc=exception_on_timeout)
-            # pylint: disable=broad-except
-            except Exception as exc:
-                content = traceback.format_exc()
-                logger.warning(f"Error in wait_for_signals in {self.name}; Traceback: {content}")
-                # INFO This will execute a callback to the parent device.stop() method
-                status.set_exception(exc=exc)
-
-        thread = threading.Thread(
-            target=wait_for_signals_wrapper,
-            args=(
-                status,
-                signal_conditions,
-                timeout,
-                check_stopped,
-                interval,
-                all_signals,
-                exception_on_timeout,
-            ),
-            daemon=True,
-        )
-        thread.start()
-        return status

debye_bec/devices/test_utils/utils.py

@@ -1,9 +0,0 @@
-def patch_dual_pvs(device):
-    device.wait_for_connection(all_signals=True)
-    for walk in device.walk_signals():
-        if not hasattr(walk.item, "_read_pv"):
-            continue
-        if not hasattr(walk.item, "_write_pv"):
-            continue
-        if walk.item._read_pv.pvname.endswith("_RBV"):
-            walk.item._read_pv = walk.item._write_pv

debye_bec/scans/__init__.py

@@ -1 +0,0 @@
-from .mono_bragg_scans import XASSimpleScan, XASSimpleScanWithXRD

debye_bec/scans/mono_bragg_scans.py

@@ -1,167 +0,0 @@
-""" This module contains the scan classes for the mono bragg motor of the Debye beamline."""
-
-import time
-
-import numpy as np
-from bec_lib.device import DeviceBase
-from bec_server.scan_server.scans import AsyncFlyScanBase
-
-
-class XASSimpleScan(AsyncFlyScanBase):
-
-    scan_name = "xas_simple_scan"
-    scan_type = "fly"
-    scan_report_hint = "device_progress"
-    required_kwargs = []
-    use_scan_progress_report = False
-    pre_move = False
-    gui_config = {
-        "Movement Parameters": ["start", "stop"],
-        "Scan Parameters": ["scan_time", "scan_duration"],
-    }
-
-    def __init__(
-        self,
-        start: float,
-        stop: float,
-        scan_time: float,
-        scan_duration: float,
-        motor: DeviceBase = "mo1_bragg",
-        **kwargs,
-    ):
-        """The xas_simple_scan is used to start a simple oscillating scan on the mono bragg motor.
-        Start and Stop define the energy range for the scan, scan_time is the time for one scan cycle and scan_duration
-        is the duration of the scan. If scan duration is set to 0, the scan will run infinitely.
-
-        Args:
-            start (float): Start energy for the scan.
-            stop (float): Stop energy for the scan.
-            scan_time (float): Time for one scan cycle.
-            scan_duration (float): Duration of the scan.
-            motor (DeviceBase, optional): Motor device to be used for the scan. Defaults to "mo1_bragg".
-        Examples:
-            >>> scans.xas_simple_scan(start=8000, stop=9000, scan_time=1, scan_duration=10)
-        """
-        super().__init__(**kwargs)
-        self.motor = motor
-        self.start = start
-        self.stop = stop
-        self.scan_time = scan_time
-        self.scan_duration = scan_duration
-        self.primary_readout_cycle = 1
-
-    def prepare_positions(self):
-        """Prepare the positions for the scan.
-
-        Use here only start and end energy defining the range for the scan.
-        """
-        self.positions = np.array([self.start, self.stop], dtype=float)
-        self.num_pos = None
-        yield None
-
-    def pre_scan(self):
-        """Pre Scan action. Ensure the motor movetype is set to energy, then check limits for start/end energy.
-        #TODO Remove once the motor movetype is removed and ANGLE motion is a pseudo motor.
-        """
-        yield from self.stubs.send_rpc_and_wait(self.motor, "move_type.set", "energy")
-
-        self._check_limits()
-        # Ensure parent class pre_scan actions to be called.
-        super().pre_scan()
-
-    def scan_report_instructions(self):
-        """
-        Return the instructions for the scan report.
-        """
-        yield from self.stubs.scan_report_instruction({"device_progress": [self.motor]})
-
-    def scan_core(self):
-        """Run the scan core.
-        Kickoff the oscillation on the Bragg motor and wait for the completion of the motion.
-        """
-        # Start the oscillation on the Bragg motor.
-        yield from self.stubs.kickoff(device=self.motor)
-        yield from self.stubs.complete(device=self.motor)
-
-        # Get the target DIID (instruction number) for the stubs.complete call
-        target_diid = self.DIID - 1
-
-        while True:
-            # Readout monitored devices
-            yield from self.stubs.read_and_wait(group="primary", wait_group="readout_primary")
-            # Check if complete call on Mo1 Bragg has been finished
-            status = self.stubs.get_req_status(
-                device=self.motor, RID=self.metadata["RID"], DIID=target_diid
-            )
-            if status:
-                break
-            time.sleep(self.primary_readout_cycle)
-            self.point_id += 1
-
-        self.num_pos = self.point_id + 1
-
-
-class XASSimpleScanWithXRD(XASSimpleScan):
-    scan_name = "xas_simple_scan_with_xrd"
-    gui_config = {
-        "Movement Parameters": ["start", "stop"],
-        "Scan Parameters": ["scan_time", "scan_duration"],
-        "Low Energy Range": ["xrd_enable_low", "num_trigger_low", "exp_time_low", "cycle_low"],
-        "High Energy Range": ["xrd_enable_high", "num_trigger_high", "exp_time_high", "cycle_high"],
-    }
-
-    def __init__(
-        self,
-        start: float,
-        stop: float,
-        scan_time: float,
-        scan_duration: float,
-        xrd_enable_low: bool,
-        num_trigger_low: int,
-        exp_time_low: float,
-        cycle_low: int,
-        xrd_enable_high: bool,
-        num_trigger_high: int,
-        exp_time_high: float,
-        cycle_high: float,
-        motor: DeviceBase = "mo1_bragg",
-        **kwargs,
-    ):
-        """The xas_simple_scan_with_xrd is an oscillation motion on the mono motor
-        with XRD triggering at low and high energy ranges.
-        If scan duration is set to 0, the scan will run infinitely.
-
-        Args:
-            start (float): Start energy for the scan.
-            stop (float): Stop energy for the scan.
-            scan_time (float): Time for one oscillation .
-            scan_duration (float): Total duration of the scan.
-            xrd_enable_low (bool): Enable XRD triggering for the low energy range.
-            num_trigger_low (int): Number of triggers for the low energy range.
-            exp_time_low (float): Exposure time for the low energy range.
-            cycle_low (int): Specify how often the triggers should be considered, every nth cycle for low
-            xrd_enable_high (bool): Enable XRD triggering for the high energy range.
-            num_trigger_high (int): Number of triggers for the high energy range.
-            exp_time_high (float): Exposure time for the high energy range.
-            cycle_high (int): Specify how often the triggers should be considered, every nth cycle for high
-            motor (DeviceBase, optional): Motor device to be used for the scan. Defaults to "mo1_bragg".
-        
-        Examples:
-            >>> scans.xas_simple_scan_with_xrd(start=8000, stop=9000, scan_time=1, scan_duration=10, xrd_enable_low=True, num_trigger_low=5, cycle_low=2, exp_time_low=100, xrd_enable_high=False, num_trigger_high=3, cycle_high=1, exp_time_high=1000)
-            """
-        super().__init__(
-            start=start,
-            stop=stop,
-            scan_time=scan_time,
-            scan_duration=scan_duration,
-            motor=motor,
-            **kwargs,
-        )
-        self.xrd_enable_low = xrd_enable_low
-        self.num_trigger_low = num_trigger_low
-        self.exp_time_low = exp_time_low
-        self.cycle_low = cycle_low
-        self.xrd_enable_high = xrd_enable_high
-        self.num_trigger_high = num_trigger_high
-        self.exp_time_high = exp_time_high
-        self.cycle_high = cycle_high

debye_bec/scans/scan_plugin_template.py

@@ -1,33 +0,0 @@
-"""
-SCAN PLUGINS
-
-All new scans should be derived from ScanBase. ScanBase provides various methods that can be customized and overriden
-but they are executed in a specific order:
-
-- self.initialize                        # initialize the class if needed
-- self.read_scan_motors                  # used to retrieve the start position (and the relative position shift if needed)
-- self.prepare_positions                 # prepare the positions for the scan. The preparation is split into multiple sub fuctions:
-    - self._calculate_positions          # calculate the positions
-    - self._set_positions_offset         # apply the previously retrieved scan position shift (if needed)
-    - self._check_limits                 # tests to ensure the limits won't be reached
-- self.open_scan                         # send an open_scan message including the scan name, the number of points and the scan motor names
-- self.stage                             # stage all devices for the upcoming acquisiton
-- self.run_baseline_readings             # read all devices to get a baseline for the upcoming scan
-- self.pre_scan                          # perform additional actions before the scan starts
-- self.scan_core                         # run a loop over all position
-    - self._at_each_point(ind, pos)      # called at each position with the current index and the target positions as arguments
-- self.finalize                          # clean up the scan, e.g. move back to the start position; wait everything to finish
-- self.unstage                           # unstage all devices that have been staged before
-- self.cleanup                           # send a close scan message and perform additional cleanups if needed
-"""
-
-# import time
-
-# import numpy as np
-
-# from bec_lib import bec_logger, messages
-# from bec_lib.endpoints import MessageEndpoints
-# from bec_server.scan_server.errors import ScanAbortion
-# from bec_server.scan_server.scans import FlyScanBase, RequestBase, ScanArgType, ScanBase
-
-# logger = bec_logger.logger

pyproject.toml

@@ -1,70 +0,0 @@
-[build-system]
-requires = ["hatchling"]
-build-backend = "hatchling.build"
-
-[project]
-name = "debye_bec"
-version = "0.0.0"
-description = "Custom device implementations based on the ophyd hardware abstraction layer"
-requires-python = ">=3.10"
-classifiers = [
-    "Development Status :: 3 - Alpha",
-    "Programming Language :: Python :: 3",
-    "Topic :: Scientific/Engineering",
-]
-dependencies = ["numpy", "bec_lib", "h5py", "ophyd_devices"]
-
-[project.optional-dependencies]
-dev = [
-    "bec_server",
-    "black",
-    "isort",
-    "coverage",
-    "pylint",
-    "pytest",
-    "pytest-random-order",
-]
-
-[project.entry-points."bec"]
-plugin_bec = "debye_bec"
-
-[project.entry-points."bec.deployment.device_server"]
-plugin_ds_startup = "debye_bec.deployments.device_server.startup:run"
-
-[project.entry-points."bec.file_writer"]
-plugin_file_writer = "debye_bec.file_writer"
-
-[project.entry-points."bec.scans"]
-plugin_scans = "debye_bec.scans"
-
-[project.entry-points."bec.ipython_client_startup"]
-plugin_ipython_client_pre = "debye_bec.bec_ipython_client.startup.pre_startup"
-plugin_ipython_client_post = "debye_bec.bec_ipython_client.startup"
-
-[project.entry-points."bec.widgets"]
-plugin_widgets = "debye_bec.bec_widgets"
-
-[tool.hatch.build.targets.wheel]
-include = ["*"]
-
-[tool.isort]
-profile = "black"
-line_length = 100
-multi_line_output = 3
-include_trailing_comma = true
-
-[tool.black]
-line-length = 100
-skip-magic-trailing-comma = true
-
-[tool.pylint.basic]
-# Good variable names regexes, separated by a comma. If names match any regex,
-# they will always be accepted
-good-names-rgxs = [
-    ".*scanID.*",
-    ".*RID.*",
-    ".*pointID.*",
-    ".*ID.*",
-    ".*_2D.*",
-    ".*_1D.*",
-]

setup.cfg

@@ -0,0 +1,21 @@
+[metadata]
+name = bec_plugins
+description = BEC plugins to modify the behaviour of services within the BEC framework
+long_description = file: README.md
+long_description_content_type = text/markdown
+url = https://gitlab.psi.ch/bec/bec
+project_urls =
+    Bug Tracker = https://gitlab.psi.ch/bec/bec/issues
+classifiers =
+    Programming Language :: Python :: 3
+    Development Status :: 3 - Alpha
+    Topic :: Scientific/Engineering
+
+[options]
+package_dir =
+    = .
+packages = find:
+python_requires = >=3.10
+
+[options.packages.find]
+where = .

setup.py

@@ -0,0 +1,7 @@
+from setuptools import setup
+
+if __name__ == "__main__":
+    setup(
+        install_requires=["numpy", "h5py", "bec-lib"],
+        extras_require={"dev": ["pytest", "pytest-random-order", "pytest-redis", "coverage"]},
+    )

tests/test_file_writer_service.py

@@ -2,12 +2,10 @@ from unittest import mock
 
 import numpy as np
 import pytest
-from bec_lib import messages
-from bec_lib.endpoints import MessageEndpoints
+from bec_lib import MessageEndpoints, ServiceConfig, messages
 from bec_lib.redis_connector import MessageObject
-from bec_lib.service_config import ServiceConfig
 
-from debye_bec.services.NIDAQ_writer import NIDAQWriterService
+from bec_plugins.services.NIDAQ_writer import NIDAQWriterService
 
 
 def test_nidaq_starts_consumers():
@@ -34,7 +32,8 @@ class NIWriterMock(NIDAQWriterService):
 @pytest.fixture(scope="function")
 def nidaq():
     service = NIWriterMock(
-        config=ServiceConfig(redis={"host": "test", "port": 6379}), connector_cls=mock.MagicMock()
+        config=ServiceConfig(redis={"host": "test", "port": 6379}),
+        connector_cls=mock.MagicMock(),
     )
     yield service
 
@@ -47,8 +46,11 @@ def test_nidaq_scan_status_consumer(nidaq):
 
 
 def test_scan_status_callback(nidaq):
-    scan_status_msg = messages.ScanStatusMessage(scan_id="test", status="open", info={})
-    msg_obj = MessageObject(topic="test", value=scan_status_msg.dumps())
+    scan_status_msg = messages.ScanStatusMessage(scanID="test", status="open", info={})
+    msg_obj = MessageObject(
+        topic="test",
+        value=scan_status_msg.dumps(),
+    )
     with mock.patch.object(nidaq, "handle_scan_status") as mock_handle:
         nidaq._scan_status_callback(msg_obj, nidaq)
         mock_handle.assert_called_once_with(scan_status_msg)
@@ -81,10 +83,10 @@ def test_nidaq_reads_data_from_strea(nidaq):
             mock_handle.assert_called_once()
 
 
-@pytest.mark.parametrize("scan_status", ["open", "closed", "aborted", "halted"])
+@pytest.mark.parametrize("scan_status", ["open", "closed", "aborted", "halted", None])
 def test_nidaq_handle_scan_status(nidaq, scan_status):
     scan_status_msg = messages.ScanStatusMessage(
-        scan_id="test", status=scan_status, info={"scan_number": 5}
+        scanID="test", status=scan_status, info={"scan_number": 5}
     )
     nidaq.handle_scan_status(scan_status_msg)
     if scan_status == "open":
@@ -97,14 +99,9 @@ def test_nidaq_handle_scan_status(nidaq, scan_status):
 
 def test_nidaq_handle_ni_data(nidaq):
     data = [
+        messages.DeviceMessage(signals={"signal1": list(range(10)), "signal2": list(range(10))}),
         messages.DeviceMessage(
-            signals={"signal1": {"value": list(range(10))}, "signal2": {"value": list(range(10))}}
-        ),
-        messages.DeviceMessage(
-            signals={
-                "signal1": {"value": list(range(10, 20))},
-                "signal2": {"value": list(range(10, 20))},
-            }
+            signals={"signal1": list(range(10, 20)), "signal2": list(range(10, 20))}
         ),
     ]
 
@@ -117,7 +114,7 @@ def test_nidaq_handle_ni_data(nidaq):
 def test_nidaq_write_data_without_filename(nidaq):
     signal = {"signal1": np.asarray(range(20)), "signal2": np.asarray(range(20))}
 
-    with mock.patch("debye_bec.services.NIDAQ_writer.NIDAQ_writer.h5py") as mock_h5py:
+    with mock.patch("bec_plugins.services.NIDAQ_writer.NIDAQ_writer.h5py") as mock_h5py:
         nidaq.write_data(signal)
         mock_h5py.File.assert_not_called()
 
@@ -125,7 +122,7 @@ def test_nidaq_write_data_without_filename(nidaq):
 def test_nidaq_write_data_with_filename(nidaq):
     signal = {"signal1": np.asarray(range(20)), "signal2": np.asarray(range(20))}
     nidaq.filename = "test.h5"
-    with mock.patch("debye_bec.services.NIDAQ_writer.NIDAQ_writer.h5py") as mock_h5py:
+    with mock.patch("bec_plugins.services.NIDAQ_writer.NIDAQ_writer.h5py") as mock_h5py:
         nidaq.write_data(signal)
         mock_h5py.File.assert_called_once_with("test.h5", "a")
 
@@ -134,7 +131,7 @@ def test_nidaq_write_data_reshape(nidaq):
     signal = {"signal1": np.asarray(range(20)), "signal2": np.asarray(range(20))}
     nidaq.filename = "test.h5"
     nidaq.reshape_dataset = True
-    with mock.patch("debye_bec.services.NIDAQ_writer.NIDAQ_writer.h5py") as mock_h5py:
+    with mock.patch("bec_plugins.services.NIDAQ_writer.NIDAQ_writer.h5py") as mock_h5py:
         nidaq.write_data(signal)
         mock_h5py.File.assert_called_once_with("test.h5", "a")
 
@@ -143,7 +140,7 @@ def test_nidaq_write_data_without_reshape(nidaq):
     signal = {"signal1": np.asarray(range(20)), "signal2": np.asarray(range(20))}
     nidaq.filename = "test.h5"
     nidaq.reshape_dataset = False
-    with mock.patch("debye_bec.services.NIDAQ_writer.NIDAQ_writer.h5py") as mock_h5py:
+    with mock.patch("bec_plugins.services.NIDAQ_writer.NIDAQ_writer.h5py") as mock_h5py:
         nidaq.write_data(signal)
         mock_h5py.File.assert_called_once_with("test.h5", "a")
         file_handle = mock_h5py.File().__enter__()
@@ -164,7 +161,7 @@ def test_nidaq_write_data_reshapes_data(nidaq):
     signal = {"signal1": np.asarray(range(20)), "signal2": np.asarray(range(20))}
     nidaq.filename = "test.h5"
     nidaq.reshape_dataset = True
-    with mock.patch("debye_bec.services.NIDAQ_writer.NIDAQ_writer.h5py") as mock_h5py:
+    with mock.patch("bec_plugins.services.NIDAQ_writer.NIDAQ_writer.h5py") as mock_h5py:
         file_handle = mock_h5py.File().__enter__()
         file_handle.__contains__.side_effect = signal.__contains__
         nidaq.write_data(signal)

tests/tests_bec_ipython_client/README.md

@@ -1,31 +0,0 @@
-# Getting Started with Testing using pytest
-
-BEC is using the [pytest](https://docs.pytest.org/en/8.0.x/) framework. 
-It can be install via 
-``` bash
-pip install pytest
-```
-in your *python environment*. 
-We note that pytest is part of the optional-dependencies `[dev]` of the plugin package.
-
-## Introduction
-
-Tests in this package should be stored in the `tests` directory.
-We suggest to sort tests of different submodules, i.e. `scans` or `devices` in the respective folder structure, and to folow a naming convention of `<test_module_name.py>`.
-
-To run all tests, navigate to the directory of the plugin from the command line, and run the command 
-
-``` bash
-pytest -v --random-order ./tests
-```
-Note, the python environment needs to be active.
-The additional arg `-v` allows pytest to run in verbose mode which provides more detailed information about the tests being run.
-The argument `--random-order` instructs pytest to run the tests in random order, which is the default in the CI pipelines. 
-
-## Test examples
-
-Writing tests can be quite specific for the given function. 
-We recommend writing tests as isolated as possible, i.e. try to test single functions instead of full classes.
-A very useful class to enable isolated testing is [MagicMock](https://docs.python.org/3/library/unittest.mock.html).
-In addition, we also recommend to take a look at the [How-to guides from pytest](https://docs.pytest.org/en/8.0.x/how-to/index.html).
-

tests/tests_bec_widgets/README.md

@@ -1,31 +0,0 @@
-# Getting Started with Testing using pytest
-
-BEC is using the [pytest](https://docs.pytest.org/en/8.0.x/) framework. 
-It can be install via 
-``` bash
-pip install pytest
-```
-in your *python environment*. 
-We note that pytest is part of the optional-dependencies `[dev]` of the plugin package.
-
-## Introduction
-
-Tests in this package should be stored in the `tests` directory.
-We suggest to sort tests of different submodules, i.e. `scans` or `devices` in the respective folder structure, and to folow a naming convention of `<test_module_name.py>`.
-
-To run all tests, navigate to the directory of the plugin from the command line, and run the command 
-
-``` bash
-pytest -v --random-order ./tests
-```
-Note, the python environment needs to be active.
-The additional arg `-v` allows pytest to run in verbose mode which provides more detailed information about the tests being run.
-The argument `--random-order` instructs pytest to run the tests in random order, which is the default in the CI pipelines. 
-
-## Test examples
-
-Writing tests can be quite specific for the given function. 
-We recommend writing tests as isolated as possible, i.e. try to test single functions instead of full classes.
-A very useful class to enable isolated testing is [MagicMock](https://docs.python.org/3/library/unittest.mock.html).
-In addition, we also recommend to take a look at the [How-to guides from pytest](https://docs.pytest.org/en/8.0.x/how-to/index.html).
-

tests/tests_devices/README.md

@@ -1,31 +0,0 @@
-# Getting Started with Testing using pytest
-
-BEC is using the [pytest](https://docs.pytest.org/en/8.0.x/) framework. 
-It can be install via 
-``` bash
-pip install pytest
-```
-in your *python environment*. 
-We note that pytest is part of the optional-dependencies `[dev]` of the plugin package.
-
-## Introduction
-
-Tests in this package should be stored in the `tests` directory.
-We suggest to sort tests of different submodules, i.e. `scans` or `devices` in the respective folder structure, and to folow a naming convention of `<test_module_name.py>`.
-
-To run all tests, navigate to the directory of the plugin from the command line, and run the command 
-
-``` bash
-pytest -v --random-order ./tests
-```
-Note, the python environment needs to be active.
-The additional arg `-v` allows pytest to run in verbose mode which provides more detailed information about the tests being run.
-The argument `--random-order` instructs pytest to run the tests in random order, which is the default in the CI pipelines. 
-
-## Test examples
-
-Writing tests can be quite specific for the given function. 
-We recommend writing tests as isolated as possible, i.e. try to test single functions instead of full classes.
-A very useful class to enable isolated testing is [MagicMock](https://docs.python.org/3/library/unittest.mock.html).
-In addition, we also recommend to take a look at the [How-to guides from pytest](https://docs.pytest.org/en/8.0.x/how-to/index.html).
-

tests/tests_devices/test_mo1_bragg.py

@@ -1,436 +0,0 @@
-# pylint: skip-file
-import os
-import threading
-import time
-from dataclasses import fields
-from unittest import mock
-
-import ophyd
-import pytest
-from bec_lib.messages import ScanQueueMessage, ScanStatusMessage
-from bec_server.scan_server.scan_assembler import ScanAssembler
-from bec_server.scan_server.scan_queue import RequestBlock
-from bec_server.scan_server.scan_worker import ScanWorker
-from bec_server.scan_server.tests.fixtures import scan_server_mock
-from ophyd.utils import LimitError
-from ophyd_devices.tests.utils import MockPV
-
-# from bec_server.device_server.tests.utils import DMMock
-from debye_bec.devices.mo1_bragg import (
-    Mo1Bragg,
-    Mo1BraggError,
-    MoveType,
-    ScanControlLoadMessage,
-    ScanControlMode,
-    ScanControlScanStatus,
-)
-
-# TODO move this function to ophyd_devices, it is duplicated in csaxs_bec and needed for other pluging repositories
-from debye_bec.devices.test_utils.utils import patch_dual_pvs
-
-
-@pytest.fixture(scope="function")
-def scan_worker_mock(scan_server_mock):
-    scan_server_mock.device_manager.connector = mock.MagicMock()
-    scan_worker = ScanWorker(parent=scan_server_mock)
-    yield scan_worker
-
-
-@pytest.fixture(scope="function")
-def mock_bragg():
-    name = "bragg"
-    prefix = "X01DA-OP-MO1:BRAGG:"
-    with (
-        mock.patch.object(ophyd, "cl") as mock_cl,
-        mock.patch("debye_bec.devices.mo1_bragg.Mo1Bragg", "_on_init"),
-    ):
-        mock_cl.get_pv = MockPV
-        mock_cl.thread_class = threading.Thread
-        dev = Mo1Bragg(name=name, prefix=prefix)
-        patch_dual_pvs(dev)
-        yield dev
-
-
-def test_init(mock_bragg):
-    dev = mock_bragg
-    assert dev.name == "bragg"
-    assert dev.prefix == "X01DA-OP-MO1:BRAGG:"
-    assert dev.move_type.get() == MoveType.ENERGY
-    assert dev.crystal.offset_si111._read_pvname == "X01DA-OP-MO1:BRAGG:offset_si111_RBV"
-    assert dev.move_abs._read_pvname == "X01DA-OP-MO1:BRAGG:move_abs.PROC"
-
-
-def test_check_value(mock_bragg):
-    dev = mock_bragg
-    dev.low_lim._read_pv.mock_data = 0
-    dev.high_lim._read_pv.mock_data = 1
-    dev.low_limit_angle._read_pv.mock_data = 10
-    dev.high_limit_angle._read_pv.mock_data = 20
-    # Check that limits are taken correctly from angle or energy
-    # Energy first
-    move_type = MoveType.ENERGY
-    dev.move_type.set(move_type)
-    # nothing happens
-    dev.check_value(0.5)
-    with pytest.raises(LimitError):
-        dev.check_value(15)
-    # Angle next
-    move_type = MoveType.ANGLE
-    dev.move_type.set(move_type)
-    dev.check_value(15)
-    with pytest.raises(LimitError):
-        dev.check_value(0.5)
-
-
-def test_egu(mock_bragg):
-    dev = mock_bragg
-    assert dev.egu == "eV"
-    dev.move_type.set(MoveType.ANGLE)
-    assert dev.egu == "deg"
-
-
-def test_move_succeeds(mock_bragg):
-    dev = mock_bragg
-    dev.move_abs._read_pv.mock_data = 0
-    # Move succeeds
-    with mock.patch.object(dev.motor_is_moving._read_pv, "mock_data", side_effect=[0, 1]):
-        status = dev.move(0.5)
-        # Sleep needed to allow thread to resolive in _move_and_finish, i.e. and the 0.25s sleep inside the function
-        time.sleep(1)
-        assert status.done is True
-        assert status.success is True
-        assert dev.setpoint.get() == 0.5
-        assert dev.move_abs.get() == 1
-
-
-def test_stop_move(mock_bragg):
-    dev = mock_bragg
-    dev.move_abs._read_pv.mock_data = 0
-    dev.motor_is_moving._read_pv.mock_data = 0
-    # Move fails
-    status = dev.move(0.5)
-    time.sleep(1)
-    assert status.done is False
-    assert dev._stopped == False
-    dev.stop()
-    time.sleep(0.5)
-    assert dev._stopped == True
-    assert status.done is True
-    assert status.success is False
-
-
-def test_set_xtal(mock_bragg):
-    dev = mock_bragg
-    dev.set_xtal("111")
-    # Default values for mock
-    assert dev.crystal.offset_si111.get() == 0
-    assert dev.crystal.offset_si311.get() == 0
-    assert dev.crystal.d_spacing_si111.get() == 0
-    assert dev.crystal.d_spacing_si311.get() == 0
-    assert dev.crystal.xtal_enum.get() == 0
-    dev.set_xtal("311", offset_si111=1, offset_si311=2, d_spacing_si111=3, d_spacing_si311=4)
-    assert dev.crystal.offset_si111.get() == 1
-    assert dev.crystal.offset_si311.get() == 2
-    assert dev.crystal.d_spacing_si111.get() == 3
-    assert dev.crystal.d_spacing_si311.get() == 4
-    assert dev.crystal.xtal_enum.get() == 1
-
-
-def test_set_xas_settings(mock_bragg):
-    dev = mock_bragg
-    dev.move_type.set(MoveType.ENERGY)
-    dev.set_xas_settings(low=0.5, high=1, scan_time=0.1)
-    assert dev.scan_settings.s_scan_energy_lo.get() == 0.5
-    assert dev.scan_settings.s_scan_energy_hi.get() == 1
-    assert dev.scan_settings.s_scan_scantime.get() == 0.1
-    dev.move_type.set(MoveType.ANGLE)
-    dev.set_xas_settings(low=10, high=20, scan_time=1)
-    assert dev.scan_settings.s_scan_angle_lo.get() == 10
-    assert dev.scan_settings.s_scan_angle_hi.get() == 20
-    assert dev.scan_settings.s_scan_scantime.get() == 1
-
-
-def test_set_xrd_settings(mock_bragg):
-    dev = mock_bragg
-    dev.set_xrd_settings(
-        enable_low=True,
-        enable_high=False,
-        num_trigger_low=1,
-        num_trigger_high=7,
-        exp_time_low=1,
-        exp_time_high=3,
-        cycle_low=1,
-        cycle_high=5,
-    )
-    assert dev.scan_settings.xrd_enable_lo_enum.get() == True
-    assert dev.scan_settings.xrd_enable_hi_enum.get() == False
-    assert dev.scan_settings.xrd_n_trigger_lo.get() == 1
-    assert dev.scan_settings.xrd_n_trigger_hi.get() == 7
-    assert dev.scan_settings.xrd_time_lo.get() == 1
-    assert dev.scan_settings.xrd_time_hi.get() == 3
-    assert dev.scan_settings.xrd_every_n_lo.get() == 1
-    assert dev.scan_settings.xrd_every_n_hi.get() == 5
-
-
-def test_set_control_settings(mock_bragg):
-    dev = mock_bragg
-    dev.set_scan_control_settings(mode=ScanControlMode.SIMPLE, scan_duration=10)
-    assert dev.scan_control.scan_mode_enum.get() == ScanControlMode.SIMPLE
-    assert dev.scan_control.scan_duration.get() == 10
-    dev.set_scan_control_settings(mode=ScanControlMode.ADVANCED, scan_duration=5)
-    assert dev.scan_control.scan_mode_enum.get() == ScanControlMode.ADVANCED
-    assert dev.scan_control.scan_duration.get() == 5
-
-
-def test_update_scan_parameters(mock_bragg):
-    dev = mock_bragg
-    msg = ScanStatusMessage(
-        scan_id="my_scan_id",
-        status="closed",
-        info={
-            "kwargs": {
-                "start": 0,
-                "stop": 5,
-                "scan_time": 1,
-                "scan_duration": 10,
-                "xrd_enable_low": True,
-                "xrd_enable_high": False,
-                "num_trigger_low": 1,
-                "num_trigger_high": 7,
-                "exp_time_low": 1,
-                "exp_time_high": 3,
-                "cycle_low": 1,
-                "cycle_high": 5,
-            }
-        },
-        metadata={},
-    )
-    mock_bragg.scaninfo.scan_msg = msg
-    for field in fields(dev.scan_parameter):
-        assert getattr(dev.scan_parameter, field.name) == None
-    dev._update_scan_parameter()
-    for field in fields(dev.scan_parameter):
-        assert getattr(dev.scan_parameter, field.name) == msg.content["info"]["kwargs"].get(
-            field.name, None
-        )
-
-
-def test_kickoff_scan(mock_bragg):
-    dev = mock_bragg
-    dev.scan_control.scan_status._read_pv.mock_data = ScanControlScanStatus.READY
-    dev.scan_control.scan_duration._read_pv.mock_data = 1
-    dev.scan_control.scan_start_timer._read_pv.mock_data = 0
-    dev.scan_control.scan_start_infinite._read_pv.mock_data = 0
-    status = dev.kickoff()
-    assert status.done is False
-    dev.scan_control.scan_status._read_pv.mock_data = ScanControlScanStatus.RUNNING
-    time.sleep(0.2)
-    assert status.done is True
-    assert dev.scan_control.scan_start_timer.get() == 1
-
-    dev.scan_control.scan_duration._read_pv.mock_data = 0
-    dev.scan_control.scan_start_timer._read_pv.mock_data = 0
-    dev.scan_control.scan_start_infinite._read_pv.mock_data = 0
-    dev.scan_control.scan_start_infinite._read_pv.mock_data = 0
-    status = dev.kickoff()
-    dev.kickoff()
-    assert dev.scan_control.scan_start_infinite.get() == 1
-
-
-def test_complete(mock_bragg):
-    dev = mock_bragg
-    dev.scan_control.scan_done._read_pv.mock_data = 0
-    # Normal case
-    status = dev.complete()
-    assert status.done is False
-    assert status.success is False
-    dev.scan_control.scan_done._read_pv.mock_data = 1
-    time.sleep(0.2)
-    assert status.done is True
-    assert status.success is True
-
-    # Stop called case
-    dev.scan_control.scan_done._read_pv.mock_data = 0
-    status = dev.complete()
-    assert status.done is False
-    assert status.success is False
-    dev.stop()
-    time.sleep(0.2)
-    assert status.done is True
-    assert status.success is False
-
-
-def test_unstage(mock_bragg):
-    mock_bragg.timeout_for_pvwait = 0.5
-    mock_bragg.scan_control.scan_val_reset._read_pv.mock_data = 0
-    mock_bragg.scan_control.scan_msg._read_pv.mock_data = ScanControlLoadMessage.PENDING
-
-    with mock.patch.object(mock_bragg.scan_control.scan_val_reset, "put") as mock_put:
-        mock_bragg.unstage()
-        assert mock_put.call_count == 0
-        mock_bragg.scan_control.scan_msg._read_pv.mock_data = ScanControlLoadMessage.SUCCESS
-        with pytest.raises(TimeoutError):
-            mock_bragg.unstage()
-            assert mock_put.call_count == 1
-
-
-@pytest.mark.parametrize(
-    "msg",
-    [
-        ScanQueueMessage(
-            scan_type="monitor_scan",
-            parameter={
-                "args": {},
-                "kwargs": {"device": "mo1_bragg", "start": 0, "stop": 10, "relative": True},
-                "num_points": 100,
-            },
-            queue="primary",
-            metadata={"RID": "test1234"},
-        ),
-        ScanQueueMessage(
-            scan_type="xas_simple_scan",
-            parameter={
-                "args": {},
-                "kwargs": {
-                    "motor": "mo1_bragg",
-                    "start": 0,
-                    "stop": 10,
-                    "scan_time": 1,
-                    "scan_duration": 10,
-                },
-                "num_points": 100,
-            },
-            queue="primary",
-            metadata={"RID": "test1234"},
-        ),
-        ScanQueueMessage(
-            scan_type="xas_simple_scan_with_xrd",
-            parameter={
-                "args": {},
-                "kwargs": {
-                    "motor": "mo1_bragg",
-                    "start": 0,
-                    "stop": 10,
-                    "scan_time": 1,
-                    "scan_duration": 10,
-                    "xrd_enable_low": True,
-                    "xrd_enable_high": False,
-                    "num_trigger_low": 1,
-                    "num_trigger_high": 7,
-                    "exp_time_low": 1,
-                    "exp_time_high": 3,
-                    "cycle_low": 1,
-                    "cycle_high": 5,
-                },
-                "num_points": 10,
-            },
-            queue="primary",
-            metadata={"RID": "test1234"},
-        ),
-    ],
-)
-def test_stage(mock_bragg, scan_worker_mock, msg):
-    """This test is important to check that the stage method of the device is working correctly.
-    Changing the kwargs names in the scans is tightly linked to the logic on the device, thus
-    it is important to check that the stage method is working correctly for the current implementation.
-
-    Therefor, this test creates a scaninfo message using the scan.open_scan() method to always check
-    agains the currently implemented scans vs. the logic on the device"""
-    # Create a scaninfo message using scans the ScanQueueMessages above, 3 cases of fly scan; for the general case the procedure is not defined yet
-    worker = scan_worker_mock
-    scan_server = worker.parent
-    rb = RequestBlock(msg, assembler=ScanAssembler(parent=scan_server))
-    with mock.patch.object(worker, "current_instruction_queue_item"):
-        worker.scan_motors = []
-        worker.readout_priority = {
-            "monitored": [],
-            "baseline": [],
-            "async": [],
-            "continuous": [],
-            "on_request": [],
-        }
-        open_scan_msg = list(rb.scan.open_scan())[0]
-        worker._initialize_scan_info(rb, open_scan_msg, msg.content["parameter"].get("num_points"))
-        scan_status_msg = ScanStatusMessage(
-            scan_id="test1234", status="closed", info=worker.current_scan_info, metadata={}
-        )
-        mock_bragg.scaninfo.scan_msg = scan_status_msg
-
-        # Ensure that ScanControlLoadMessage is set to SUCCESS
-        mock_bragg.scan_control.scan_msg._read_pv.mock_data = ScanControlLoadMessage.SUCCESS
-        with (
-            mock.patch.object(mock_bragg.scaninfo, "load_scan_metadata") as mock_load_scan_metadata,
-            mock.patch.object(mock_bragg, "_check_scan_msg") as mock_check_scan_msg,
-            mock.patch.object(mock_bragg, "on_unstage"),
-        ):
-            scan_name = scan_status_msg.content["info"].get("scan_name", "")
-            # Chek the not implemented fly scan first, should raise Mo1BraggError
-            if scan_name not in ["xas_simple_scan", "xas_simple_scan_with_xrd"]:
-                with pytest.raises(Mo1BraggError):
-                    mock_bragg.stage()
-                assert mock_check_scan_msg.call_count == 1
-                assert mock_load_scan_metadata.call_count == 1
-            else:
-                with (
-                    mock.patch.object(mock_bragg, "set_xas_settings") as mock_xas_settings,
-                    mock.patch.object(mock_bragg, "set_xrd_settings") as mock_xrd_settings,
-                    mock.patch.object(
-                        mock_bragg, "set_scan_control_settings"
-                    ) as mock_set_scan_control_settings,
-                ):
-                    # Check xas_simple_scan
-                    if scan_name == "xas_simple_scan":
-                        mock_bragg.stage()
-                        assert mock_xas_settings.call_args == mock.call(
-                            low=scan_status_msg.content["info"]["kwargs"]["start"],
-                            high=scan_status_msg.content["info"]["kwargs"]["stop"],
-                            scan_time=scan_status_msg.content["info"]["kwargs"]["scan_time"],
-                        )
-                        assert mock_xrd_settings.call_args == mock.call(
-                            enable_low=False,
-                            enable_high=False,
-                            num_trigger_low=0,
-                            num_trigger_high=0,
-                            exp_time_low=0,
-                            exp_time_high=0,
-                            cycle_low=0,
-                            cycle_high=0,
-                        )
-                        assert mock_set_scan_control_settings.call_args == mock.call(
-                            mode=ScanControlMode.SIMPLE,
-                            scan_duration=scan_status_msg.content["info"]["kwargs"][
-                                "scan_duration"
-                            ],
-                        )
-                    # Check xas_simple_scan_with_xrd
-                    elif scan_name == "xas_simple_scan_with_xrd":
-                        mock_bragg.stage()
-                        assert mock_xas_settings.call_args == mock.call(
-                            low=scan_status_msg.content["info"]["kwargs"]["start"],
-                            high=scan_status_msg.content["info"]["kwargs"]["stop"],
-                            scan_time=scan_status_msg.content["info"]["kwargs"]["scan_time"],
-                        )
-                        assert mock_xrd_settings.call_args == mock.call(
-                            enable_low=scan_status_msg.content["info"]["kwargs"]["xrd_enable_low"],
-                            enable_high=scan_status_msg.content["info"]["kwargs"][
-                                "xrd_enable_high"
-                            ],
-                            num_trigger_low=scan_status_msg.content["info"]["kwargs"][
-                                "num_trigger_low"
-                            ],
-                            num_trigger_high=scan_status_msg.content["info"]["kwargs"][
-                                "num_trigger_high"
-                            ],
-                            exp_time_low=scan_status_msg.content["info"]["kwargs"]["exp_time_low"],
-                            exp_time_high=scan_status_msg.content["info"]["kwargs"][
-                                "exp_time_high"
-                            ],
-                            cycle_low=scan_status_msg.content["info"]["kwargs"]["cycle_low"],
-                            cycle_high=scan_status_msg.content["info"]["kwargs"]["cycle_high"],
-                        )
-                        assert mock_set_scan_control_settings.call_args == mock.call(
-                            mode=ScanControlMode.SIMPLE,
-                            scan_duration=scan_status_msg.content["info"]["kwargs"][
-                                "scan_duration"
-                            ],
-                        )

tests/tests_file_writer/README.md

@@ -1,31 +0,0 @@
-# Getting Started with Testing using pytest
-
-BEC is using the [pytest](https://docs.pytest.org/en/8.0.x/) framework. 
-It can be install via 
-``` bash
-pip install pytest
-```
-in your *python environment*. 
-We note that pytest is part of the optional-dependencies `[dev]` of the plugin package.
-
-## Introduction
-
-Tests in this package should be stored in the `tests` directory.
-We suggest to sort tests of different submodules, i.e. `scans` or `devices` in the respective folder structure, and to folow a naming convention of `<test_module_name.py>`.
-
-To run all tests, navigate to the directory of the plugin from the command line, and run the command 
-
-``` bash
-pytest -v --random-order ./tests
-```
-Note, the python environment needs to be active.
-The additional arg `-v` allows pytest to run in verbose mode which provides more detailed information about the tests being run.
-The argument `--random-order` instructs pytest to run the tests in random order, which is the default in the CI pipelines. 
-
-## Test examples
-
-Writing tests can be quite specific for the given function. 
-We recommend writing tests as isolated as possible, i.e. try to test single functions instead of full classes.
-A very useful class to enable isolated testing is [MagicMock](https://docs.python.org/3/library/unittest.mock.html).
-In addition, we also recommend to take a look at the [How-to guides from pytest](https://docs.pytest.org/en/8.0.x/how-to/index.html).
-

tests/tests_scans/README.md

@@ -1,31 +0,0 @@
-# Getting Started with Testing using pytest
-
-BEC is using the [pytest](https://docs.pytest.org/en/8.0.x/) framework. 
-It can be install via 
-``` bash
-pip install pytest
-```
-in your *python environment*. 
-We note that pytest is part of the optional-dependencies `[dev]` of the plugin package.
-
-## Introduction
-
-Tests in this package should be stored in the `tests` directory.
-We suggest to sort tests of different submodules, i.e. `scans` or `devices` in the respective folder structure, and to folow a naming convention of `<test_module_name.py>`.
-
-To run all tests, navigate to the directory of the plugin from the command line, and run the command 
-
-``` bash
-pytest -v --random-order ./tests
-```
-Note, the python environment needs to be active.
-The additional arg `-v` allows pytest to run in verbose mode which provides more detailed information about the tests being run.
-The argument `--random-order` instructs pytest to run the tests in random order, which is the default in the CI pipelines. 
-
-## Test examples
-
-Writing tests can be quite specific for the given function. 
-We recommend writing tests as isolated as possible, i.e. try to test single functions instead of full classes.
-A very useful class to enable isolated testing is [MagicMock](https://docs.python.org/3/library/unittest.mock.html).
-In addition, we also recommend to take a look at the [How-to guides from pytest](https://docs.pytest.org/en/8.0.x/how-to/index.html).
-

tests/tests_scans/test_mono_bragg_scans.py

@@ -1,303 +0,0 @@
-# pylint: skip-file
-from unittest import mock
-
-from bec_lib.messages import DeviceInstructionMessage
-from bec_server.device_server.tests.utils import DMMock
-
-from debye_bec.scans import XASSimpleScan, XASSimpleScanWithXRD
-
-
-def test_xas_simple_scan():
-    # create a fake device manager that we can use to add devices
-    device_manager = DMMock()
-    device_manager.add_device("mo1_bragg")
-
-    request = XASSimpleScan(
-        start=0, stop=5, scan_time=1, scan_duration=10, device_manager=device_manager
-    )
-    request.metadata["RID"] = "my_test_request_id"
-    with (
-        mock.patch.object(request.stubs, "get_req_status", side_effect=[False, True]),
-        mock.patch.object(request.stubs, "_get_from_rpc", return_value=True),
-    ):
-        reference_commands = list(request.run())
-
-        for cmd in reference_commands:
-            if not cmd:
-                continue
-            if "RID" in cmd.metadata:
-                cmd.metadata["RID"] = "my_test_request_id"
-            if "rpc_id" in cmd.parameter:
-                cmd.parameter["rpc_id"] = "my_test_rpc_id"
-
-        assert reference_commands == [
-            None,
-            None,
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "monitored", "DIID": 0, "RID": "my_test_request_id"},
-                device=None,
-                action="scan_report_instruction",
-                parameter={"device_progress": ["mo1_bragg"]},
-            ),
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "monitored", "DIID": 1, "RID": "my_test_request_id"},
-                device=None,
-                action="open_scan",
-                parameter={
-                    "scan_motors": [],
-                    "readout_priority": {
-                        "monitored": [],
-                        "baseline": [],
-                        "on_request": [],
-                        "async": [],
-                    },
-                    "num_points": None,
-                    "positions": [0.0, 5.0],
-                    "scan_name": "xas_simple_scan",
-                    "scan_type": "fly",
-                },
-            ),
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "monitored", "DIID": 2, "RID": "my_test_request_id"},
-                device=None,
-                action="stage",
-                parameter={},
-            ),
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "baseline", "DIID": 3, "RID": "my_test_request_id"},
-                device=None,
-                action="baseline_reading",
-                parameter={},
-            ),
-            DeviceInstructionMessage(
-                metadata={
-                    "readout_priority": "monitored",
-                    "DIID": 4,
-                    "RID": "my_test_request_id",
-                    "response": True,
-                },
-                device="mo1_bragg",
-                action="rpc",
-                parameter={
-                    "device": "mo1_bragg",
-                    "func": "move_type.set",
-                    "rpc_id": "my_test_rpc_id",
-                    "args": ("energy",),
-                    "kwargs": {},
-                },
-            ),
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "monitored", "DIID": 5, "RID": "my_test_request_id"},
-                device="mo1_bragg",
-                action="kickoff",
-                parameter={"configure": {}, "wait_group": "kickoff"},
-            ),
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "monitored", "DIID": 6, "RID": "my_test_request_id"},
-                device="mo1_bragg",
-                action="complete",
-                parameter={},
-            ),
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "monitored", "DIID": 7, "RID": "my_test_request_id"},
-                device=None,
-                action="read",
-                parameter={"group": "primary", "wait_group": "readout_primary"},
-            ),
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "monitored", "DIID": 8, "RID": "my_test_request_id"},
-                device=None,
-                action="wait",
-                parameter={"type": "read", "group": "primary", "wait_group": "readout_primary"},
-            ),
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "monitored", "DIID": 9, "RID": "my_test_request_id"},
-                device=None,
-                action="read",
-                parameter={"group": "primary", "wait_group": "readout_primary"},
-            ),
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "monitored", "DIID": 10, "RID": "my_test_request_id"},
-                device=None,
-                action="wait",
-                parameter={"type": "read", "group": "primary", "wait_group": "readout_primary"},
-            ),
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "monitored", "DIID": 11, "RID": "my_test_request_id"},
-                device=None,
-                action="wait",
-                parameter={"type": "read", "group": "primary", "wait_group": "readout_primary"},
-            ),
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "monitored", "DIID": 12, "RID": "my_test_request_id"},
-                device=None,
-                action="complete",
-                parameter={},
-            ),
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "monitored", "DIID": 13, "RID": "my_test_request_id"},
-                device=None,
-                action="unstage",
-                parameter={},
-            ),
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "monitored", "DIID": 14, "RID": "my_test_request_id"},
-                device=None,
-                action="close_scan",
-                parameter={},
-            ),
-        ]
-
-
-def test_xas_simple_scan_with_xrd():
-    # create a fake device manager that we can use to add devices
-    device_manager = DMMock()
-    device_manager.add_device("mo1_bragg")
-
-    request = XASSimpleScanWithXRD(
-        start=0,
-        stop=5,
-        scan_time=1,
-        scan_duration=10,
-        xrd_enable_low=True,
-        num_trigger_low=1,
-        exp_time_low=1,
-        cycle_low=1,
-        xrd_enable_high=True,
-        num_trigger_high=2,
-        exp_time_high=3,
-        cycle_high=4,
-        device_manager=device_manager,
-    )
-    request.metadata["RID"] = "my_test_request_id"
-    with (
-        mock.patch.object(request.stubs, "get_req_status", side_effect=[False, True]),
-        mock.patch.object(request.stubs, "_get_from_rpc", return_value=True),
-    ):
-        reference_commands = list(request.run())
-
-        for cmd in reference_commands:
-            if not cmd:
-                continue
-            if "RID" in cmd.metadata:
-                cmd.metadata["RID"] = "my_test_request_id"
-            if "rpc_id" in cmd.parameter:
-                cmd.parameter["rpc_id"] = "my_test_rpc_id"
-
-        assert reference_commands == [
-            None,
-            None,
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "monitored", "DIID": 0, "RID": "my_test_request_id"},
-                device=None,
-                action="scan_report_instruction",
-                parameter={"device_progress": ["mo1_bragg"]},
-            ),
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "monitored", "DIID": 1, "RID": "my_test_request_id"},
-                device=None,
-                action="open_scan",
-                parameter={
-                    "scan_motors": [],
-                    "readout_priority": {
-                        "monitored": [],
-                        "baseline": [],
-                        "on_request": [],
-                        "async": [],
-                    },
-                    "num_points": None,
-                    "positions": [0.0, 5.0],
-                    "scan_name": "xas_simple_scan_with_xrd",
-                    "scan_type": "fly",
-                },
-            ),
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "monitored", "DIID": 2, "RID": "my_test_request_id"},
-                device=None,
-                action="stage",
-                parameter={},
-            ),
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "baseline", "DIID": 3, "RID": "my_test_request_id"},
-                device=None,
-                action="baseline_reading",
-                parameter={},
-            ),
-            DeviceInstructionMessage(
-                metadata={
-                    "readout_priority": "monitored",
-                    "DIID": 4,
-                    "RID": "my_test_request_id",
-                    "response": True,
-                },
-                device="mo1_bragg",
-                action="rpc",
-                parameter={
-                    "device": "mo1_bragg",
-                    "func": "move_type.set",
-                    "rpc_id": "my_test_rpc_id",
-                    "args": ("energy",),
-                    "kwargs": {},
-                },
-            ),
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "monitored", "DIID": 5, "RID": "my_test_request_id"},
-                device="mo1_bragg",
-                action="kickoff",
-                parameter={"configure": {}, "wait_group": "kickoff"},
-            ),
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "monitored", "DIID": 6, "RID": "my_test_request_id"},
-                device="mo1_bragg",
-                action="complete",
-                parameter={},
-            ),
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "monitored", "DIID": 7, "RID": "my_test_request_id"},
-                device=None,
-                action="read",
-                parameter={"group": "primary", "wait_group": "readout_primary"},
-            ),
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "monitored", "DIID": 8, "RID": "my_test_request_id"},
-                device=None,
-                action="wait",
-                parameter={"type": "read", "group": "primary", "wait_group": "readout_primary"},
-            ),
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "monitored", "DIID": 9, "RID": "my_test_request_id"},
-                device=None,
-                action="read",
-                parameter={"group": "primary", "wait_group": "readout_primary"},
-            ),
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "monitored", "DIID": 10, "RID": "my_test_request_id"},
-                device=None,
-                action="wait",
-                parameter={"type": "read", "group": "primary", "wait_group": "readout_primary"},
-            ),
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "monitored", "DIID": 11, "RID": "my_test_request_id"},
-                device=None,
-                action="wait",
-                parameter={"type": "read", "group": "primary", "wait_group": "readout_primary"},
-            ),
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "monitored", "DIID": 12, "RID": "my_test_request_id"},
-                device=None,
-                action="complete",
-                parameter={},
-            ),
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "monitored", "DIID": 13, "RID": "my_test_request_id"},
-                device=None,
-                action="unstage",
-                parameter={},
-            ),
-            DeviceInstructionMessage(
-                metadata={"readout_priority": "monitored", "DIID": 14, "RID": "my_test_request_id"},
-                device=None,
-                action="close_scan",
-                parameter={},
-            ),
-        ]

tests/tests_services/README.md

@@ -1,31 +0,0 @@
-# Getting Started with Testing using pytest
-
-BEC is using the [pytest](https://docs.pytest.org/en/8.0.x/) framework. 
-It can be install via 
-``` bash
-pip install pytest
-```
-in your *python environment*. 
-We note that pytest is part of the optional-dependencies `[dev]` of the plugin package.
-
-## Introduction
-
-Tests in this package should be stored in the `tests` directory.
-We suggest to sort tests of different submodules, i.e. `scans` or `devices` in the respective folder structure, and to folow a naming convention of `<test_module_name.py>`.
-
-To run all tests, navigate to the directory of the plugin from the command line, and run the command 
-
-``` bash
-pytest -v --random-order ./tests
-```
-Note, the python environment needs to be active.
-The additional arg `-v` allows pytest to run in verbose mode which provides more detailed information about the tests being run.
-The argument `--random-order` instructs pytest to run the tests in random order, which is the default in the CI pipelines. 
-
-## Test examples
-
-Writing tests can be quite specific for the given function. 
-We recommend writing tests as isolated as possible, i.e. try to test single functions instead of full classes.
-A very useful class to enable isolated testing is [MagicMock](https://docs.python.org/3/library/unittest.mock.html).
-In addition, we also recommend to take a look at the [How-to guides from pytest](https://docs.pytest.org/en/8.0.x/how-to/index.html).
-