224 lines
7.7 KiB
Python
224 lines
7.7 KiB
Python
import types
|
|
import sys
|
|
|
|
from aare.common.coordinate import AerotechCoordinate, Coordinate
|
|
from aare.common.exception_handler import CriticalTellException, MountingFailed
|
|
from aare.common.models import DewarAddress, SampleShortInfo
|
|
from aare.devices.tell_client import TellEventValueEnum
|
|
from aare.daq.operations.mounting.models import MountingContext, MountingResult
|
|
from aare.daq.operations.mounting.service import MountingService
|
|
|
|
|
|
if "jfjoch_client.models.scan_result" not in sys.modules:
|
|
jfjoch_client_mod = types.ModuleType("jfjoch_client")
|
|
jfjoch_client_models_mod = types.ModuleType("jfjoch_client.models")
|
|
jfjoch_client_scan_result_mod = types.ModuleType("jfjoch_client.models.scan_result")
|
|
|
|
jfjoch_client_scan_result_mod.ScanResult = dict
|
|
jfjoch_client_models_mod.scan_result = jfjoch_client_scan_result_mod
|
|
jfjoch_client_mod.models = jfjoch_client_models_mod
|
|
sys.modules["jfjoch_client"] = jfjoch_client_mod
|
|
sys.modules["jfjoch_client.models"] = jfjoch_client_models_mod
|
|
sys.modules["jfjoch_client.models.scan_result"] = jfjoch_client_scan_result_mod
|
|
|
|
|
|
def _make_sample(sample_id: int, name: str) -> SampleShortInfo:
|
|
return SampleShortInfo(
|
|
db_id=sample_id,
|
|
puck_name="puck1",
|
|
dewar_name="dew1",
|
|
sample_name=name,
|
|
run_number=1,
|
|
user="group1",
|
|
pin=sample_id,
|
|
location=DewarAddress(segment="A", pos=1),
|
|
)
|
|
|
|
|
|
def _make_context(previous_sample=None):
|
|
tell = types.SimpleNamespace(
|
|
mount=lambda **kwargs: TellEventValueEnum.SUCCESS,
|
|
unmount=lambda **kwargs: None,
|
|
dry=lambda **kwargs: None,
|
|
check_enable_motion=lambda: None,
|
|
wait_not_busy=lambda timeout=360.0: None,
|
|
set_in_mount_position=lambda value: None,
|
|
)
|
|
|
|
devs = types.SimpleNamespace(
|
|
smargon_move_home=lambda: None,
|
|
aerotech_pos=None,
|
|
tell=tell,
|
|
magnet_position_sensor=types.SimpleNamespace(value=0),
|
|
)
|
|
|
|
streak = {"count": 0}
|
|
|
|
cfg = types.SimpleNamespace(
|
|
current_sample=previous_sample,
|
|
get_mount_failure_streak=lambda: streak["count"],
|
|
increment_mount_failure_streak=lambda: streak.__setitem__("count", streak["count"] + 1) or streak["count"],
|
|
reset_mount_failure_streak=lambda: streak.__setitem__("count", 0),
|
|
record_mount_failure=lambda: streak.__setitem__("count", streak["count"] + 1) or streak["count"],
|
|
record_mount_success=lambda: streak.__setitem__("count", 0),
|
|
)
|
|
|
|
return MountingContext(
|
|
cfg=cfg,
|
|
devs=devs,
|
|
mount_position=AerotechCoordinate(at_mm=Coordinate(x=0, y=0, z=0), omega_deg=0),
|
|
)
|
|
|
|
|
|
def test_execute_mount_success(mock_logger):
|
|
previous_sample = _make_sample(1, "old")
|
|
target_sample = _make_sample(2, "new")
|
|
ctx = _make_context(previous_sample=previous_sample)
|
|
|
|
service = MountingService(context=ctx, logger=mock_logger)
|
|
|
|
result = service.execute(target=target_sample)
|
|
|
|
assert isinstance(result, MountingResult)
|
|
assert result.success is True
|
|
assert result.mounted_sample == target_sample
|
|
assert result.previous_sample == previous_sample
|
|
assert result.did_unmount_previous is True
|
|
assert ctx.cfg.current_sample == target_sample
|
|
|
|
|
|
def test_execute_unmount_success(mock_logger):
|
|
previous_sample = _make_sample(1, "old")
|
|
ctx = _make_context(previous_sample=previous_sample)
|
|
|
|
service = MountingService(context=ctx, logger=mock_logger)
|
|
|
|
result = service.execute(target=None)
|
|
|
|
assert result.success is True
|
|
assert result.mounted_sample is None
|
|
assert result.previous_sample == previous_sample
|
|
assert result.did_unmount_previous is True
|
|
assert ctx.cfg.current_sample is None
|
|
|
|
|
|
def test_execute_mount_returns_failed_result_for_no_pin_in_gripper(mock_logger):
|
|
target_sample = _make_sample(2, "new")
|
|
ctx = _make_context()
|
|
ctx.devs.tell.mount = lambda **kwargs: TellEventValueEnum.NO_PIN_IN_GRIPPER
|
|
|
|
service = MountingService(context=ctx, logger=mock_logger)
|
|
|
|
result = service.execute(target=target_sample)
|
|
|
|
assert result.success is False
|
|
assert isinstance(result.error, MountingFailed)
|
|
assert result.is_error is True
|
|
|
|
|
|
def test_execute_mount_returns_failed_result_for_unhandled_tell_response(mock_logger):
|
|
target_sample = _make_sample(2, "new")
|
|
ctx = _make_context()
|
|
ctx.devs.tell.mount = lambda **kwargs: TellEventValueEnum.UNKNOWN
|
|
|
|
service = MountingService(context=ctx, logger=mock_logger)
|
|
|
|
result = service.execute(target=target_sample)
|
|
|
|
assert result.success is False
|
|
assert isinstance(result.error, CriticalTellException)
|
|
assert result.is_error is True
|
|
|
|
|
|
def test_dry_unmounts_current_sample_before_drying(mock_logger):
|
|
previous_sample = _make_sample(1, "old")
|
|
ctx = _make_context(previous_sample=previous_sample)
|
|
unmount_calls = []
|
|
dry_calls = []
|
|
|
|
ctx.devs.tell.unmount = lambda **kwargs: unmount_calls.append(kwargs)
|
|
ctx.devs.tell.dry = lambda **kwargs: dry_calls.append(kwargs)
|
|
|
|
service = MountingService(context=ctx, logger=mock_logger)
|
|
service.dry(park=True, unmount=True)
|
|
|
|
assert len(unmount_calls) == 1
|
|
assert ctx.cfg.current_sample is None
|
|
assert dry_calls == [{"wait_cold": -1, "wait": True}]
|
|
|
|
|
|
def test_execute_mount_triggers_dry_on_third_consecutive_failure(mock_logger):
|
|
target_sample = _make_sample(2, "new")
|
|
dry_calls = []
|
|
|
|
ctx = _make_context()
|
|
ctx.devs.tell.mount = lambda **kwargs: TellEventValueEnum.NO_PIN_IN_GRIPPER
|
|
ctx.devs.tell.dry = lambda **kwargs: dry_calls.append(kwargs)
|
|
ctx.cfg.increment_mount_failure_streak()
|
|
ctx.cfg.increment_mount_failure_streak()
|
|
|
|
service = MountingService(context=ctx, logger=mock_logger)
|
|
|
|
result = service.execute(target=target_sample)
|
|
|
|
assert result.success is False
|
|
assert isinstance(result.error, MountingFailed)
|
|
assert result.error.critical is False
|
|
assert dry_calls == [{"wait": True}]
|
|
|
|
|
|
def test_execute_mount_stops_automation_on_fifth_consecutive_failure(mock_logger):
|
|
target_sample = _make_sample(2, "new")
|
|
dry_calls = []
|
|
|
|
ctx = _make_context()
|
|
ctx.devs.tell.mount = lambda **kwargs: TellEventValueEnum.NO_PIN_IN_GRIPPER
|
|
ctx.devs.tell.dry = lambda **kwargs: dry_calls.append(kwargs)
|
|
ctx.cfg.increment_mount_failure_streak()
|
|
ctx.cfg.increment_mount_failure_streak()
|
|
ctx.cfg.increment_mount_failure_streak()
|
|
ctx.cfg.increment_mount_failure_streak()
|
|
|
|
service = MountingService(context=ctx, logger=mock_logger)
|
|
|
|
result = service.execute(target=target_sample)
|
|
|
|
assert result.success is False
|
|
assert isinstance(result.error, MountingFailed)
|
|
assert result.error.critical is True
|
|
assert "Mount failed 5 times in a row" in str(result.error)
|
|
assert {"wait_cold": -1, "wait": True} in dry_calls
|
|
|
|
|
|
def test_execute_mount_success_resets_failure_streak(mock_logger):
|
|
previous_sample = _make_sample(1, "old")
|
|
target_sample = _make_sample(2, "new")
|
|
ctx = _make_context(previous_sample=previous_sample)
|
|
ctx.cfg.increment_mount_failure_streak()
|
|
ctx.cfg.increment_mount_failure_streak()
|
|
|
|
service = MountingService(context=ctx, logger=mock_logger)
|
|
|
|
result = service.execute(target=target_sample)
|
|
|
|
assert result.success is True
|
|
assert ctx.cfg.get_mount_failure_streak() == 0
|
|
|
|
|
|
def test_execute_mount_critical_tell_error_does_not_increment_failure_streak(mock_logger):
|
|
target_sample = _make_sample(2, "new")
|
|
ctx = _make_context()
|
|
|
|
def mount(**kwargs):
|
|
raise CriticalTellException("critical tell problem")
|
|
|
|
ctx.devs.tell.mount = mount
|
|
|
|
service = MountingService(context=ctx, logger=mock_logger)
|
|
|
|
result = service.execute(target=target_sample)
|
|
|
|
assert result.success is False
|
|
assert isinstance(result.error, CriticalTellException)
|
|
assert ctx.cfg.get_mount_failure_streak() == 0
|