diff --git a/src/aare/beamline_dispatch/beamline_dispatch.py b/src/aare/beamline_dispatch/beamline_dispatch.py index 80e1bb47..879ff005 100644 --- a/src/aare/beamline_dispatch/beamline_dispatch.py +++ b/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) diff --git a/src/aare/beamline_dispatch/default/beamline_dispatch.py b/src/aare/beamline_dispatch/default/beamline_dispatch.py index 500dd6cf..a2830190 100644 --- a/src/aare/beamline_dispatch/default/beamline_dispatch.py +++ b/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 diff --git a/src/aare/beamline_dispatch/protocols.py b/src/aare/beamline_dispatch/protocols.py index e562c9d4..61f2ebcc 100644 --- a/src/aare/beamline_dispatch/protocols.py +++ b/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: ... diff --git a/src/aare/beamline_dispatch/simulated/beamline_dispatch.py b/src/aare/beamline_dispatch/simulated/beamline_dispatch.py index b023bf79..b2521b1e 100644 --- a/src/aare/beamline_dispatch/simulated/beamline_dispatch.py +++ b/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) diff --git a/src/aare/beamline_dispatch/x06da/beamline_dispatch.py b/src/aare/beamline_dispatch/x06da/beamline_dispatch.py index 1ef2578a..f92d4466 100644 --- a/src/aare/beamline_dispatch/x06da/beamline_dispatch.py +++ b/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): diff --git a/src/aare/beamline_dispatch/x06sa/beamline_dispatch.py b/src/aare/beamline_dispatch/x06sa/beamline_dispatch.py index fcae2cec..9224cbd0 100644 --- a/src/aare/beamline_dispatch/x06sa/beamline_dispatch.py +++ b/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): ... diff --git a/src/aare/beamline_dispatch/x10sa/beamline_dispatch.py b/src/aare/beamline_dispatch/x10sa/beamline_dispatch.py index 6ef814dd..b12a9344 100644 --- a/src/aare/beamline_dispatch/x10sa/beamline_dispatch.py +++ b/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): diff --git a/src/aare/devices/bec_worker.py b/src/aare/devices/bec_worker.py index 2b3d3384..25c4c1fa 100644 --- a/src/aare/devices/bec_worker.py +++ b/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() diff --git a/src/aare/devices/mx_lib.py b/src/aare/devices/mx_lib.py deleted file mode 100644 index c0807eda..00000000 --- a/src/aare/devices/mx_lib.py +++ /dev/null @@ -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 diff --git a/tests/unit/devices/test_mx_lib.py b/tests/unit/devices/test_mx_lib.py deleted file mode 100644 index 4baeb224..00000000 --- a/tests/unit/devices/test_mx_lib.py +++ /dev/null @@ -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)