wip fix bp, remove more cruft

src/aare/beamline_dispatch/beamline_dispatch.py

@@ -13,12 +13,12 @@ def get_beamline_dispatch() -> BeamlineDispatch:
         case MXBeamline.X06DA:
             from .x06da import X06daDispatch
 
-            return X06daDispatch()
+            return X06daDispatch(MXBeamline.X06DA)
         case MXBeamline.X06SA:
             from .x06sa import X06saDispatch
 
-            return X06saDispatch()
+            return X06saDispatch(MXBeamline.X06SA)  # pyright: ignore
         case MXBeamline.X10SA:
             from .x10sa import X10saDispatch
 
-            return X10saDispatch()
+            return X10saDispatch(MXBeamline.X10SA)

src/aare/beamline_dispatch/default/beamline_dispatch.py

@@ -1,8 +1,15 @@
 import os
+from abc import ABC
 
 from aarecommon.config.beamline import MXBeamline
+from aarecommon.config.logger import setup_logger
+from bec_ipython_client import BECIPythonClient
+from bec_ipython_client.signals import OperationMode
+from bec_lib.client import ServiceConfig
 
-from aare.beamline_dispatch.protocols import AuthDispatch, BeamlineDispatch, ConfigKeys
+from aare.beamline_dispatch.protocols import AuthDispatch, BeamlineDispatch, BecSetup, ConfigKeys
+
+logger = setup_logger("aareDAQ")
 
 
 class DefaultAuthDispatch(AuthDispatch):
@@ -45,13 +52,28 @@ class DefaultConfigKeys(ConfigKeys):
         return f"/sls/{self.bl_lower}/config/bec/production/{self.bl_name}_bec/{self.bl_name}_bec/device_configs/{self.bl_name}-devices.yaml"
 
 
-class DefaultDispatch(BeamlineDispatch):
+class DefaultBecSetup(BecSetup):
+    def __init__(self, beamline: MXBeamline) -> None:
+        super().__init__(beamline)
+        self._beamline = beamline
+
+    def create_client(self) -> BECIPythonClient:
+        bl = self._beamline.value.lower()
+        logger.debug(f"Initializing BECClientWorker for {bl} beamline")
+        host = f"{bl}-bec-001.psi.ch"
+        service_config = ServiceConfig(redis={"host": host, "port": 6379})
+        service_config.config["log_writer"]["base_path"] = "/tmp/logs"
+        return BECIPythonClient(config=service_config, mode=OperationMode.Procedure)
+
+
+class DefaultDispatch(BeamlineDispatch, ABC):
     """Default implementation for anything which can vary between beamlines and/or simulation.
     Should be safe and fail rather than assuming anything."""
 
     def __init__(self, beamline: MXBeamline) -> None:
         self._auth = DefaultAuthDispatch()
         self._config = DefaultConfigKeys(beamline)
+        self._bec_setup = DefaultBecSetup(beamline)
 
     @property
     def auth(self):
@@ -60,3 +82,7 @@ class DefaultDispatch(BeamlineDispatch):
     @property
     def config(self):
         return self._config
+
+    @property
+    def bec_setup(self):
+        return self._bec_setup

src/aare/beamline_dispatch/protocols.py

@@ -1,6 +1,9 @@
 from abc import ABC, abstractmethod
 from typing import Any
 
+from aarecommon.models.models import MXBeamline
+from bec_ipython_client import BECIPythonClient
+
 
 class AuthDispatch(ABC):
     @abstractmethod
@@ -28,6 +31,11 @@ class BecMacros(ABC):
     def mono_pitch_scan(plot=True): ...
 
 
+class BecSetup(ABC):
+    def __init__(self, beamline: MXBeamline) -> None: ...
+    def create_client(self) -> BECIPythonClient: ...
+
+
 class ConfigKeys(ABC):
     @property
     @abstractmethod
@@ -51,4 +59,7 @@ class BeamlineDispatch(ABC):
     def bec_macros(self) -> BecMacros: ...
     @property
     @abstractmethod
+    def bec_setup(self) -> BecSetup: ...
+    @property
+    @abstractmethod
     def config(self) -> ConfigKeys: ...

src/aare/beamline_dispatch/simulated/beamline_dispatch.py

@@ -1,9 +1,11 @@
 from typing import Any
+from unittest.mock import MagicMock
 
 from aarecommon.config.beamline import MXBeamline
+from bec_ipython_client import BECIPythonClient
 
 from aare.beamline_dispatch.default.beamline_dispatch import DefaultConfigKeys, DefaultDispatch
-from aare.beamline_dispatch.protocols import AuthDispatch, BecMacros
+from aare.beamline_dispatch.protocols import AuthDispatch, BecMacros, BecSetup
 
 
 class SimulatedAuthDispatch(AuthDispatch):
