Aerotech: think we are removing automation1 api, WIP
This commit is contained in:
appleb_m committed
1 parent
6a7e61c1a7
commit
3c769106bf
1 file changed
+145
-145
+145
-145
@@ -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
|
||||
|
||||
Reference in new issue
Block a user