mx/AareDAQ
0
1
Fork 0
Code Issues 52 Pull Requests Actions Packages Projects Releases 72 Wiki Activity

DAQ: removed automation 1 dependency

Browse Source
This commit is contained in:
appleb_m
2026-05-12 17:05:19 +02:00
parent ab4dd89486
commit a413df48eb
1 changed files with 0 additions and 187 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aare/devices/automation1_python_api.py
-187
View File
@@ -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)