@@ -25,13 +27,22 @@ class SimulatedBecMacros(BecMacros):
     @staticmethod
     def bl_energy(energy_ev, move_gap=True, mono_scan=True, plot=True): ...
     @staticmethod
-    def get_current_energy(): ...
+    def get_current_energy() -> float: ...
     @staticmethod
     def mono_pitch_scan(plot=True): ...
 
 
+class SimulatedBecSetup(BecSetup):
+    def __init__(self, beamline: MXBeamline) -> None:
+        super().__init__(beamline)
+        self._beamline = beamline
+
+    def create_client(self) -> BECIPythonClient:
+        return MagicMock(spec=BECIPythonClient)
+
+
 class SimulatedDispatch(DefaultDispatch):
-    def __init__(self) -> None:
+    def __init__(self) -> None:  # pyright: ignore[reportMissingSuperCall]
         self._auth = SimulatedAuthDispatch()
         self._bec_macros = SimulatedBecMacros()
         self._config = DefaultConfigKeys(MXBeamline.SIMULATED)

src/aare/beamline_dispatch/x06da/beamline_dispatch.py

@@ -1,49 +1,47 @@
 from typing import Any
 
 from aarecommon.config.beamline import MXBeamline
+from pxiii_bec.macros.build_devices import (  # pyright: ignore[reportMissingImports, reportMissingModuleSource]
+    save_and_reload,
+    save_current_position,
+)
+from pxiii_bec.macros.init_beamline import (  # pyright: ignore[reportMissingImports, reportMissingModuleSource]
+    init_beamline_environment,
+)
+from pxiii_bec.macros.pxiii_energy import (  # pyright: ignore[reportMissingImports, reportMissingModuleSource]
+    bl_energy,
+    get_current_energy,
+    mono_pitch_scan,
+)
 
 from aare.beamline_dispatch.default.beamline_dispatch import DefaultDispatch
 from aare.beamline_dispatch.protocols import BecMacros
 
 
 class X06daBecMacros(BecMacros):
-    def __init__(self) -> None:
-        from pxiii_bec.macros.build_devices import (  # pyright: ignore[reportMissingImports, reportMissingModuleSource]
-            save_and_reload,
-            save_current_position,
-        )
-        from pxiii_bec.macros.init_beamline import (  # pyright: ignore[reportMissingImports, reportMissingModuleSource]
-            init_beamline_environment,
-        )
-        from pxiii_bec.macros.pxiii_energy import (  # pyright: ignore[reportMissingImports, reportMissingModuleSource]
-            bl_energy,
-            get_current_energy,
-            mono_pitch_scan,
-        )
-
-        self.save_and_reload = save_and_reload
-        self.save_current_position = save_current_position
-        self.init_beamline_environment = init_beamline_environment
-
-        def _bl_energy(energy_ev, move_gap=True, mono_scan=True, plot=True):
-            return bl_energy(energy_ev=energy_ev, plot=plot)
-
-        self.bl_energy = _bl_energy
-        self.get_current_energy = get_current_energy
-        self.mono_pitch_scan = mono_pitch_scan
+    @staticmethod
+    def save_and_reload() -> tuple[Any, Any]:
+        return save_and_reload()
 
     @staticmethod
-    def save_and_reload() -> tuple[Any, Any]: ...
+    def save_current_position(device, position, axis=None, force=False, max_delta=0.5) -> None:
+        return save_current_position
+
     @staticmethod
-    def save_current_position(device, position, axis=None, force=False, max_delta=0.5) -> None: ...
+    def init_beamline_environment() -> tuple[Any, Any]:
+        return init_beamline_environment()
+
     @staticmethod
-    def init_beamline_environment() -> tuple[Any, Any]: ...
+    def bl_energy(energy_ev, move_gap=True, mono_scan=True, plot=True):
+        return bl_energy(energy_ev=energy_ev, plot=plot)
+
     @staticmethod
-    def bl_energy(energy_ev, move_gap=True, mono_scan=True, plot=True): ...
+    def get_current_energy() -> float:
+        return get_current_energy()
+
     @staticmethod
-    def get_current_energy(): ...
-    @staticmethod
-    def mono_pitch_scan(plot=True): ...
+    def mono_pitch_scan(plot=True):
+        return mono_pitch_scan(plot)
 
 
 class X06daDispatch(DefaultDispatch):

src/aare/beamline_dispatch/x06sa/beamline_dispatch.py

@@ -1,4 +1,6 @@
+from abc import ABC
+
 from aare.beamline_dispatch.default.beamline_dispatch import DefaultDispatch
 
 
-class X06saDispatch(DefaultDispatch): ...
+class X06saDispatch(DefaultDispatch, ABC): ...

src/aare/beamline_dispatch/x10sa/beamline_dispatch.py

@@ -1,43 +1,47 @@
 from typing import Any
 
 from aarecommon.config.beamline import MXBeamline
