diff --git a/src/aare/devices/aerotech.py b/src/aare/devices/aerotech.py index 2d236f7e..463087f7 100644 --- a/src/aare/devices/aerotech.py +++ b/src/aare/devices/aerotech.py @@ -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()