Aerotech: added custom exception for communication error
This commit is contained in:
1 parent
3d4e252438
commit
98f665370a
1 file changed
+86
-12
@@ -7,6 +7,7 @@ from aare.common.beamline import MXBeamline, mx_beamline
|
||||
from aarescan_client import ApiClient, Status, RotationRequest, Configuration, DefaultApi, AxisStatus, Target
|
||||
|
||||
from aare.common.coordinate import AerotechCoordinate, Coordinate
|
||||
from aare.common.exception_handler import AerotechCommunicationError
|
||||
|
||||
AEROTECH_HOME = AerotechCoordinate(at_mm=Coordinate(x=0, y=0, z=0), omega_deg=0)
|
||||
|
||||
@@ -52,14 +53,30 @@ class AerotechController(object):
|
||||
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()
|
||||
try:
|
||||
self.__api.cancel_post()
|
||||
except Exception as e:
|
||||
raise AerotechCommunicationError(
|
||||
"Aerotech cancel failed",
|
||||
endpoint="cancel_post",
|
||||
base_url=self.__base,
|
||||
operation="POST",
|
||||
) from e
|
||||
|
||||
def is_idle(self) -> bool:
|
||||
status = self.__api.status_get()
|
||||
return status.state == 'Idle'
|
||||
try:
|
||||
status = self.__api.status_get()
|
||||
return status.state == 'Idle'
|
||||
except Exception as e:
|
||||
raise AerotechCommunicationError(
|
||||
"Aerotech status check failed",
|
||||
endpoint="status_get",
|
||||
base_url=self.__base,
|
||||
operation="GET",
|
||||
) from e
|
||||
|
||||
def get_position(self) -> AerotechCoordinate:
|
||||
status = self.__api.status_get()
|
||||
status = self.status()
|
||||
return AerotechCoordinate(
|
||||
at_mm=Coordinate(
|
||||
x=status.x.pos,
|
||||
@@ -69,7 +86,6 @@ class AerotechController(object):
|
||||
omega_deg=status.u.pos,
|
||||
)
|
||||
|
||||
|
||||
def status(self) -> Status:
|
||||
if self.__simulated:
|
||||
return Status(state=Status.State.IDLE,
|
||||
@@ -82,7 +98,15 @@ class AerotechController(object):
|
||||
u=AxisStatus(pos=self.__pos.u, vel=self.__vel,
|
||||
enabled=False, homed=False, moving=False,fault=False),
|
||||
)
|
||||
return self.__api.status_get()
|
||||
try:
|
||||
return self.__api.status_get()
|
||||
except Exception as e:
|
||||
raise AerotechCommunicationError(
|
||||
"Aerotech status request failed",
|
||||
endpoint="status_get",
|
||||
base_url=self.__base,
|
||||
operation="GET",
|
||||
) from e
|
||||
|
||||
def move_home(self, wait:bool=True, incremental:bool=False):
|
||||
if self.__simulated:
|
||||
@@ -91,10 +115,26 @@ class AerotechController(object):
|
||||
return self.position(AEROTECH_HOME, wait=wait, incremental=incremental)
|
||||
|
||||
def home_aerotech(self):
|
||||
return self.__api.home_post()
|
||||
try:
|
||||
return self.__api.home_post()
|
||||
except Exception as e:
|
||||
raise AerotechCommunicationError(
|
||||
"Aerotech home failed",
|
||||
endpoint="home_post",
|
||||
base_url=self.__base,
|
||||
operation="POST",
|
||||
) from e
|
||||
|
||||
def wait_till_done(self, timeout=60):
|
||||
return self.__api.wait_till_done_post(timeout=timeout)
|
||||
try:
|
||||
return self.__api.wait_till_done_post(timeout=timeout)
|
||||
except Exception as e:
|
||||
raise AerotechCommunicationError(
|
||||
"Aerotech wait_till_done failed",
|
||||
endpoint="wait_till_done_post",
|
||||
base_url=self.__base,
|
||||
operation="POST",
|
||||
) from e
|
||||
|
||||
def position(
|
||||
self,
|
||||
@@ -107,7 +147,15 @@ class AerotechController(object):
|
||||
return self.__pos
|
||||
|
||||
payload = self.__make_aerotech_target(target, wait=wait, incremental=incremental)
|
||||
return self.__api.position_post(payload)
|
||||
try:
|
||||
return self.__api.position_post(payload)
|
||||
except Exception as e:
|
||||
raise AerotechCommunicationError(
|
||||
"Aerotech position move failed",
|
||||
endpoint="position_post",
|
||||
base_url=self.__base,
|
||||
operation="POST",
|
||||
) from e
|
||||
|
||||
def rotation_scan(self, rotation_deg: float | int,
|
||||
time_sec: float | int,
|
||||
@@ -122,7 +170,16 @@ class AerotechController(object):
|
||||
if self.__simulated:
|
||||
return payload
|
||||
|
||||
return self.__api.rotation_scan_post(payload)
|
||||
try:
|
||||
return self.__api.rotation_scan_post(payload)
|
||||
except Exception as e:
|
||||
raise AerotechCommunicationError(
|
||||
"Aerotech rotation scan failed",
|
||||
endpoint="rotation_scan_post",
|
||||
base_url=self.__base,
|
||||
operation="POST",
|
||||
) from e
|
||||
|
||||
|
||||
def grid_scan(self,
|
||||
grid_elem_count_y: int,
|
||||
@@ -143,7 +200,15 @@ class AerotechController(object):
|
||||
)
|
||||
if self.__simulated:
|
||||
return payload
|
||||
return self.__api.grid_scan_post(payload)
|
||||
try:
|
||||
return self.__api.grid_scan_post(payload)
|
||||
except Exception as e:
|
||||
raise AerotechCommunicationError(
|
||||
"Aerotech grid scan failed",
|
||||
endpoint="grid_scan_post",
|
||||
base_url=self.__base,
|
||||
operation="POST",
|
||||
) from e
|
||||
|
||||
def screening_scan(self,
|
||||
rotation_deg: float | int,
|
||||
@@ -161,7 +226,15 @@ class AerotechController(object):
|
||||
)
|
||||
if self.__simulated:
|
||||
return payload
|
||||
return self.__api.screening_post(payload)
|
||||
try:
|
||||
return self.__api.screening_post(payload)
|
||||
except Exception as e:
|
||||
raise AerotechCommunicationError(
|
||||
"Aerotech screening scan failed",
|
||||
endpoint="screening_post",
|
||||
base_url=self.__base,
|
||||
operation="POST",
|
||||
) from e
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
@@ -170,3 +243,4 @@ if __name__ == "__main__":
|
||||
controller = AerotechController(beamline)
|
||||
#controller.print_status(colored=True, compact=True)
|
||||
print(controller.get_position())
|
||||
controller.cancel()
|
||||
Reference in new issue
Block a user