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

ALC: remove background function and button

Browse Source
This commit is contained in:
appleb_m committed 2026-03-24 09:28:11 +01:00
1 parent c25e20a10c
commit 322370ac08
5 files changed
+145 -394
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/aare/daq/daq.py
-25
View File
@@ -754,31 +754,6 @@ class AareDAQ:
self.__cfg.state_busy = False
raise e
def get_background(self):
#if self.sample is not None:
# raise Exception("Background cannot be measured when sample is mounted")
self.__cfg.try_set_busy(timeout=360)
self.__set_state(BeamlineStateEnum.SampleAlignment)
try:
self.__devs.lamp_light = 2.5
self.__cfg.zoom_mode = ZoomModeEnum.LoopCenter
zoom_settings = self.__cfg.zoom_settings.z
for zoom_value in zoom_settings:
exposure = zoom_settings[zoom_value].exposure
gain = zoom_settings[zoom_value].gain
self.__devs.samcam_settings = SampleCameraSettings(exposure=exposure, gain=gain)
self.__devs.set_zoom(zoom_value, wait=True)
time.sleep(5.0) # Wait for settings to stabilize
image = self.__devs.samcam_get_image(gray=False)
self.__cfg.put_alc_bkg(zoom_value, exposure, gain, image)
bgr_array = cv2.cvtColor(image, cv2.COLOR_RGB2BGR)
cv2.imwrite(f"bkg{zoom_value:.0f}_{exposure*1000:.0f}_{gain:.0f}.jpg", bgr_array)
self.__cfg.state_busy = False
except Exception as e:
self.__cfg.state_busy = False
raise e
@property
def dtz(self) -> float:
tmp = self.__cfg.dtz
src/aare/daq/server.py
-7
View File
@@ -611,13 +611,6 @@ async def cancel(token: str = Depends(oauth2_scheme)):
daq.cancel()
# ALC routines
@app.post("/alc/background")
async def alc_background(token: str = Depends(oauth2_scheme)) -> str:
auth.check_jwt_rw(cfg, auth.parse_token(token))
daq.get_background()
return "OK"
@app.post("/alc/center_loop")
async def alc_center_loop(token: str = Depends(oauth2_scheme)) -> str:
logger.debug(f"ALC")
src/aare/devices/aerotech.py
+145 -359
View File
@@ -1,386 +1,172 @@
import os
import sys
import threading
import time
from enum import Enum
from typing import Union
from typing import Optional, Union
import automation1 as a1
from epics import PV, Motor, caput, caget
from aarescan_client.models.grid_request import GridRequest
from aarescan_client.models.screen_request import ScreenRequest
from aare.common.beamline import MXBeamline, mx_beamline
from aare.devices.my_motor import MyMotor
from aarescan_client import ApiClient, Status, RotationRequest, Configuration, DefaultApi, AxisStatus, Target
class TaskEnum(Enum):
TASK_0 = 0
TASK_1 = 1
TASK_2 = 2
TASK_3 = 3
TASK_4 = 4
from aare.common.coordinate import AerotechCoordinate, Coordinate
class AxisEnum(Enum):
X = "gmx"
Y = "gmy"
Z = "gmz"
OMEGA = "Omega"
class AerotechRunEnum(Enum):
STOP = 0
START = 1
RUN = 2
LOAD = 3
PAUSE = 4
RESET = 5
class VariableTypeEnum(Enum):
INT = 0
REAL = 1
STRING = 2
class AerotechControllerEpics:
def __init__(self, beamline: MXBeamline):
BEAMLINE = beamline.value.upper()
self.aerotech_pv_prefix = f"{BEAMLINE}-ES-DF1"
trx_prefix = f"{self.aerotech_pv_prefix}:TRX1"
try_prefix = f"{self.aerotech_pv_prefix}:TRY1"
trz_prefix = f"{self.aerotech_pv_prefix}:TRZ1"
rotu_prefix = f"{self.aerotech_pv_prefix}:ROTU"
self.gmx = MyMotor(trx_prefix)
self.gmy = MyMotor(try_prefix)
self.gmz = MyMotor(trz_prefix)
self.omega = MyMotor(rotu_prefix)
self.__enable_all = PV(f"{self.aerotech_pv_prefix}:EnableAll")
self.__disable_all = PV(f"{self.aerotech_pv_prefix}:DisableAll")
self.__acknowledge_all = PV(f"{self.aerotech_pv_prefix}:AckAll")
self.__stop_all = PV(f"{self.aerotech_pv_prefix}:StopAll")
self.__task_filename = PV(f"{self.aerotech_pv_prefix}:TASK:FILENAME")
self.__task_id = PV(f"{self.aerotech_pv_prefix}:TASK:TASKIDX") # values can be 1 to 4 DO NOT USE 0!!!!
self.__task_run_enum = PV(f"{self.aerotech_pv_prefix}:TASK:SWITCH") # 0 Stop, 1 Start, 2 Run,
# 3 Load, 4 Pause, 5 Reset
self.enable_all()
def enable_all(self):
self.__enable_all.put(1)
self.__enable_all.put(0)
def acknowledge_all(self):
self.__acknowledge_all.put(1)
self.__acknowledge_all.put(0)
def _disable_all(self):
self.__disable_all.put(1)
self.__disable_all.put(0)
def stop_all(self):
self.__stop_all.put(1)
self.__stop_all.put(0)
def __task_stop(self):
self.__task_run_enum.put(0)
def __task_start(self):
self.__task_run_enum.put(1)
def __task_run(self):
self.__task_run_enum.put(2)
def __task_pause(self):
self.__task_run_enum.put(4)
def __task_load(self):
self.__task_run_enum.put(3)
def __task_reset(self):
self.__task_run_enum.put(5)
def __set_task_id(self, task_id: int):
if task_id not in range(1, 5):
raise ValueError(f"Invalid task id {task_id}")
self.__task_id.put(task_id)
def __put(self, axis: MyMotor, attr: str, value: float):
try:
axis.put(attr=attr, value=value)
except Exception as e:
raise ValueError(f"Error setting {axis.name} {attr} to {value}: {e}")
def set_offset(self, axis: MyMotor, offset: float):
self.__put(axis=axis, attr="OFF", value=offset)
def home_all(self, task_id: int = 3):
#self.__task_id.put(2)
self.__task_reset()
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()
#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)
AEROTECH_HOME = AerotechCoordinate(at_mm=Coordinate(x=0, y=0, z=0), omega_deg=0)
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
:param index: select a variable to change. an integer between 0..256 for int and real variable for 0..31 for strings
:param value: the value to set can be int, float (REAL) or string
:param timeout: timeout for vairable change, default 10.0 seconds """
var_type = self.__get_var_type(value)
caput(f"{self.aerotech_pv_prefix}:VAR:{var_type.upper()}-ADDR", index)
if var_type == 'STRING':
var_type = 'STRING-SHORT'
caput(f"{self.aerotech_pv_prefix}:VAR:{var_type.upper()}", value)
start = time.perf_counter()
while time.perf_counter() - start < timeout:
rbv = self.__read_global_from_index(index, var_type)
if rbv == str(value):
return
time.sleep(0.1)
class AerotechController(object):
raise TimeoutError(f"Timeout setting global variable {index} to {value}")
def __read_global_feedback(self, index, var_type: str):
caput(f"{self.aerotech_pv_prefix}:VAR:{var_type.upper()}-RBV.PROC", 1)
time.sleep(0.5)
return caget(f"{self.aerotech_pv_prefix}:VAR:{var_type.upper()}-RBV", as_string=True)
def __read_global_from_index(self, index, var_type: str):
return caget(f"{self.aerotech_pv_prefix}:VAR:{var_type.upper()}{index}_RBV", as_string=True)
def __get_var_type(self, value):
if type(value) is int:
return VariableTypeEnum.INT.name
elif type(value) is float:
return VariableTypeEnum.REAL.name
elif type(value) is str:
return VariableTypeEnum.STRING.name
def __init__(self, bl: MXBeamline):
if bl == MXBeamline.X06DA:
self.__simulated = False
self.__base = "http://x06da-smargopolo.psi.ch:5234"
elif bl == MXBeamline.X10SA:
self.__simulated = False
self.__base = "http://x10sa-smargopolo.psi.ch:5234"
elif bl == MXBeamline.SIMULATED:
self.__simulated = True
self.__pos = AEROTECH_HOME
self.__vel = 0
else:
raise ValueError(f"Invalid type {type(value)} for global variable")
raise Exception("unknown beamline")
def get_global_variable(self, index: int, var_type:VariableTypeEnum):
return self.__read_global_from_index(index, var_type.name)
if not self.__simulated:
self.__client = ApiClient(Configuration(host=self.__base))
self.__api = DefaultApi(self.__client)
def set_global_variable(self, index: int, value: Union[int, float, str]):
self.__set_global(index, value)
def __make_aerotech_target(
self,
coord: AerotechCoordinate,
wait: bool = False,
incremental: bool = False,
) -> Target:
at_mm = coord.at_mm
return Target(
x=at_mm.x if at_mm is not None else None,
y=at_mm.y if at_mm is not None else None,
z=at_mm.z if at_mm is not None else None,
u=coord.omega_deg,
var_async=not wait,
incremental=incremental,
)
def __make_aerotech_coordinate(self, target: Target) -> AerotechCoordinate:
return AerotechCoordinate(at_mm=Coordinate(x=target.x,y=target.y,z=target.z), omega_deg=target.u)
def cancel(self):
self.__api.cancel_post()
def is_idle(self) -> bool:
status = self.__api.status_get()
return status.state == 'Idle'
def get_position(self) -> AerotechCoordinate:
status = self.__api.status_get()
return AerotechCoordinate(
at_mm=Coordinate(
x=status.x.pos,
y=status.y.pos,
z=status.z.pos,
),
omega_deg=status.u.pos,
)
class AerotechController:
def __init__(self, controller_ip: str):
if controller_ip is None:
self.controller = None
else:
self.controller = a1.Controller.connect(controller_ip)
def status(self) -> Status:
if self.__simulated:
return Status(state=Status.State.IDLE,
x=AxisStatus(pos=self.__pos.x, vel=self.__vel,
enabled=False, homed=False, moving=False,fault=False),
y=AxisStatus(pos=self.__pos.y, vel=self.__vel,
enabled=False, homed=False, moving=False,fault=False),
z=AxisStatus(pos=self.__pos.z, vel=self.__vel,
enabled=False, homed=False, moving=False,fault=False),
u=AxisStatus(pos=self.__pos.u, vel=self.__vel,
enabled=False, homed=False, moving=False,fault=False),
)
return self.__api.status_get()
self.status_item_configuration = a1.StatusItemConfiguration()
self.start_controller()
def move_home(self, wait:bool=True, incremental:bool=False):
if self.__simulated:
self.__pos = AEROTECH_HOME
return self.__pos
return self.position(AEROTECH_HOME, wait=wait, incremental=incremental)
def start_controller(self):
self.controller.start()
def home_aerotech(self):
return self.__api.home_post()
def disconnect(self):
self.controller.disconnect()
def wait_till_done(self, timeout=60):
return self.__api.wait_till_done_post(timeout=timeout)
def enable_motion(self, axis: str):
self.controller.runtime.commands.motion.enable(axis.upper())
def position(
self,
target: AerotechCoordinate,
/,
wait: bool = True,
incremental: bool = False,
):
if self.__simulated:
return self.__pos
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())
payload = self.__make_aerotech_target(target, wait=wait, incremental=incremental)
return self.__api.position_post(payload)
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 rotation_scan(self, rotation_deg: float | int,
time_sec: float | int,
start_pos_deg: float | int,
var_async: bool = False):
payload = RotationRequest(
rotation_deg=rotation_deg,
time_sec=time_sec,
start_pos_deg=start_pos_deg,
var_async=var_async
)
if self.__simulated:
return payload
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}")
return self.__api.rotation_scan_post(payload)
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 grid_scan(self,
grid_elem_count_y: int,
grid_elem_size_y_um: int | float,
time_sec: int|float,
grid_elem_size_x_um: Optional[Union[float, int]] = None,
grid_elem_count_x: Optional[int] = None,
var_async: Optional[bool] = False
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")
):
payload = GridRequest(
grid_elem_count_x=grid_elem_count_x,
grid_elem_count_y=grid_elem_count_y,
grid_elem_size_x_um=grid_elem_size_x_um,
grid_elem_size_y_um=grid_elem_size_y_um,
time_sec=time_sec,
var_async=var_async,
)
if self.__simulated:
return payload
return self.__api.grid_scan_post(payload)
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 screening_scan(self,
rotation_deg: float | int,
wedge_deg: float | int,
time_sec: float | int,
steps: int,
var_async: bool = False
):
payload = ScreenRequest(
rotation_deg=rotation_deg,
wedge_deg=wedge_deg,
time_sec=time_sec,
steps=steps,
var_async=var_async
)
if self.__simulated:
return payload
return self.__api.screening_post(payload)
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()
#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))
# 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()
#print(beamline)
controller = AerotechController(beamline)
#controller.print_status(colored=True, compact=True)
print(controller.get_position())
src/aare/gui/main_window.py
-1
View File
@@ -314,7 +314,6 @@ class MainWindow(QMainWindow):
self.beamline.samcam.changed.connect(self.daq.samcam_settings)
self.beamline.samcam.screenshot_requested.connect(self.daq.send_screenshot_db)
self.beamline.loopctr.background.clicked.connect(self.daq.alc_background)
self.beamline.loopctr.find_tip.clicked.connect(self.daq.center_loop)
self.beamline.loopctr.bounding_box.clicked.connect(self.daq.ml_bounding_box)
self.daq.raster_generated_by_ml.connect(self.raster.update_active_grid_request)
src/aare/gui/panels/loop_centering_panel.py
-2
View File
@@ -14,8 +14,6 @@ class LoopCenteringPanel(QWidget):
self.find_tip = QPushButton("Center", parent=self)
grid_layout.addWidget(self.find_tip, 1, 0)
self.background = QPushButton("Bkg", parent=self)
grid_layout.addWidget(self.background, 1, 1)
self.bounding_box = QPushButton("Box", parent=self)
grid_layout.addWidget(self.bounding_box, 1, 2)