Added further fucntionality to the automation 1 python api wrapper

This commit is contained in:
2026-02-06 15:56:41 +01:00
parent ab6ecff677
commit ead99f0457
+36 -5
View File
@@ -215,6 +215,15 @@ class AerotechController:
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)
@@ -256,13 +265,13 @@ class AerotechController:
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 or state != a1.TaskState.Idle:
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")
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}")
@@ -271,22 +280,44 @@ class AerotechController:
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()
#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()
print('initialise aerotech EPICS')
#aerotech_epics = AerotechControllerEpics(beamline)
#aerotech_epics.omega.speed = 80.0
#print(aerotech_epics.get_global_variable(0, VariableTypeEnum.REAL))