+from pxii_bec.macros.build_devices import (  # pyright: ignore[reportMissingImports, reportMissingModuleSource]
+    save_and_reload,
+    save_current_position,
+)
+from pxii_bec.macros.init_beamline import (  # pyright: ignore[reportMissingImports, reportMissingModuleSource]
+    init_beamline_environment,
+)
+from pxii_bec.macros.pxii_energy import (  # pyright: ignore[reportMissingImports, reportMissingModuleSource]
+    bl_energy,
+    get_current_energy,
+    mono_pitch_scan,
+)
 
 from aare.beamline_dispatch.default.beamline_dispatch import DefaultDispatch
 from aare.beamline_dispatch.protocols import BecMacros
 
 
 class X10SaBecMacros(BecMacros):
-    def __init__(self) -> None:
-        from pxii_bec.macros.build_devices import (  # type: ignore
-            save_and_reload,
-            save_current_position,
-        )
-        from pxii_bec.macros.init_beamline import init_beamline_environment  # type: ignore
-        from pxii_bec.macros.pxii_energy import (  # type: ignore
-            bl_energy,
-            get_current_energy,
-            mono_pitch_scan,
-        )
-
-        self.save_and_reload = save_and_reload
-        self.save_current_position = save_current_position
-        self.init_beamline_environment = init_beamline_environment
-        self.bl_energy = bl_energy
-        self.get_current_energy = get_current_energy
-        self.mono_pitch_scan = mono_pitch_scan
+    @staticmethod
+    def save_and_reload() -> tuple[Any, Any]:
+        return save_and_reload()
 
     @staticmethod
-    def save_and_reload() -> tuple[Any, Any]: ...
+    def save_current_position(device, position, axis=None, force=False, max_delta=0.5) -> None:
+        return save_current_position
+
     @staticmethod
-    def save_current_position(device, position, axis=None, force=False, max_delta=0.5) -> None: ...
+    def init_beamline_environment() -> tuple[Any, Any]:
+        return init_beamline_environment()
+
     @staticmethod
-    def init_beamline_environment() -> tuple[Any, Any]: ...
+    def bl_energy(energy_ev, move_gap=True, mono_scan=True, plot=True):
+        return bl_energy(energy_ev=energy_ev, move_gap=move_gap, mono_scan=mono_scan, plot=plot)
+
     @staticmethod
-    def bl_energy(energy_ev, move_gap=True, mono_scan=True, plot=True): ...
+    def get_current_energy() -> float:
+        return get_current_energy()
+
     @staticmethod
-    def get_current_energy(): ...
-    @staticmethod
-    def mono_pitch_scan(plot=True): ...
+    def mono_pitch_scan(plot=True):
+        return mono_pitch_scan(plot)
 
 
 class X10saDispatch(DefaultDispatch):

src/aare/devices/bec_worker.py

@@ -2,17 +2,12 @@ import sys
 import time
 from enum import Enum, StrEnum
 
-from aarecommon.config.beamline import cfg_get, mx_beamline
 from aarecommon.config.logger import setup_logger
 from aarecommon.config.logger_events import log_timing
 from aarecommon.errors.exception_handler import BECCommunicationError, BECException
 from aarecommon.models.beamline import MXBeamline
-from bec_ipython_client import BECIPythonClient
-from bec_ipython_client.signals import OperationMode
 from bec_lib.device import RPCError, ScanRequestError
 from bec_lib.devicemanager import DeviceContainer
-from bec_lib.procedures.helper import FrontendProcedureHelper
-from bec_lib.service_config import ServiceConfig
 
 from aare.beamline_dispatch.beamline_dispatch import get_beamline_dispatch
 
@@ -53,9 +48,9 @@ class BeamlineState(str, Enum):
 class BecDevices(StrEnum):
     DET_Z = "det_z"
     FL_BRIGHT = "fl_bright"
-    BL_BRIGHT = "bl_bright"
-    BL_POS = "bl_pos"
-    SCAM_ZOOM = "scam_zoom"
+    BACKLIGHT_BRIGHT = "bl_bright"
+    BACKLIGHT_POS = "bl_pos"
+    SAMCAM_ZOOM = "scam_zoom"
     CRYO_POS = "cryo_pos"
     CRYO_TEMP = "cryo_temp"
     CRYO_X = "cryo_x"
@@ -71,44 +66,28 @@ class MissingDevicesError(BECException): ...
 class BECClientWorker:
     def __init__(self, beamline: MXBeamline, name: str = "default"):
         self.dispatch = get_beamline_dispatch()
