From ead99f0457e1ffd0a00b2e80a12d0a92c12c7d62 Mon Sep 17 00:00:00 2001 From: appleb_m Date: Fri, 6 Feb 2026 15:56:41 +0100 Subject: [PATCH] Added further fucntionality to the automation 1 python api wrapper --- src/aare/devices/aerotech.py | 41 +++++++++++++++++++++++++++++++----- 1 file changed, 36 insertions(+), 5 deletions(-) diff --git a/src/aare/devices/aerotech.py b/src/aare/devices/aerotech.py index 89719ca6..5884e40b 100644 --- a/src/aare/devices/aerotech.py +++ b/src/aare/devices/aerotech.py @@ -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))