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

Aerotech: added functionality to epics controller and to python api.

Browse Source
This commit is contained in:
appleb_m
2026-01-29 16:48:13 +01:00
parent f9970a66cb
commit 4aeee4bf95
1 changed files with 196 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aare/devices/aerotech.py
+196 -40
View File
@@ -5,36 +5,24 @@ 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
from aare.devices.my_motor import MyMotor
class TaskEnum(Enum):
TASK_0 = 0
TASK_1 = 1
TASK_2 = 2
TASK_3 = 3
TASK_4 = 4
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.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):
if not self.controller.runtime.commands.motion.is_enabled(axis.upper()):
self.enable_motion(axis)
self.controller.runtime.commands.motion.home(axis.upper())
class AxisEnum(Enum):
X = "gmx"
Y = "gmy"
Z = "gmz"
OMEGA = "Omega"
class AerotechRunEnum(Enum):
STOP = 0
@@ -97,10 +85,10 @@ class AerotechControllerEpics:
self.__task_run_enum.put(2)
def __task_pause(self):
self.__task_run_enum.put(3)
self.__task_run_enum.put(4)
def __task_load(self):
self.__task_run_enum.put(4)
self.__task_run_enum.put(3)
def __task_reset(self):
self.__task_run_enum.put(5)
@@ -119,19 +107,27 @@ class AerotechControllerEpics:
def set_offset(self, axis: MyMotor, offset: float):
self.__put(axis=axis, attr="OFF", value=offset)
def home_all(self):
def home_all(self, task_id: int = 3):
#self.__task_id.put(2)
self.__task_reset()
time.sleep(0.1)
self.__task_id.put(2)
time.sleep(0.2)
self.__task_id.put(task_id)
print(self.__task_id.get())
time.sleep(0.2)
self.__task_filename.put("home_all.a1exe")
print(self.__task_filename.get())
self.__task_load()
print(self.__task_run_enum.get())
# self.__task_load()
#time.sleep(0.2)
#print(self.__task_run_enum.get())
time.sleep(1.0)
self.__task_run()
print(self.__task_run_enum.get())
def get_tast_status(self, task_id: int = 3):
task_status = PV(f"{self.aerotech_pv_prefix}:TASK:T{task_id}:STATUS")
return task_status.get(as_string=True)
def __set_global(self, index:int, value: Union[int, float, str],
timeout:float=10.0):
"""Set a global variable in aerotech, it takes ~200 ms for the value to be set
@@ -178,22 +174,182 @@ class AerotechControllerEpics:
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 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 or 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 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)
if __name__ == "__main__":
beamline = mx_beamline()
print(beamline)
### test on 10S
#aerotech = AerotechController(controller_ip="129.129.118.96")
aerotech = AerotechController(controller_ip="129.129.118.96")
#aerotech.enable_motion("X")
#aerotech.home_motor("X")
#aerotech.disconnect()
aerotech_epics = AerotechControllerEpics(beamline)
aerotech.home_all()
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))
aerotech_epics.set_global_variable(0, 100.0)
print(aerotech_epics.get_global_variable(0, VariableTypeEnum.REAL))
#aerotech_epics.omega.move(90.0, wait=True)
#print(aerotech_epics.get_global_variable(0, VariableTypeEnum.REAL))
#aerotech_epics.set_global_variable(0, 100.0)
#print(aerotech_epics.get_global_variable(0, VariableTypeEnum.REAL))
# print('enable all motors')
# aerotech_epics.enable_all()
# print('home_all')
# aerotech_epics.home_all()
# print('wait for home to finish')
# start = time.perf_counter()
# status = aerotech_epics.get_tast_status(2)
# print(f'current status: {status}')
# if status == 'Idle':
# time.sleep(0.2)
# test_counter = 0
# while status != 'Ready':
# time.sleep(0.1)
# if time.perf_counter() - start > 360.0:
# raise TimeoutError("Timeout waiting for home all task to finish")
# elif status == 'Idle':
# time.sleep(0.5)
# if test_counter == 1:
# raise RuntimeError("Home all task failed to start")
# time.sleep(1.0)
# print('restarting home all')
# aerotech_epics.home_all()
# time.sleep(1.0)
# test_counter = 1
#
# status = arotech_epics.get_tast_status(2)
# for i in range(10):
# print(f'moving to {(i+1)*90}')
# aerotech_epics.omega.move(90.0, relative=True, wait=True)
# time.sleep(1.0)
# if i == 5:
# print('simulating disable')
# aerotech_epics._disable_all()
# time.sleep(10.0)
# print('re-enabling')
# aerotech_epics.enable_all()
# time.sleep(1.0)
# print('home all')
# aerotech_epics.home_all()
# print('wait for home to finish')
# start = time.perf_counter()
# status = aerotech_epics.get_tast_status(2)
# print(f'current status: {status}')
# test_counter = 0
# while status != 'Ready':
# time.sleep(0.1)
# if time.perf_counter() - start > 360.0:
# raise TimeoutError("Timeout waiting for home all task to finish")
# status = aerotech_epics.get_tast_status(2)
# time.sleep(0.5)
# if test_counter == 1:
# raise RuntimeError("Home all task failed to start")
# time.sleep(1.0)
# print('restarting home all')
# aerotech_epics.home_all()
# time.sleep(1.0)
# test_counter = 1
#aerotech_epics.home_all()