662 lines
25 KiB
Python
662 lines
25 KiB
Python
import time
|
|
from enum import Enum
|
|
from typing import List, Optional
|
|
|
|
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
|
|
from aarecommon.models.beamline import MXBeamline
|
|
from bec_ipython_client import BECIPythonClient
|
|
from bec_ipython_client.signals import OperationMode
|
|
from bec_lib.procedures.helper import FrontendProcedureHelper
|
|
from bec_lib.service_config import ServiceConfig
|
|
|
|
logger = setup_logger("aareDAQ")
|
|
|
|
# specify up to 10 queue to runs in parallel, request more if needed!
|
|
# st = client.proc.request_new("sleep", ((), {"time_s":5}), queue="test")
|
|
|
|
# to see all deevices
|
|
# devs.show_all
|
|
|
|
# helper fucntions
|
|
# helper.get.active_and_pending_queue_names()
|
|
# 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"
|
|
DATA_COLLECTION = "data_collection"
|
|
DC_XRF = "DC_XRF"
|
|
MANUAL_SAMPLE_EXCHANGE = "manual_sample_exchange"
|
|
BEAM_VISUALISATION = "beam_visualisation"
|
|
FLUX_MEASUREMENT = "flux_measurement"
|
|
BEAMSTOP_ALIGNMENT = "beamstop_alignment"
|
|
MAINTENANCE = "maintenance"
|
|
XTAL_SNAPSHOT = "xtal_snapshot"
|
|
|
|
|
|
class BECClientWorker:
|
|
def __init__(self, beamline: MXBeamline, name: str = "default"):
|
|
BEAMLINE = beamline.value.lower()
|
|
self.beamline = beamline
|
|
if self.beamline is MXBeamline.X06DA:
|
|
self._beamline_name = "pxiii"
|
|
elif self.beamline is MXBeamline.X10SA:
|
|
self._beamline_name = "pxii"
|
|
elif self.beamline is MXBeamline.X06DA:
|
|
self._beamline_name = "pxi"
|
|
elif self.beamline is MXBeamline.SIMULATED:
|
|
self._beamline_name = "SIMULATED"
|
|
else:
|
|
raise ValueError(f"Unknown beamline: {beamline}")
|
|
|
|
if self.beamline is MXBeamline.SIMULATED:
|
|
self.simulated = True
|
|
|
|
else:
|
|
self.simulated = False
|
|
logger.debug(f"Initializing BECClientWorker for {BEAMLINE} beamline")
|
|
host = cfg_get("daq.hardware.bec_url", f"{BEAMLINE}-bec-001.psi.ch")
|
|
service_config = ServiceConfig(redis={"host": host, "port": 6379})
|
|
service_config.config["log_writer"]["base_path"] = "/tmp/logs"
|
|
# service_config.config["user_macros"]["base_path"]=f'/sls/{BEAMLINE}/config/bec/production/pxiii_bec/pxiii_bec'
|
|
# print(service_config.config)
|
|
self.client = BECIPythonClient(config=service_config, mode=OperationMode.Procedure)
|
|
self.client.start()
|
|
# self.client.config.update_session_with_file("/sls/x10sa/config/bec/production/bec/bec_lib/bec_lib/config_helper.py")
|
|
self.dev = self.client.device_manager.devices
|
|
print(self.dev.keys())
|
|
self.scans = self.client.scans
|
|
self.macros = self.client.macros
|
|
self.__load_user_macros()
|
|
print(self.__list_all_macros())
|
|
self.helper = FrontendProcedureHelper(self.client.connector)
|
|
self.__set_scilog_tags()
|
|
try:
|
|
self.__init_beamline_environment()
|
|
except Exception as e:
|
|
logger.error(f"Error initialising BEC devices: {e}")
|
|
exit(1)
|
|
logger.debug(f"simulated is {self.simulated}")
|
|
|
|
def __init_beamline_environment(self):
|
|
try:
|
|
self.position_devices, self.planner = init_beamline_environment()
|
|
self.__backlight_brightness = self.position_devices["bl_bright"]
|
|
self.__frontlight_brightness = self.position_devices["fl_bright"]
|
|
self.__zoom = self.dev.scam_zoom
|
|
self._ring_current = self.dev.sls_current
|
|
except Exception as e:
|
|
logger.error(f"Error initialising planar and position devices: {e}")
|
|
self.position_devices = None
|
|
self.planner = None
|
|
self.__backlight_brightness = None
|
|
self.__frontlight_brightness = None
|
|
try:
|
|
self.__zoom = self.dev.scam_zoom
|
|
self._ring_current = self.dev.sls_current
|
|
except Exception as e:
|
|
logger.error(f"Error initialising zoom and ring_current: {e}")
|
|
self.__zoom = None
|
|
self.ring_current = None
|
|
raise Exception(f"Error initialising BEC devices: {e}")
|
|
|
|
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)
|
|
if tags is None:
|
|
tags = ["error", "bec"]
|
|
else:
|
|
tags += ["error", "bec"]
|
|
try:
|
|
bec_alarms = self.client.alarms()
|
|
if bec_alarms:
|
|
alarms = [alarm for alarm in bec_alarms]
|
|
logger.debug(f"alarms: {alarms}")
|
|
last_alarm = self.client.show_last_alarm()
|
|
message += f"\n\nAlarm: {last_alarm}\n\n"
|
|
logger.error(f"last_alarm: {last_alarm}")
|
|
except Exception as e:
|
|
logger.error(f"Couldn't raise BEC alarms: {e}")
|
|
|
|
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(
|
|
f"BEC internal assertion failed during '{operation}'",
|
|
operation=operation,
|
|
exception=exc,
|
|
) from exc
|
|
|
|
raise BECCommunicationError(message, operation=operation, exception=exc) from exc
|
|
|
|
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:
|
|
logger.debug(f"Simulating macro {macro_name}")
|
|
return None
|
|
try:
|
|
return self.client.proc.run_macro(macro_name, *args, queue=queue)
|
|
except Exception as e:
|
|
self._raise_bec_error(e, operation=f"run_macro:{macro_name}")
|
|
|
|
def run_macro_blocked(self, macro_name: str, *args, queue: str = "default", **kwargs):
|
|
if self.simulated:
|
|
logger.debug(f"Simulating macro {macro_name}")
|
|
return None
|
|
try:
|
|
status = self.run_macro(macro_name, *args, queue=queue)
|
|
print(status)
|
|
status.wait()
|
|
print(status)
|
|
return status
|
|
except Exception as e:
|
|
self._raise_bec_error(e, operation=f"run_macro_blocked:{macro_name}")
|
|
|
|
@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):
|
|
logger.info(
|
|
f"BEC move_to completed: {state.value} in {time.perf_counter() - start:.2f}s"
|
|
)
|
|
return True
|
|
else:
|
|
logger.warning(
|
|
f"Failed to move to {state.value} in {time.perf_counter() - start:.2f}s"
|
|
)
|
|
return False
|
|
except Exception as e:
|
|
self._raise_bec_error(e, operation=f"planner.move_to:{state.value}")
|
|
|
|
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:
|
|
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:
|
|
self._raise_bec_error(e, operation="list_all_user_macros")
|
|
|
|
def __list_all_macros(self):
|
|
result = self.macros.list_user_macros()
|
|
if result is None:
|
|
return []
|
|
return result
|
|
|
|
def load_user_macros(self):
|
|
if self.simulated:
|
|
logger.debug("Simulating load_user_macros")
|
|
return None
|
|
try:
|
|
return self.__load_user_macros()
|
|
except Exception as e:
|
|
self._raise_bec_error(e, operation="load_user_macros")
|
|
|
|
def __load_user_macros(self):
|
|
result = self.macros.load_all_user_macros()
|
|
if result is None:
|
|
logger.warning("BEC load_all_user_macros returned None")
|
|
return result
|
|
|
|
def reinitialise_planner_and_position_devices(self, method: str = "auto"):
|
|
"""
|
|
Reinitialise BEC position devices and planner.
|
|
|
|
Args:
|
|
method:
|
|
"auto" - use the beamline-specific default.
|
|
"beamline" - force init_beamline_environment().
|
|
"sample" - force init_se_devices() + planner creation.
|
|
|
|
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(
|
|
f"/sls/{self.beamline}/config/bec/production/{self._beamline_name}_bec/{self._beamline_name}_bec/device_configs/{self._beamline_name}-devices.yaml"
|
|
)
|
|
self.__init_beamline_environment()
|
|
logger.info(f"Reinitialised BEC planner and position devices using method={method}")
|
|
return self.list_position_devices()
|
|
except Exception as e:
|
|
self._raise_bec_error(
|
|
e, operation=f"reinitialise_planner_and_position_devices:{method}"
|
|
)
|
|
|
|
def shutdown_client(self):
|
|
self.client.shutdown()
|
|
|
|
def mono_pitch_scan_runner(self, plot: bool = False):
|
|
try:
|
|
mono_pitch_scan(plot)
|
|
except Exception as e:
|
|
self._raise_bec_error(e, operation="mono_pitch_scan", tags=["mono_pitch_scan"])
|
|
if self.beamline is MXBeamline.X06DA:
|
|
addtional_text = [f"New dcm_pitch position: {self.dev.dcm_pitch.position:5f}"]
|
|
else:
|
|
addtional_text = [f"New dcm_theta2 position: {self.dev.dccm_theta2.position:5f}"]
|
|
self.scilog_msg(
|
|
message="Mono pitch scan completed",
|
|
bold=True,
|
|
color="green",
|
|
tags=["mono_pitch_scan"],
|
|
additonal_text=addtional_text,
|
|
)
|
|
|
|
def check_current_energy(self):
|
|
"""Get the current energy from the BEC in eV"""
|
|
energy_ev = get_current_energy()
|
|
energy_kev = energy_ev / 1000
|
|
return energy_kev
|
|
|
|
def change_energy(self, value: float | int, plot: bool = False):
|
|
current_energy = self.check_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:{current_energy:.1f} to {value} eV",
|
|
tags=["energy_change"],
|
|
)
|
|
|
|
if abs(value - self.check_current_energy()) > 1:
|
|
logger.warning(
|
|
f"Energy change may have failed, current energy: {self.check_current_energy()} eV"
|
|
)
|
|
if beamline is 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} ",
|
|
f"Previous energy: {current_energy:.1f} eV ",
|
|
f"Requested energy: {value:.1f} eV ",
|
|
f"New current energy: {self.check_current_energy():.1f} eV",
|
|
]
|
|
else:
|
|
additonal_text = [
|
|
f"New dccm_theta1 position: {self.dev.dccm_theta1.position:4g} mrad",
|
|
f"New dccm_theta2 position: {self.dev.dccm_theta2.position:4g} mrad",
|
|
f"Previous energy: {current_energy:.1f} eV ",
|
|
f"Requested energy: {value:.1f} eV ",
|
|
f"New current energy: {self.check_current_energy():.1f} 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=additonal_text,
|
|
)
|
|
|
|
def get_det_z(self):
|
|
try:
|
|
return self.dev.det_z.position
|
|
except Exception as e:
|
|
self._raise_bec_error(e, operation="get_det_z", tags=["det_z"])
|
|
|
|
def det_z(self, value: float, timeout: int | None = None):
|
|
"""timeout is None or integer in s"""
|
|
try:
|
|
status = self.scans.mv(self.dev.det_z, 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_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="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 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="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="backlight toggle, could not change backlight on/off ",
|
|
tags=["backlight"],
|
|
)
|
|
|
|
def save_current_bs_pos(self):
|
|
save_current_position(self.dev.bs_z, "safe")
|
|
|
|
def save_current_collimator_pos(self):
|
|
save_current_position(self.dev.coll_y, "work")
|
|
|
|
def save_current_aerotech_position(self):
|
|
save_current_position(self.dev.aerotech, "work", axis="x")
|
|
save_current_position(self.dev.aerotech, "work", axis="y")
|
|
save_current_position(self.dev.aerotech, "work", axis="z")
|
|
|
|
self.save_config_and_reload_devices()
|
|
|
|
def save_config_and_reload_devices(self):
|
|
self.position_devices, self.planner = save_and_reload()
|
|
|
|
@property
|
|
def zoom(self):
|
|
return self.__zoom.position
|
|
|
|
@zoom.setter
|
|
def zoom(self, value: float):
|
|
self.scans.umv(self.__zoom, value, relative=False)
|
|
|
|
@property
|
|
def ring_current(self) -> float:
|
|
return self._ring_current.get()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import time
|
|
|
|
print(time.ctime(), " starting BEC Client")
|
|
beamline = mx_beamline()
|
|
try:
|
|
client = BECClientWorker(beamline)
|
|
except Exception as e:
|
|
print(f"Error: {e}")
|
|
try:
|
|
client.shutdown_client()
|
|
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.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 as e:
|
|
# client._raise_bec_error(e, operation="send message")
|
|
client.shutdown_client()
|
|
print(f"Error: {e}")
|
|
|
|
# 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()
|