-
-        if beamline is MXBeamline.SIMULATED:
-            self.simulated = True
-
-        else:
-            self.simulated = False
-            logger.debug(
-                f"Initializing BECClientWorker for {self.dispatch.config.bl_lower} beamline"
+        self.client = self.dispatch.bec_setup.create_client()
+        self.client.start()
+        self.dev: DeviceContainer = self.client.device_manager.devices
+        missing_devices = {str(d.value) for d in list(BecDevices) if d not in self.dev}
+        if missing_devices:
+            raise MissingDevicesError(
+                f"{missing_devices} were not found in the BEC session. Available: {list(self.dev.keys())}. If no devices are available, load a BEC device config."
             )
-            host = cfg_get(
-                "daq.hardware.bec_url", f"{self.dispatch.config.bl_lower}-bec-001.psi.ch"
-            )
-            service_config = ServiceConfig(redis={"host": host, "port": 6379})
-            service_config.config["log_writer"]["base_path"] = "/tmp/logs"
-            self.client = BECIPythonClient(config=service_config, mode=OperationMode.Procedure)
-            self.client.start()
-            self.dev: DeviceContainer = self.client.device_manager.devices
-            missing_devices = {str(d) for d in list(BecDevices) if d not in self.dev}
-            if missing_devices:
-                raise MissingDevicesError(
-                    f"{missing_devices} were not found in the BEC session. Available: {list(self.dev.keys())}. If no devices are available, load a BEC device config."
-                )
-            self.scans = self.client.scans
-            self.macros = self.dispatch.bec_macros
-            self.helper = FrontendProcedureHelper(self.client.connector)
-            self._set_scilog_tags()
-            try:
-                self._init_beamline_environment()
-            except Exception:
-                logger.exception("Error initialising BEC devices")
-                sys.exit(1)
-        logger.debug(f"simulated is {self.simulated}")
+        self.scans = self.client.scans
+        self.macros = self.dispatch.bec_macros
+        self._set_scilog_tags()
+        try:
+            self._init_beamline_environment()
+        except Exception:
+            logger.exception("Error initialising BEC devices")
+            sys.exit(1)
 
     def _init_beamline_environment(self):
         self.position_devices, self.planner = self.macros.init_beamline_environment()
         self._backlight_brightness = self.position_devices["bl_bright"]
         self._frontlight_brightness = self.position_devices["fl_bright"]
-        self._zoom = self.dev[BecDevices.SCAM_ZOOM]
+        self._zoom = self.dev[BecDevices.SAMCAM_ZOOM]
         self._ring_current = self.dev.sls_current
 
     def _bec_error(
@@ -218,11 +197,7 @@ class BECClientWorker:
     @log_timing(logger, "BEC move_to")
     def move_to(self, state: BeamlineState):
         start = time.perf_counter()
-        logger.debug(f"simulated is {self.simulated}")
         logger.info(f"BEC move_to requested: {state.value}")
-        if self.simulated:
-            logger.debug(f"Simulating move to {state.value}")
-            return True
         try:
             self.planner.move_to(state)
             if self.planner.is_state(state):
@@ -239,40 +214,17 @@ class BECClientWorker:
             raise self._bec_error(e, operation=f"planner.move_to:{state.value}") from e
 
     def is_state(self, state: BeamlineState):
-        if self.simulated:
-            logger.debug(f"Simulating check_beamline_state: {state.value}")
-            return True
         return self.planner.is_state(state)
 
     def current_state(self):
-        if self.simulated:
-            logger.debug("Simulating check_beamline_state")
-            return BeamlineState.MAINTENANCE
         return self.planner.current_state()
 
     def list_position_devices(self):
         """List the position devices available for the BEC worker"""
-        if self.simulated or self.position_devices is None:
+        if self.position_devices is None:
             return []
         return list(self.position_devices.keys())
 
-    def show_all_devices(self):
-        return self.dev.show_all
-
-    def list_all_user_macros(self) -> list[str] | None:
-        if self.simulated:
-            logger.debug("Simulating list_all_user_macros")
-            return []
-        try:
-            self._list_all_macros()
-            raw_macros = [name for name, _ in self.client.macros._update_handler.macros.items()]
-            if raw_macros is None:
-                logger.warning("BEC returned no user macros; treating as empty list")
-                return []
-            return [str(macro) for macro in raw_macros]
-        except Exception as e:
-            raise self._bec_error(e, operation="list_all_user_macros") from e
-
     def reinitialise_planner_and_position_devices(self, method: str = "auto"):
         """
         Reinitialise BEC position devices and planner.
@@ -286,9 +238,6 @@ class BECClientWorker:
         Returns:
             List of position device names after reinitialisation.
         """
-        if self.simulated:
-            logger.debug(f"Simulating reinitialise_planner_and_position_devices(method={method})")
-            return []
         try:
             self.client.config.update_session_with_file(self.dispatch.config.device_yml_path)
             self._init_beamline_environment()
@@ -342,7 +291,7 @@ class BECClientWorker:
             logger.warning(
                 f"Energy change may have failed, current energy: {self.check_current_energy()} eV"
             )
-        if beamline is MXBeamline.X10SA:
+        if self.dispatch.config.bl_upper == MXBeamline.X10SA:
             additonal_text = [
                 f"New dcm_bragg position: {self.dev.dcm_bragg.position:4g} mrad",
                 f"New dcm_pitch position: {self.dev.dcm_pitch.position:4g} ",
@@ -415,15 +364,12 @@ class BECClientWorker:
     @backlight_brightness.setter
     def backlight_brightness(self, value: int | str):
         """Set the backlight brightness to the specified value"""
-        if self.simulated:
-            return
         try:
             self._backlight_brightness.move(value)
         except Exception as e:
             raise self._bec_error(
                 e, operation=f"backlight_brightness:{value}", tags=["backlight"]
             ) from e
-            raise
 
     def get_backlight_pos(self) -> BrightnessEnum:
         """Returns the current backlight brightness position"""
@@ -431,8 +377,6 @@ class BECClientWorker:
 
     def backlight_toggle(self):
         """Turn the backlight on or off"""
-        if self.simulated:
-            return
         try:
             current = self.get_backlight_pos()
             if current is BrightnessEnum.ON:
@@ -478,132 +422,3 @@ class BECClientWorker:
         except (RPCError, ScanRequestError) as e:
             logger.warning(f"Ring current BEC device is not available: {e}")
             return 0
-
-
-if __name__ == "__main__":
-    import time
-
-    print(time.ctime(), " starting BEC Client")
-    beamline = mx_beamline()
-    try:
-        client = BECClientWorker(beamline)
-    except Exception:
-        logger.exception("Failed to start BEC client")
-        try:
-            client.shutdown_client()
-        except Exception:
-            logger.exception("Failed to shut the BEC client down cleanly")
-            sys.exit(1)
-    # print(client.get_det_cov(actual=True))
-    # print(client.is_state(BeamlineState.ROBOT_SAMPLE_EXCHANGE))
-    try:
-        print("startting backlight brightness test")
-        print("initial value")
-        # print(client.client.show_last_alarm())
-        # print(client._raise_bec_error(exc=Exception("test"), operation="test", tags=["test"]))
-        print(client.ring_current)
-        for i in range(10):
-            print(client._ring_current.get())
-            time.sleep(1)
-        # print('setting to 5')
-        # client.backlight_brightness = 5
-        # print(client.get_backlight_pos())
-        # print(client.backlight_brightness)
-        # print('setting to 1.2')
-        # client.backlight_brightness = 1.2
-        # print(client.get_backlight_pos())
-        # print(client.backlight_brightness)
-        # print('setting to ON')
-        # client.backlight_brightness = BrightnessEnum.ON
-        # print(client.get_backlight_pos())
-        # print(client.backlight_brightness)
-        # print('setting to 1.0')
-        # client.backlight_brightness = 1.0
-        # print(client.get_backlight_pos())
-        # print(client.backlight_brightness)
-        # print('setting to 0.0')
-        # client.backlight_brightness = 0.0
-        # print(client.get_backlight_pos())
-        # print(client.backlight_brightness)
-        # print('setting to OFF')
-        # client.backlight_brightness = BrightnessEnum.OFF
-        # client.backlight_toggle()
-        # print(client.get_backlight_pos())
-        # print(client.backlight_brightness)
-        # client.scilog_msg("Testing scilog messages with color = yellow and italic", italic=True,
-        #                  color="green", warning=False)
-    except Exception:
-        client.shutdown_client()
-        logger.exception("BEC client smoke test failed")
-
-    # try:
-    #     det_value = 980
-    #     print(f"moving detector to vale:{det_value}")
-    #     start = time.perf_counter()
-    #     # print(client.get_det_z())
-    #     # client.det_z(value=det_value, timeout=10)
-    #     # print(time.perf_counter() - start)
-    #     # print(client.get_det_z)
-    #     # det_value = 985
-    #     # print(f"moving detector to vale:{det_value}")
-    #     status = client.det_z(value=det_value)
-    #     status.wait(timeout=5)
-    #     print(status)
-    #     print(client.get_det_z())
-    #     #client.mono_pitch_scan_runner()
-    #     #client.change_energy(12000)
-    #
-    # except Exception as e:
-    #     print(f"Error: {e}")
-    # # client.planner.current_state()
-    # client.planner.move_to(BeamlineState.SAMPLE_ALIGNMENT)
-    # time.sleep(2.0)
-    # client.planner.current_state()
-    # client.move_to(BeamlineState.SAMPLE_ALIGNMENT)
-    # time.sleep(2.0)
-    # client.planner.current_state()
-
-    # client.load_user_macros()
-    # client.macros.mono_pitch_scan(False)
-    # status = client.run_macro("planner.current_state", queue="default")
-    # client.planner.move_to('manual_sample_exchange')
-    # print(status)
-    # status.wait()
-    # print(status)
-    # try:
-    # status=client.proc.run_macro("mono_pitch_scan", queue="test")
-    # planner.move_to('manual_sample_exchange')
-
-    # print(status)
-    # status.wait()
-    # status.cancel()
-    # print(status)
-    # client.client.macros.mono_pictch_scan(False)
-    # try:
-    #     # a=client.a2e_runner(160, "iln")
-    #     # print(a)
-    #     # print(convert_from_energy(12))
-    #     # energy = get_current_energy()
-    #     # pos = get_dcm_motors_positions(energy)
-    #     # print(energy, pos)
-    #     print(bs_z_policy(15.0))
-    #     #client.scans.umv(client.dev.xeye_x, 0, relative=False)
-    #     #client.mono_pitch_scan_runner()
-    #     #client.mono_pitch_scan_runner()
-    #     #b=client.run_macro_blocked("a2e", 160, "iln", queue="test")
-    #     #b = client.run_macro_blocked("mono_pitch_scan", False, queue="default")
-    #
-    #     #status = client.mono_pitch_scan_runner
-    #     #print(status)
-    #
-    #     #client.rse2sa()
-    #     #time.sleep(10.0)
-    #     #time.sleep(10.0)
-    #     #print(status)
-    #     #client.common2rse()
-    #     #print(status)
-    # except GuardViolation as e:
-    #     print(f"GuardViolation: {e}")
-    # except RuntimeError as e:
-    #     print(f"RuntimeError: {e}")
-    client.shutdown_client()

src/aare/devices/mx_lib.py

@@ -1,289 +0,0 @@
-import re
-import time
-from collections.abc import Callable
-from typing import Any
-
-from aarecommon.config.logger import setup_logger
-from epics import PV, Motor, poll
-
-logger = setup_logger("aareDAQ")
-
-
-def wait_for_movement_to_finish(*motors):
-    """
-    Wait for all {motors} passed in argument to finish movement.
-
-    Requirements: epics.Motor -- each {motor} is an instance of this
-                  class.
-
-    The overall timeout is the maximum of (.RBV - .VAL)/.VELO for each
-    concerned {motor}.
-
-    Returns: nothing
-    """
-    poll(0.3)
-    longest = 0.0
-    for motor in motors:
-        time_to_target = motor.readback / motor.slew_speed
-        longest = max(longest, time_to_target)
-
-    timeout = time.time() + 1.5 * longest
-    done = False
-    while not done and time.time() < timeout:
-        done = all(m.done_moving for m in motors)
-
-    if time.time() > timeout:
-        print("TIMEOUT waiting for motors to be done moving; current motor positions:")
-        for m in motors:
-            print(f"...{m._prefix[:-1]} ({m.units})  VAL={m.drive:.1f}; RBV={m.readback:.1f}")
-        raise TimeoutError("timeout waiting for movement to finish.")
-
-
-class ValueWaitTimeout(Exception):
-    pass
-
-
-def pv_wait(
-    pv: PV | Motor,
-    value: Any,
-    *,
-    timeout: float = 60.0,
-    polling: float = 0.1,
-    tolerance: float | None = None,
-    verbose: bool = False,
-):
-    """wait until an epics.PV reaches a value
-    pv: epics.PV | epics.Motor
-        the PV on which you want to wait on
-    value/target: any
-        this value depends on the PV type: str, enum, double, ...
-    :timeout: float default = 60.0
-        timeout in seconds
-    :polling: float default = 0.1
-        polling interval in seconds
-    tolerance: float or None
-        provide a tolerance to accept when comparing values, currently None by default, however certain PVs and motors
-        have inbuilt tolerances that can be used instead
-    verbose: bool default = False
-        NotImplemented
-    """
-    if isinstance(pv, Motor):
-
-        def checker(m, target=None, tolerance=None):
-            if tolerance is None:
-                tolerance = m.get("RDBD")
-                if tolerance is None:
-                    print(f"WARNING: motor {m._prefix[:-1]} has no RDBD, using 0.001")
-                    tolerance = 0.001  # FIXME
-            if target is None:
-                target = m.drive
-
-            diff = abs(target - m.readback)
-            return m.done_moving and diff < tolerance
-
-        wait_motor_position(
-            pv, checker, target=value, tolerance=tolerance, timeout=timeout, polling=polling
-        )
-
-    elif is_epics_type(pv, "double") or is_epics_type(pv, "time_double"):
-        wait_float_condition(pv, value, tolerance=tolerance, timeout=timeout, polling=polling)
-
-    elif is_epics_type(pv, "enum") or is_epics_type(pv, "time_enum"):
-        wait_enum_condition(pv, value, timeout=timeout, polling=polling)
-
-    elif is_epics_type(pv, "string") or is_epics_type(pv, "time_string"):
-        wait_string_condition(pv, value, timeout=timeout, polling=polling)
-    else:
-        raise ValueError(f"pv_wait() don't know how to handle {pv.pvname} of type {pv.type}")
-
-
-def is_epics_type(pv: PV, pv_type: str) -> bool:
-    """Check to see if a PV is of a certain type such as double, enum, string, ..."""
-    if isinstance(pv_type, type):
-        pv_type = pv_type.__name__
-
-    return pv_type == pv.type
-
-
-def wait_string_condition(
-    pv: PV, target: str | re.Pattern, *, timeout: float = 60.0, polling: float = 0.1
-):
-    """wait until an epics.PV of type string reaches target
-    :pv: epics.PV
-        PV should be of type string
-    :target: str or re.Pattern
-    :timeout: float default = 60.0
-        timeout in seconds
-    :polling: float default = 0.1
-        polling interval in seconds
-    """
-
-    if not (isinstance(pv, PV) and "string" not in pv.type):
-        raise AttributeError("argument 'pv' must be an epics.PV of type string")
-
-    if not isinstance(target, re.Pattern):
-        target = re.compile(f"^{target}$", re.IGNORECASE)
-
-    tout = time.time() + timeout
-
-    while time.time() < tout:
-        if target.match(pv.char_value):
-            return
-        poll(polling)
-
-    if time.time() > tout:
-        raise TimeoutError(
-            f"timeout waiting for string {pv.pvname} == {target}; actual value == {pv.char_value}"
-        )
-
-
-def wait_float_condition(
-    pv: PV,
-    value: float,
-    *,
-    timeout: float = 60.0,
-    polling: float = 0.1,
-    tolerance: float | None = None,
-):
-    """wait until an epics.PV of type double reaches value
-    pv: epics.PV
-        the PV enum on which you want to wait on
-
-    value: float
-        the target value
-
-    polling: float
-        how often pv is checked during wait loop
-
-    tolerance: float or None
-        the tolerance to accept when comparing values, if None (default) we
-        try to figure an appropriate value
-
-
-    timeout: float default = 60.0
-        a timeout in seconds
-
-    return: nothing
-
-    raises: TimeoutError if a timeout occurs
-    """
-
-    try:
-        value = float(value)
-    except ValueError:
-        raise AttributeError("argument 'value' must be a number")
-
-    if tolerance is None:
-        # If pv.precision is missing/None, fall back to a default
-        precision = getattr(pv, "precision", None)
-        if precision is None:
-            tolerance = 1e-6
-        else:
-            tolerance = pow(10, -(precision - 1))
-
-    tout = time.time() + timeout
-
-    while time.time() < tout:
-        if abs(pv.value - value) < tolerance:
-            return
-        poll(polling)
-
-    if time.time() > tout:
-        raise TimeoutError(f"timeout waiting for enum {pv.pvname} == {value}")
-
-
-def wait_motor_position(
-    motor: Motor, tester: Callable, *, timeout: float = 50.0, polling: float = 0.1, **kwargs
-):
-    """wait until an epics.Motor reaches a condition
-    motor: epics.Motor
-        the motor on which you want to wait on
-
-    tester: Callable
-        a Callable which is passed motor and **kwargs
-           => Callable(motor, **kwargs)
-
-    timeout: double
-        a timeout in seconds
-
-    **kwargs: extra arguments to tester
-
-    return: nothing
-
-    raises: TimeoutError if a timeout occurs
-    """
-    if not callable(tester):
-        raise TypeError("argument 'tester' must be a function")
-
-    try:
-        move_time = abs(motor.drive - motor.readback) / motor.speed
-    except Exception:
-        logger.debug("Could not compute the motor move time; using the 1 s default", exc_info=True)
-        move_time = 1.0  # in case of unusual motor record
-
-    tout = move_time + time.time() + timeout
-
-    while time.time() < tout:
-        if tester(motor, **kwargs):
-            return
-        poll(polling)
-
-    if time.time() > tout:
-        raise TimeoutError(
-            f"timeout waiting for a condition on {motor} {motor.drive} != {motor.readback}"
-        )
-
-
-def wait_enum_condition(
-    pv: PV, value: str | int | re.Pattern, *, timeout: float = 60.0, polling=0.1
-):
-    """wait until an epics.PV enum reaches value
-    pv: epics.PV
-        the PV enum on which you want to wait on
-
-    value: int or str
-        if int check against index of current value
-
-    timeout: double
-        a timeout in seconds
-
-    return: nothing
-
-    raises: TimeoutError if a timeout occurs
-    """
-    if not (isinstance(pv, PV) and pv.type.lower().endswith("enum")):
-        raise AttributeError("argument 'pv' must be an epics.PV of type enum")
-
-    if not (isinstance(value, (str, int, re.Pattern))):
-        raise TypeError("argument 'value' must be either an int, str, or re.Pattern")
-
-    if type(value) is int:
-        tester = lambda pv: value == pv.get()
-    elif type(value) is str:
-        tester = lambda pv: str(value) == pv.get(as_string=True).lower()
-        value = str(value).lower()  # it's already a str :-/
-    elif isinstance(value, re.Pattern):
-        tester = lambda pv: value.match(pv.get(as_string=True))
-    else:
-        raise AttributeError("argument 'value' must be either an int, str, or re.Pattern")
-
-    tout = time.time() + timeout
-
-    while time.time() < tout:
-        if tester(pv):
-            return
-        poll(polling)
-
-    if time.time() > tout:
-        raise TimeoutError(
-            f"timeout waiting for enum {pv.pvname} == {value},"
-            f"current value is {pv.get(as_string=True)}"
-        )
-
-
-def clean_filename(filename: str) -> str:
-    cleaned = re.sub(r"[^A-Za-z0-9._-]", "_", filename.strip())
-    cleaned = cleaned.strip("._-")
-    if not cleaned:
-        raise ValueError("Filename is empty after sanitization.")
-    return cleaned

tests/unit/devices/test_mx_lib.py

@@ -1,92 +0,0 @@
-from unittest.mock import MagicMock, patch
-
-import pytest
-from epics import PV, Motor
-
-from aare.devices.mx_lib import clean_filename, is_epics_type, pv_wait, wait_for_movement_to_finish
-
-
-def test_clean_filename():
-    assert clean_filename("my file!.txt") == "my_file_.txt"
-    assert clean_filename("  /path/to/somewhere/file.txt  ") == "path_to_somewhere_file.txt"
-    assert clean_filename("abcABC123") == "abcABC123"
-    assert clean_filename("file._-") == "file"
-
-    with pytest.raises(ValueError):
-        clean_filename("!!!")
-
-
-def test_is_epics_type():
-    mock_pv = MagicMock(spec=PV)
-    mock_pv.type = "double"
-    assert is_epics_type(mock_pv, "double") is True
-    assert is_epics_type(mock_pv, "enum") is False
-
-    class FakeType:
-        pass
-
-    assert is_epics_type(mock_pv, FakeType) is False
-
-
-@patch("aare.devices.mx_lib.poll")
-def test_wait_for_movement_to_finish(mock_poll):
-    mock_motor = MagicMock(spec=Motor)
-    mock_motor.readback = 10.0
-    mock_motor.slew_speed = 1.0
-    mock_motor.done_moving = True
-
-    wait_for_movement_to_finish(mock_motor)
-    assert mock_poll.called
-
-
-@patch("aare.devices.mx_lib.poll")
-def test_wait_for_movement_to_finish_timeout(mock_poll):
-    mock_motor = MagicMock(spec=Motor)
-    mock_motor.readback = 10.0
-    mock_motor.slew_speed = 1.0
-    mock_motor.done_moving = False
-    mock_motor.drive = 11.0
-    mock_motor.units = "mm"
-    mock_motor._prefix = "MOT1:"
-
-    with patch("time.time", side_effect=[0, 0, 100, 101]), pytest.raises(TimeoutError):
-        wait_for_movement_to_finish(mock_motor)
-
-
-@patch("aare.devices.mx_lib.wait_motor_position")
-def test_pv_wait_motor(mock_wait_motor):
-    mock_motor = MagicMock(spec=Motor)
-    pv_wait(mock_motor, 10.0)
-    mock_wait_motor.assert_called_once()
-
-
-@patch("aare.devices.mx_lib.wait_float_condition")
-def test_pv_wait_double(mock_wait_float):
-    mock_pv = MagicMock(spec=PV)
-    mock_pv.type = "double"
-    pv_wait(mock_pv, 10.0)
-    mock_wait_float.assert_called_once()
-
-
-@patch("aare.devices.mx_lib.wait_enum_condition")
-def test_pv_wait_enum(mock_wait_enum):
-    mock_pv = MagicMock(spec=PV)
-    mock_pv.type = "enum"
-    pv_wait(mock_pv, "READY")
-    mock_wait_enum.assert_called_once()
-
-
-@patch("aare.devices.mx_lib.wait_string_condition")
-def test_pv_wait_string(mock_wait_string):
-    mock_pv = MagicMock(spec=PV)
-    mock_pv.type = "string"
-    pv_wait(mock_pv, "hello")
-    mock_wait_string.assert_called_once()
-
-
-def test_pv_wait_unknown_type():
-    mock_pv = MagicMock(spec=PV)
-    mock_pv.type = "unknown"
-    mock_pv.pvname = "TEST:PV"
-    with pytest.raises(ValueError):
-        pv_wait(mock_pv, 1.0)