diff --git a/src/aare/devices/aerotech.py b/src/aare/devices/aerotech.py index 5884e40b..bed65a09 100644 --- a/src/aare/devices/aerotech.py +++ b/src/aare/devices/aerotech.py @@ -5,7 +5,7 @@ import time from enum import Enum from typing import Union -import automation1 as a1 +#import automation1 as a1 from epics import PV, Motor, caput, caget from aare.common.beamline import MXBeamline, mx_beamline @@ -173,150 +173,150 @@ class AerotechControllerEpics: def set_global_variable(self, index: int, value: Union[int, float, str]): self.__set_global(index, value) - -class AerotechController: - def __init__(self, controller_ip: str): - if controller_ip is None: - self.controller = None - else: - self.controller = a1.Controller.connect(controller_ip) - - self.status_item_configuration = a1.StatusItemConfiguration() - self.start_controller() - - def start_controller(self): - self.controller.start() - - def disconnect(self): - self.controller.disconnect() - - def enable_motion(self, axis: str): - self.controller.runtime.commands.motion.enable(axis.upper()) - - def home_motor(self, axis: str): - self.__configure_axis_status(axis, a1.AxisStatusItem.AxisEnabled) - result = self.get_status_via_status_items(name=axis, status_item=a1.AxisStatusItem.AxisEnabled) - if not result == a1.AxisStatusItem.AxisEnabled: - self.enable_motion(axis) - self.controller.runtime.commands.motion.home(axis.upper()) - - def __configure_axis_status(self, axis_name: str, axis_status_item: a1.AxisStatusItem): - self.status_item_configuration.axis.add(axis_status_item=axis_status_item, axis=axis_name) - - def __configure_task_status(self, task_id:int, task_status_item: a1.TaskStatusItem): - self.status_item_configuration.task.add(task_status_item=task_status_item, task=f"Task {task_id}") - - def get_status_via_status_items(self, name: str | int, status_item: a1.TaskStatusItem | a1.AxisStatusItem): - result = self.controller.runtime.status.get_status_items(self.status_item_configuration) - if int(name): - return result.task.get(status_item, f"Task {name}").value - elif str(name): - return result.axis.get(status_item, name).value - else: - raise ValueError(f"Invalid status item {status_item} for task {name}") - - def set_global_variable(self, index:int, value: Union[int, float, str]): - if type(value) is int: - self.controller.runtime.variables.global_.set_integer(index, value) - elif type(value) is float: - self.controller.runtime.variables.global_.set_real(index, value) - elif type(value) is str: - self.controller.runtime.variables.global_.set_string(index, value) - else: - raise ValueError(f"Invalid type {type(value)} for global variable") - - def get_axis_status_via_status_items(self, axis_name:str, status_item: a1.AxisStatusItem): - result = self.controller.runtime.status.get_status_items(self.status_item_configuration) - return result.task.get(status_item, axis_name).value - - def wait_for_status_to_change(self, name: str | int, status_item:a1.TaskStatusItem | a1.AxisStatusItem, enum, timeout: float = 60.0): - start = time.perf_counter() - while self.get_status_via_status_items(name, status_item) == enum: - time.sleep(0.1) - if time.perf_counter() - start > timeout: - print(self.get_status_via_status_items(name, status_item)) - raise TimeoutError(f"Timeout waiting for task {name} to finish") - return - - def wait_program_finish(self, task_id:int=3, timeout:float=60.0): - start = time.perf_counter() - status_item = a1.TaskStatusItem.TaskState - while True: - status = self.get_status_via_status_items(task_id, status_item) - if status != a1.TaskState.ProgramRunning: - if status == a1.TaskState.Idle: - return - elif status == a1.TaskState.ProgramComplete: - print(f"Program completed on Task {task_id}") - return - elif status == a1.TaskState.Error: - raise RuntimeError(f"Task {task_id} failed to start") - elif status == a1.TaskState.ProgramPaused: - print(f"Program paused on Task {task_id}, not sure how") - else: - raise RuntimeError(f"Unknown status {status} for task {task_id}") - if time.perf_counter() - start > timeout: - print(self.get_status_via_status_items(task_id, status_item)) - raise TimeoutError(f"Timeout waiting for task {task_id} to finish") - time.sleep(0.05) - print(f"Task {task_id} finished with status {status}") - - def __run_program(self, script_name:str, task_id:int=3, timeout:float=60.0): - self.__configure_task_status(task_id, a1.TaskStatusItem.TaskState) - state = self.get_status_via_status_items(task_id, a1.TaskStatusItem.TaskState) - print(f"Task {task_id} is in state {state}: {a1.TaskState(state).name}") - if state != a1.TaskState.ProgramComplete and state != a1.TaskState.Idle: - if a1.TaskState.ProgramRunning == state: - self.wait_for_status_to_change(task_id, a1.TaskStatusItem.TaskState, state, timeout) - elif a1.TaskState.ProgramComplete == state: - print('can continue') - else: - print(f"Task {task_id} is not Idle: {a1.TaskState(state).name} , aborting") - return - try: - print(f"Running script {script_name} on task {task_id}") - self.controller.runtime.tasks[task_id].program.run(script_name) - self.wait_program_finish(task_id, timeout) - except Exception as e: - print(f"Error executing script {script_name}: {e}") - - def run_grid_scan(self, cell_height_mm:float, num_rows:int, - row_width_mm:float, time_per_row_s: float, task_id:int =3): - self.set_global_variable(0, cell_height_mm) - self.set_global_variable(1, row_width_mm) - self.set_global_variable(2, time_per_row_s) - self.set_global_variable(1, num_rows) - self.set_global_variable(0, 1) - #self.__run_program(script_name="grid_scan.a1exe", task_id=task_id, timeout=120.0) - def home_all(self, task_id:int = 3): - self.__run_program(script_name="home_all.a1exe", task_id=task_id, timeout=120.0) - - def rotation_scan(self, task_id = 3, start_angle:float = 0.0, end_angle:float = 360.0, step_size:float = 1.0, num_steps:int = 10): - self.__run_program(script_name="rotation_scan.a1exe", task_id=task_id, timeout=90.0) - - def move_motor_absolute(self, axis:str, position:float, speed:float=1.0): - self.controller.runtime.commands.motion.moveabsolute(axis.upper(), [position], [speed]) - - def move_motor_linear(self, axis: str, position: float, speed: float = 1.0): - self.controller.runtime.commands.motion.movelinear(axis.upper(), [position], speed) - -if __name__ == "__main__": - beamline = mx_beamline() - print(beamline) - ### test on 10S - aerotech = AerotechController(controller_ip="129.129.118.96") - #aerotech.enable_motion("X") - #aerotech.home_all() - rw = 0.320 - ch = 0.010 - nr = 10 - tpr_s = rw / ch * 0.02 - #aerotech.run_grid_scan(cell_height_mm=ch, num_rows=nr, - # row_width_mm=rw, time_per_row_s=tpr_s) - st = time.perf_counter() - aerotech.move_motor_absolute("Z", 0, 10000) - print(f"time to move: {time.perf_counter() - st}") - aerotech.disconnect() +# +# class AerotechController: +# def __init__(self, controller_ip: str): +# if controller_ip is None: +# self.controller = None +# else: +# self.controller = a1.Controller.connect(controller_ip) +# +# self.status_item_configuration = a1.StatusItemConfiguration() +# self.start_controller() +# +# def start_controller(self): +# self.controller.start() +# +# def disconnect(self): +# self.controller.disconnect() +# +# def enable_motion(self, axis: str): +# self.controller.runtime.commands.motion.enable(axis.upper()) +# +# def home_motor(self, axis: str): +# self.__configure_axis_status(axis, a1.AxisStatusItem.AxisEnabled) +# result = self.get_status_via_status_items(name=axis, status_item=a1.AxisStatusItem.AxisEnabled) +# if not result == a1.AxisStatusItem.AxisEnabled: +# self.enable_motion(axis) +# self.controller.runtime.commands.motion.home(axis.upper()) +# +# def __configure_axis_status(self, axis_name: str, axis_status_item: a1.AxisStatusItem): +# self.status_item_configuration.axis.add(axis_status_item=axis_status_item, axis=axis_name) +# +# def __configure_task_status(self, task_id:int, task_status_item: a1.TaskStatusItem): +# self.status_item_configuration.task.add(task_status_item=task_status_item, task=f"Task {task_id}") +# +# def get_status_via_status_items(self, name: str | int, status_item: a1.TaskStatusItem | a1.AxisStatusItem): +# result = self.controller.runtime.status.get_status_items(self.status_item_configuration) +# if int(name): +# return result.task.get(status_item, f"Task {name}").value +# elif str(name): +# return result.axis.get(status_item, name).value +# else: +# raise ValueError(f"Invalid status item {status_item} for task {name}") +# +# def set_global_variable(self, index:int, value: Union[int, float, str]): +# if type(value) is int: +# self.controller.runtime.variables.global_.set_integer(index, value) +# elif type(value) is float: +# self.controller.runtime.variables.global_.set_real(index, value) +# elif type(value) is str: +# self.controller.runtime.variables.global_.set_string(index, value) +# else: +# raise ValueError(f"Invalid type {type(value)} for global variable") +# +# def get_axis_status_via_status_items(self, axis_name:str, status_item: a1.AxisStatusItem): +# result = self.controller.runtime.status.get_status_items(self.status_item_configuration) +# return result.task.get(status_item, axis_name).value +# +# def wait_for_status_to_change(self, name: str | int, status_item:a1.TaskStatusItem | a1.AxisStatusItem, enum, timeout: float = 60.0): +# start = time.perf_counter() +# while self.get_status_via_status_items(name, status_item) == enum: +# time.sleep(0.1) +# if time.perf_counter() - start > timeout: +# print(self.get_status_via_status_items(name, status_item)) +# raise TimeoutError(f"Timeout waiting for task {name} to finish") +# return +# +# def wait_program_finish(self, task_id:int=3, timeout:float=60.0): +# start = time.perf_counter() +# status_item = a1.TaskStatusItem.TaskState +# while True: +# status = self.get_status_via_status_items(task_id, status_item) +# if status != a1.TaskState.ProgramRunning: +# if status == a1.TaskState.Idle: +# return +# elif status == a1.TaskState.ProgramComplete: +# print(f"Program completed on Task {task_id}") +# return +# elif status == a1.TaskState.Error: +# raise RuntimeError(f"Task {task_id} failed to start") +# elif status == a1.TaskState.ProgramPaused: +# print(f"Program paused on Task {task_id}, not sure how") +# else: +# raise RuntimeError(f"Unknown status {status} for task {task_id}") +# if time.perf_counter() - start > timeout: +# print(self.get_status_via_status_items(task_id, status_item)) +# raise TimeoutError(f"Timeout waiting for task {task_id} to finish") +# time.sleep(0.05) +# print(f"Task {task_id} finished with status {status}") +# +# def __run_program(self, script_name:str, task_id:int=3, timeout:float=60.0): +# self.__configure_task_status(task_id, a1.TaskStatusItem.TaskState) +# state = self.get_status_via_status_items(task_id, a1.TaskStatusItem.TaskState) +# print(f"Task {task_id} is in state {state}: {a1.TaskState(state).name}") +# if state != a1.TaskState.ProgramComplete and state != a1.TaskState.Idle: +# if a1.TaskState.ProgramRunning == state: +# self.wait_for_status_to_change(task_id, a1.TaskStatusItem.TaskState, state, timeout) +# elif a1.TaskState.ProgramComplete == state: +# print('can continue') +# else: +# print(f"Task {task_id} is not Idle: {a1.TaskState(state).name} , aborting") +# return +# try: +# print(f"Running script {script_name} on task {task_id}") +# self.controller.runtime.tasks[task_id].program.run(script_name) +# self.wait_program_finish(task_id, timeout) +# except Exception as e: +# print(f"Error executing script {script_name}: {e}") +# +# def run_grid_scan(self, cell_height_mm:float, num_rows:int, +# row_width_mm:float, time_per_row_s: float, task_id:int =3): +# self.set_global_variable(0, cell_height_mm) +# self.set_global_variable(1, row_width_mm) +# self.set_global_variable(2, time_per_row_s) +# self.set_global_variable(1, num_rows) +# self.set_global_variable(0, 1) +# #self.__run_program(script_name="grid_scan.a1exe", task_id=task_id, timeout=120.0) +# def home_all(self, task_id:int = 3): +# self.__run_program(script_name="home_all.a1exe", task_id=task_id, timeout=120.0) +# +# def rotation_scan(self, task_id = 3, start_angle:float = 0.0, end_angle:float = 360.0, step_size:float = 1.0, num_steps:int = 10): +# self.__run_program(script_name="rotation_scan.a1exe", task_id=task_id, timeout=90.0) +# +# def move_motor_absolute(self, axis:str, position:float, speed:float=1.0): +# self.controller.runtime.commands.motion.moveabsolute(axis.upper(), [position], [speed]) +# +# def move_motor_linear(self, axis: str, position: float, speed: float = 1.0): +# self.controller.runtime.commands.motion.movelinear(axis.upper(), [position], speed) +# +# if __name__ == "__main__": +# beamline = mx_beamline() +# print(beamline) +# ### test on 10S +# aerotech = AerotechController(controller_ip="129.129.118.96") +# #aerotech.enable_motion("X") +# #aerotech.home_all() +# rw = 0.320 +# ch = 0.010 +# nr = 10 +# tpr_s = rw / ch * 0.02 +# #aerotech.run_grid_scan(cell_height_mm=ch, num_rows=nr, +# # row_width_mm=rw, time_per_row_s=tpr_s) +# st = time.perf_counter() +# aerotech.move_motor_absolute("Z", 0, 10000) +# print(f"time to move: {time.perf_counter() - st}") +# aerotech.disconnect() #aerotech_epics = AerotechControllerEpics(beamline) #aerotech_epics.omega.speed = 80.0