220 lines
8.4 KiB
Python
220 lines
8.4 KiB
Python
import time
|
|
|
|
from aare.common.exception_handler import (
|
|
CriticalTellException,
|
|
MountingFailed,
|
|
TellCommunicationError,
|
|
UnmountingFailed,
|
|
)
|
|
from aare.devices.tell_client import TellEventValueEnum
|
|
from aare.daq.operations.mounting.models import MountingContext, MountingResult
|
|
|
|
|
|
DRY_AFTER_FAIL_COUNT = 3
|
|
STOP_AFTER_FAIL_COUNT = 5
|
|
|
|
|
|
class MountingService:
|
|
def __init__(self, *, context: MountingContext, logger):
|
|
self.ctx = context
|
|
self.logger = logger
|
|
|
|
def reset_mount_failure_counter(self, reason: str) -> None:
|
|
previous_count = self.ctx.deps.cfg.get_mount_failure_streak()
|
|
if previous_count > 0:
|
|
self.logger.warning(
|
|
f"Resetting mount failure counter from {previous_count} due to: {reason}"
|
|
)
|
|
else:
|
|
self.logger.debug(f"Mount failure counter already clear: {reason}")
|
|
self.ctx.deps.cfg.reset_mount_failure_streak()
|
|
|
|
def _magnet_position_sensor_check(self, timeout: float = 1.0) -> None:
|
|
if self.ctx.deps.devs.magnet_position_sensor.value != 0:
|
|
self.logger.warning(
|
|
"Goniometer is not in position based on magnet position sensor readout"
|
|
)
|
|
for _ in range(round(timeout * 10.0)):
|
|
if self.ctx.deps.devs.magnet_position_sensor.value == 0:
|
|
return
|
|
time.sleep(0.1)
|
|
|
|
self.logger.error(
|
|
"Goniometer didn't reach position based on magnet position sensor readout"
|
|
)
|
|
raise Exception("Goniometer is not in position based on magnet position sensor readout")
|
|
|
|
def _handle_consecutive_mount_failure(self) -> None:
|
|
count = self.ctx.deps.cfg.increment_mount_failure_streak()
|
|
self.logger.warning(f"Consecutive mount failure count is now {count}")
|
|
|
|
if count == DRY_AFTER_FAIL_COUNT:
|
|
self.logger.warning(f"Mount failed {count} times in a row; drying")
|
|
try:
|
|
self.dry(park=False)
|
|
except Exception as e:
|
|
self.logger.exception(f"Failed to dry after mount failure: {e}")
|
|
|
|
if count >= STOP_AFTER_FAIL_COUNT:
|
|
self.logger.error(f"Mount failed {count} times in a row, stopping automation")
|
|
self.logger.error("Unmounting sample and drying")
|
|
try:
|
|
self._unmount_current_sample(timeout=60.0)
|
|
self.dry(park=True)
|
|
except Exception as e:
|
|
self.logger.exception(f"Failed to clean up after repeated mount failure: {e}")
|
|
|
|
raise MountingFailed(
|
|
f"Mount failed {count} times in a row, stopping automation.",
|
|
critical=True,
|
|
)
|
|
|
|
def _mount_handler(self, target) -> None:
|
|
self._prepare_mount_hardware()
|
|
|
|
value = self.ctx.deps.devs.tell.mount(
|
|
address=target.tell_address(),
|
|
force=True,
|
|
auto_unmount=True,
|
|
read_dm=False,
|
|
wait=True,
|
|
timeout=360.0,
|
|
)
|
|
|
|
if isinstance(value, str):
|
|
self.ctx.deps.devs.tell.check_command_ok()
|
|
self.logger.error(f"{self.ctx.deps.devs.tell.get_result(self.ctx.deps.devs.tell._last_cmd_id)}")
|
|
self.logger.error(f"Unexpected string response from Tell mount: {value}")
|
|
raise CriticalTellException(f"Critical error in TELL mount: unexpected response '{value}'")
|
|
|
|
if value is None:
|
|
self.logger.error("Tell mount returned no response (None)")
|
|
raise CriticalTellException("Critical error in TELL mount: no response")
|
|
|
|
safe_val = getattr(value, "value", str(value))
|
|
self.logger.debug(f"Mount response: {safe_val}")
|
|
|
|
if value == TellEventValueEnum.NO_PIN_IN_GRIPPER:
|
|
raise MountingFailed("No Pin in Gripper")
|
|
|
|
if value == TellEventValueEnum.PIN_IS_LOST_GRIPPER:
|
|
self.logger.error("Pin is lost")
|
|
raise MountingFailed("Pin is lost")
|
|
|
|
if value in (
|
|
TellEventValueEnum.ROBOT_CLEAR_AFTER_MOUNT,
|
|
TellEventValueEnum.DRY,
|
|
TellEventValueEnum.COLD,
|
|
TellEventValueEnum.SUCCESS,
|
|
):
|
|
self.logger.info(f"Robot is {safe_val} - freeing beamline for user")
|
|
return
|
|
|
|
message = f"Mount response not handled: {safe_val}"
|
|
self.logger.error(message)
|
|
raise CriticalTellException(f"Critical error in TELL mount: {message}")
|
|
|
|
def _prepare_mount_hardware(self) -> None:
|
|
self.ctx.deps.devs.smargon_move_home()
|
|
self.ctx.deps.devs.aerotech_pos = self.ctx.settings.mount_position
|
|
self._magnet_position_sensor_check(timeout=360.0)
|
|
self.ctx.deps.devs.tell.check_enable_motion()
|
|
self.ctx.deps.devs.tell.wait_not_busy()
|
|
# enable_motion releases the door safety, so the door can only be
|
|
# validated once motion is enabled.
|
|
self.ctx.deps.devs.tell.validate_door_closed()
|
|
self.ctx.deps.devs.tell.set_in_mount_position(True)
|
|
|
|
def _unmount_current_sample(self, timeout: float = 60.0):
|
|
self._prepare_mount_hardware()
|
|
previous_sample = self.ctx.deps.cfg.current_sample
|
|
if previous_sample is not None:
|
|
self.logger.debug(f"Unmounting sample: {previous_sample}")
|
|
self.ctx.deps.devs.tell.unmount(wait=True, timeout=timeout)
|
|
self.ctx.deps.cfg.current_sample = None
|
|
return previous_sample
|
|
|
|
def dry(self, *, park: bool, unmount: bool = False) -> None:
|
|
self.ctx.deps.devs.tell.check_enable_motion()
|
|
self.ctx.deps.devs.tell.wait_not_busy()
|
|
self.ctx.deps.devs.tell.set_in_mount_position(True)
|
|
|
|
if self.ctx.deps.cfg.current_sample and unmount:
|
|
self._prepare_mount_hardware()
|
|
self._unmount_current_sample(timeout=60.0)
|
|
|
|
if park:
|
|
self.ctx.deps.devs.tell.dry(wait_cold=-1, wait=True)
|
|
else:
|
|
self.ctx.deps.devs.tell.dry(wait=True)
|
|
|
|
def execute(self, *, target) -> MountingResult:
|
|
previous_sample = self.ctx.deps.cfg.current_sample
|
|
|
|
try:
|
|
if target is None:
|
|
try:
|
|
unmounted_sample = self._unmount_current_sample(timeout=60.0)
|
|
return MountingResult(
|
|
success=True,
|
|
mounted_sample=None,
|
|
previous_sample=unmounted_sample,
|
|
did_unmount_previous=unmounted_sample is not None,
|
|
)
|
|
except Exception as e:
|
|
raise UnmountingFailed(f"Failed to unmount: {e}") from e
|
|
|
|
log_msg = f"Mounting sample: {target}"
|
|
if previous_sample is not None:
|
|
log_msg = f"Unmounting sample: {previous_sample}\nand {log_msg}"
|
|
self.logger.debug(log_msg)
|
|
|
|
try:
|
|
self._mount_handler(target)
|
|
except (MountingFailed, TellCommunicationError):
|
|
self._handle_consecutive_mount_failure()
|
|
raise
|
|
|
|
self.logger.debug("Mounting succeeded, resetting mount failure counter")
|
|
self.ctx.deps.cfg.reset_mount_failure_streak()
|
|
self.ctx.deps.cfg.current_sample = target
|
|
|
|
return MountingResult(
|
|
success=True,
|
|
mounted_sample=target,
|
|
previous_sample=previous_sample,
|
|
did_unmount_previous=previous_sample is not None,
|
|
)
|
|
|
|
except CriticalTellException as e:
|
|
self.logger.error(f"Critical Tell error: {e}")
|
|
return MountingResult(
|
|
success=False,
|
|
mounted_sample=None,
|
|
previous_sample=previous_sample,
|
|
did_unmount_previous=False,
|
|
error=e,
|
|
comment=str(e),
|
|
)
|
|
|
|
except MountingFailed as e:
|
|
self.logger.error(f"Mount failed: {e}")
|
|
return MountingResult(
|
|
success=False,
|
|
mounted_sample=None,
|
|
previous_sample=previous_sample,
|
|
did_unmount_previous=False,
|
|
error=e,
|
|
comment=str(e),
|
|
)
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Mount failed: {e}")
|
|
return MountingResult(
|
|
success=False,
|
|
mounted_sample=None,
|
|
previous_sample=previous_sample,
|
|
did_unmount_previous=False,
|
|
error=e,
|
|
comment=str(e),
|
|
) |