bec: converting epics devices in use to bec
This commit is contained in:
appleb_m committed
1 parent
7b2cdd3ae4
commit
e8c381dfa1
1 file changed
+346
-16
+346
-16
@@ -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}")
|
||||
|
||||
Reference in new issue
Block a user