From a413df48eb1015c4bce7d95da994a50df3bad23e Mon Sep 17 00:00:00 2001 From: appleb_m Date: Tue, 12 May 2026 17:05:19 +0200 Subject: [PATCH] DAQ: removed automation 1 dependency --- src/aare/devices/automation1_python_api.py | 187 --------------------- 1 file changed, 187 deletions(-) delete mode 100644 src/aare/devices/automation1_python_api.py diff --git a/src/aare/devices/automation1_python_api.py b/src/aare/devices/automation1_python_api.py deleted file mode 100644 index f3ca8c8f..00000000 --- a/src/aare/devices/automation1_python_api.py +++ /dev/null @@ -1,187 +0,0 @@ -import os -import sys -import threading -import time -from enum import Enum -from typing import Union - -import automation1 as a1 - -from aare.common.beamline import MXBeamline, mx_beamline - -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() - controller = a1.Controller.connect("129.129.118.96") - status_item_configuration = a1.StatusItemConfiguration() - controller.start() - - axis = "X" - - pso_input = a1.PsoWindowInput.iXC4ePrimaryFeedback - window_number = 0 - reverse_direction = False - execution_task_index = 3 - min_x_mm = 0 - max_x_mm = 1 - - counts_per_unit = controller.runtime.parameters.axes[axis].units.countsperunit.value - print(f"Counts per unit for {axis}: {counts_per_unit}") - units_to_counts = controller.runtime.commands.utility_and_conversion.unitstocounts(axis,5,execution_task_index) - print(f"Units to counts for {axis}: {units_to_counts}") - print(dir(controller.runtime.commands)) - - primary_emulated_quadrature_divider = controller.runtime.parameters.axes[axis].feedback.primaryemulatedquadraturedivider.value - pso_lower_bound = round(controller.runtime.commands.utility_and_conversion.unitstocounts(axis,min_x_mm,execution_task_index)/primary_emulated_quadrature_divider) - pso_upper_bound = round(controller.runtime.commands.utility_and_conversion.unitstocounts(axis,max_x_mm,execution_task_index)/primary_emulated_quadrature_divider) - - controller.runtime.commands.pso.psoreset(axis) - controller.runtime.commands.pso.psowindowconfigureinput(axis,0, pso_input, True, execution_task_index) - controller.runtime.commands.pso.psowindowconfigurefixedrange(axis,window_number,pso_lower_bound,pso_upper_bound,execution_task_index) - controller.runtime.commands.pso.psowindowoutputon(axis,window_number, execution_task_index) - controller.runtime.commands.motion.movelinear(axis,[1],0.1,execution_task_index) - controller.runtime.commands.motion.waitformotiondone(axis,execution_task_index) - controller.runtime.commands.motion.movelinear(axis,[-1],0.1,execution_task_index) - controller.runtime.commands.motion.waitformotiondone(axis,execution_task_index) - controller.runtime.commands.pso.psowindowoutputoff(axis,window_number, execution_task_index) - controller.runtime.commands.pso.psoreset(axis,execution_task_index) \ No newline at end of file