diff --git a/src/aare/devices/bec_worker.py b/src/aare/devices/bec_worker.py index 0b584204..4fda1678 100644 --- a/src/aare/devices/bec_worker.py +++ b/src/aare/devices/bec_worker.py @@ -1,4 +1,5 @@ from enum import Enum +from typing import List, Optional from bec_ipython_client import BECIPythonClient from bec_ipython_client.signals import OperationMode @@ -25,6 +26,20 @@ logger = setup_logger("aareDAQ") # helper.get.running_procedures() # helper.request.abort_queue() +class DetectorCoverEnum(str, Enum): + """Enum for the detector cover position + position devices can take string or number to move + Currently dictated string in DAQ for consitency""" + OPEN = 'open' #2 + CLOSED = 'closed' #1 + +class BrightnessEnum(str, Enum): + """Enum for the backlight brightness + position devices can take string or number to move + """ + ON = 'on' + OFF = 'off' + class BeamlineState(str, Enum): ROBOT_SAMPLE_EXCHANGE = "robot_sample_exchange" SAMPLE_ALIGNMENT = "sample_alignment" @@ -43,6 +58,10 @@ class BECClientWorker: BEAMLINE = beamline.value.lower() if beamline is MXBeamline.SIMULATED: self.simulated = True + elif beamline is MXBeamline.X06DA: + self.simulated = True + logger.debug("simulating BECClientWorker on X06DA") + else: self.simulated = False host = cfg_get("daq.hardware.bec_url", f"{BEAMLINE}-bec-001.psi.ch") @@ -56,14 +75,37 @@ class BECClientWorker: self.macros = self.client.macros self.load_user_macros() self.helper = FrontendProcedureHelper(self.client.connector) + self.__set_scilog_tags() try: + self.position_devices = self.__initialise_sample_environment_devices() self.planner = self.__planner() + self.__backlight_brightness = self.position_devices['bl_bright'] + self.__detector_cover = self.position_devices['det_cov'] except Exception as e: - raise Exception(f"Error: {e}") + #raise self._raise_bec_error(e, operation="create planner") + self.planner = None + self.simulated = True - def _raise_bec_error(self, exc: Exception, *, operation: str) -> None: + self.__zoom = self.dev.scam_zoom + self.ring_current = self.dev.sls_current.get() + logger.debug(f"simulated is {self.simulated}") + + def _raise_bec_error(self, exc: Exception, *, operation: str, tags:Optional[List[str]] = None) -> None: message = f"BEC operation '{operation}' failed: {type(exc).__name__}: {exc}" - logger.exception(message) + #logger.exception(message) + if tags is None: + tags = ["error", "bec"] + else: + tags += ["error", "bec"] + try: + self.scilog_msg( + message=message, + error=True, + error_message=f"Error during '{operation}': {exc}", + tags=tags + ) + except Exception as e: + logger.error(f"Error sending scilog message: {e}") if isinstance(exc, AssertionError): raise BECCommunicationError( @@ -78,14 +120,108 @@ class BECClientWorker: exception=exc, ) from exc + def __initialise_sample_environment_devices(self): + """Initialise the sample environment devices as positioned devices from BEC config + Is a dictionary""" + if self.simulated: + logger.debug("Simulating initialise_devices") + return None + try: + return init_se_devices() + except Exception as e: + self._raise_bec_error(e, operation="initialise_devices") + + def list_position_devices(self): + """List the position devices available for the BEC worker""" + return list(self.position_devices.keys()) + + def __get_states(self): + """Get beamline states and modifiers from beamlien_states.yaml""" + if self.simulated: + logger.debug("Simulating initialise_devices") + return None + try: + states, allow_modifiers = get_states() + return states, allow_modifiers + except Exception as e: + self._raise_bec_error(e, operation="initialise beamline states") + def __planner(self): if self.simulated is None: logger.debug("Simulating planner") return None - d = init_se_devices() - states, allow_modifiers = get_states() - deps = planner_deps() - return StateManager(states=states, deps=deps, devices=d, allow_modifiers=allow_modifiers) + self.states, self.allow_modifiers = self.__get_states() + self.deps = planner_deps() + return StateManager( + states=self.states, + deps=self.deps, + devices=self.position_devices, + allow_modifiers=self.allow_modifiers + ) + + def __set_scilog_tags(self, tags:Optional[List[str]]=None): + try: + if tags: + self.client.messaging.scilog.set_default_tags(tags) + else: + self.client.messaging.scilog.set_default_tags(["AareDAQ"]) + except Exception as e: + logger.error(f"Error setting scilog tags: {e}") + self._raise_bec_error(e, operation="set scilog tags") + + def scilog_msg( + self, + message:str, + error:bool=False, + warning:bool=False, + error_message:Optional[str]=None, + attachments=None, + bold:bool=False, + italic:bool=False, + color:Optional[str]=None, + additonal_text:Optional[List[str]]=None, + tags:Optional[List[str]]=None + ): + if color and color not in ["red", "green", "yellow","blue","pink"]: + logger.warning("specified color not in allowed list,using default") + color = None + try: + msg = self.client.messaging.scilog.new() + except Exception as e: + self._raise_bec_error(e, operation="scilog_msg") + try: + msg.add_text(message, bold=bold, italic=italic, color=color) + if error: + msg.add_text(error_message, bold=True, color="red") + elif warning: + msg.add_text(error_message, bold=True, color="yellow") + except Exception as e: + logger.error(f"Error adding text: {e}") + msg.add_text(f"Error adding text: {e}") + try: + if attachments: + for attachment in attachments: + msg.add_attachment(attachment) + except Exception as e: + logger.error(f"Error adding attachment: {e}") + msg.add_text(f"Error adding attachment: {e}") + try: + if additonal_text: + for text in additonal_text: + msg.add_text(text) + except Exception as e: + logger.error(f"Error adding additional text: {e}") + msg.add_text(f"Error adding additional text: {e}") + try: + if tags: + msg.add_tags(tags) + except Exception as e: + logger.error(f"Error setting scilog tags: {e}") + msg.add_text(f"Error setting scilog tags: {e}") + try: + msg.send() + except Exception as e: + logger.error(f"Error sending scilog message: {e}") def run_macro(self, macro_name:str, *args, queue:str = "default", **kwargs): if self.simulated: @@ -112,6 +248,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}") @@ -152,13 +289,57 @@ class BECClientWorker: self.client.shutdown() def mono_pitch_scan_runner(self, plot: bool = False): - mono_pitch_scan(plot) + try: + mono_pitch_scan(plot) + except Exception as e: + self._raise_bec_error(e, operation="mono_pitch_scan", tags=["mono_pitch_scan"]) + self.scilog_msg( + message="Mono pitch scan completed", + bold=True, + color="green", + tags=["mono_pitch_scan"], + additonal_text=[f"New dcm_pitch position: {self.dev.dcm_pitch.position:5f}"] + ) + + def __get_current_energy(self): + """Get the current energy from the BEC in eV""" + return get_current_energy() + + def change_energy(self, value: float | int, plot: bool = False): + current_energy = self.__get_current_energy() + logger.info(f"Current energy: {current_energy:.1f} eV") + logger.info(f"Change energy requested: from {current_energy:.1f} to {value:.1f} eV") + try: + bl_energy(value, move_gap=False, mono_scan=True, plot=plot) + except Exception as e: + self._raise_bec_error( + e, + operation=f"Requested energy change from:" + f"{current_energy:.1f} to {value} eV", + tags=["energy_change"] + ) + + if abs(value - self.__get_current_energy()) > 1: + logger.warning(f"Energy change may have failed, current energy: {self.__get_current_energy()} eV") + + self.scilog_msg( + message=f"Moved from {current_energy:.1f} eV to {value:.1f} eV", + bold=True, + color="green", + tags=["energy_change"], + 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} ", + f"Previous energy: {current_energy:.1f} eV ", + f"Requested energy: {value:.1f} eV ", + f"New current energy: {self.__get_current_energy():.1f} eV"] + ) - def change_energy(self, value: float, plot: bool = False): - bl_energy(value, move_gap=False, mono_scan=True, plot=plot) def get_det_z(self): - return self.dev.det_z.position + try: + return self.dev.det_z.position + except Exception as e: + self._raise_bec_error(e, operation=f"get_det_z", tags=["det_z"]) def det_z(self, value:float, timeout:int | None = None ): """timeout is None or integer in s""" @@ -168,11 +349,123 @@ class BECClientWorker: status.wait(timeout=timeout) return status except Exception as e: - self._raise_bec_error(e, operation=f"scans.mv:det_z:{value}") + self._raise_bec_error(e, operation=f"scans.mv:det_z:{value}", tags=["det_z"]) + + def get_det_y(self): + try: + return self.dev.det_y.position + except Exception as e: + self._raise_bec_error(e, operation=f"get_det_z", tags=["det_z"]) + + def det_y(self, value:float, timeout:int | None = None ): + """timeout is None or integer in s""" + try: + status = self.scans.mv(self.dev.det_y, value, relative=False) + if timeout: + status.wait(timeout=timeout) + return status + except Exception as e: + self._raise_bec_error(e, operation=f"scans.mv:det_y:{value}", tags=["det_z"]) + + @property + def detector_cover(self) -> DetectorCoverEnum: + """Returns the detector cover POSTION. + If actual is True, returns the positon as an Enum: 2 is open, 1 is closed. + otherwise returns the position value. + """ + if self.simulated: + return DetectorCoverEnum.CLOSED + try: + return DetectorCoverEnum(self.__detector_cover.position) + except Exception as e: + self._raise_bec_error(e, operation=f"get_det_cov", tags=["det_cov"]) + raise + + @detector_cover.setter + def detector_cover(self, position:DetectorCoverEnum): + """Move the detector cover to the specified position""" + if self.simulated: + logger.info(f"Simulating detector cover move to {position.value}") + return + try: + self.__detector_cover.move(position) + except Exception as e: + self._raise_bec_error(e, operation=f"position_devices['det_cov'].move:{position.value}", tags=["det_cov"]) + raise + + def detector_cover_toggle(self): + """Toggle the detector cover""" + if self.simulated: + return + try: + current = self.detector_cover + if current is DetectorCoverEnum.OPEN: + target = DetectorCoverEnum.CLOSED + else: + target = DetectorCoverEnum.OPEN + self.detector_cover = target + except Exception as e: + self._raise_bec_error( + e, + operation=f"detector cover toggle, could not change cover open/close ", + tags=["det_cov"] + ) + + @property + def backlight_brightness(self) -> BrightnessEnum: + """returns backlight brightness as an Enum: 'off' or 'on', can also be a value... + How to handle""" + try: + return BrightnessEnum(self.__backlight_brightness.actual) + except Exception as e: + self._raise_bec_error( + e, + operation=f"backlight brightness, could not get backlight brightness", + tags=["backlight"] + ) + raise + + @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: + self._raise_bec_error(e, operation=f"backlight_brightness:{value}", tags=["backlight"]) + raise + + def get_backlight_pos(self) -> BrightnessEnum: + """Returns the current backlight brightness position""" + return BrightnessEnum(self.__backlight_brightness.pos) + + 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: + target = BrightnessEnum.OFF + else: + target = BrightnessEnum.ON + self.backlight_brightness = target + except Exception as e: + self._raise_bec_error( + e, + operation=f"backlight toggle, could not change backlight on/off ", + tags=["backlight"] + ) + + @property + def zoom(self): + return self.__zoom.position + + @zoom.setter + def zoom(self, value:float): + self.scans.umv(self.__zoom, value, relative=False) - # not implemented yet - # self.cryo_pos = PD.cryo_pos - # self.xrf_pos = PD.xrf_po if __name__ == "__main__": import time @@ -188,8 +481,45 @@ if __name__ == "__main__": except Exception as e: import sys 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.backlight_brightness) + print(client.get_backlight_pos()) + # 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 as e: + #client._raise_bec_error(e, operation="send message") + client.shutdown_client() + #print(f"Error: {e}") - print(client.is_state(BeamlineState.ROBOT_SAMPLE_EXCHANGE)) # try: # det_value = 980 # print(f"moving detector to vale:{det_value}")