chore: update aarecommon dependency to 0.1.3 #95
+1
-1
@@ -6,7 +6,7 @@ readme = "README.md"
|
||||
requires-python = ">=3.11"
|
||||
dependencies = [
|
||||
"uv",
|
||||
"aarecommon>=0.1.2",
|
||||
"aarecommon>=0.1.3",
|
||||
"pydantic==2.11.4",
|
||||
"numpy==2.2.5",
|
||||
"jfjoch_client==1.0.0rc146",
|
||||
|
||||
@@ -1,251 +0,0 @@
|
||||
"""Tests for the AareException hierarchy introduced in Phase 1 of the
|
||||
exception-handling redesign.
|
||||
|
||||
These cover:
|
||||
- class-level ``critical`` default
|
||||
- instance-level override via ``critical=...`` kwarg
|
||||
- correct parent/family relationships so watchers can match by family
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from aarecommon.errors.exception_handler import (
|
||||
AareAuthError,
|
||||
AareDBCommunicationError,
|
||||
AareException,
|
||||
AareUserError,
|
||||
AerotechCommunicationError,
|
||||
AerotechException,
|
||||
AuthenticationException,
|
||||
AutomationError,
|
||||
AutoRasterSampleSkipped,
|
||||
AXCFailed,
|
||||
BeamlineBusyException,
|
||||
BeamlineBusyTimeoutException,
|
||||
BeamlineStateException,
|
||||
BECCommunicationError,
|
||||
BECException,
|
||||
CriticalTellException,
|
||||
DataCollectionException,
|
||||
JFJochCommunicationError,
|
||||
JFJochException,
|
||||
LoopCenteringFailed,
|
||||
MagnetPositionSensorErorr,
|
||||
MaintenanceStateException,
|
||||
ManualMountException,
|
||||
MountingFailed,
|
||||
RasterScanException,
|
||||
SampleException,
|
||||
SmargonCommunicationError,
|
||||
SmargonException,
|
||||
SmartMagnetFaultException,
|
||||
StateTransitionFailed,
|
||||
TellCommandWhileBusyException,
|
||||
TellCommunicationError,
|
||||
TellConnectionException,
|
||||
TellException,
|
||||
TransformationInvalidException,
|
||||
UnmountingFailed,
|
||||
UserRightsException,
|
||||
WarningTellException,
|
||||
)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Class-level criticality defaults
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_aare_exception_class_critical_default_false():
|
||||
assert AareException.critical is False
|
||||
|
||||
|
||||
def test_phase3_classes_have_critical_default_true():
|
||||
for cls in (
|
||||
TellCommunicationError,
|
||||
TellConnectionException,
|
||||
CriticalTellException,
|
||||
SmargonCommunicationError,
|
||||
AerotechCommunicationError,
|
||||
JFJochCommunicationError,
|
||||
BECCommunicationError,
|
||||
StateTransitionFailed,
|
||||
MaintenanceStateException,
|
||||
BeamlineBusyException,
|
||||
BeamlineBusyTimeoutException,
|
||||
MagnetPositionSensorErorr,
|
||||
SmartMagnetFaultException,
|
||||
TransformationInvalidException,
|
||||
):
|
||||
assert cls.critical is True, f"{cls.__name__} should default to critical=True in Phase 3"
|
||||
|
||||
|
||||
def test_other_core_classes_keep_critical_default_false():
|
||||
for cls in (
|
||||
AutomationError,
|
||||
AareUserError,
|
||||
AareAuthError,
|
||||
TellException,
|
||||
SmargonException,
|
||||
AerotechException,
|
||||
JFJochException,
|
||||
BECException,
|
||||
BeamlineStateException,
|
||||
MountingFailed,
|
||||
UnmountingFailed,
|
||||
LoopCenteringFailed,
|
||||
TellCommandWhileBusyException,
|
||||
WarningTellException,
|
||||
AareDBCommunicationError,
|
||||
):
|
||||
assert cls.critical is False, f"{cls.__name__} should remain critical=False"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Instance-level critical override
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_instance_critical_override_true():
|
||||
exc = MountingFailed("mount fail", critical=True)
|
||||
assert exc.critical is True
|
||||
# class default unchanged
|
||||
assert MountingFailed.critical is False
|
||||
|
||||
|
||||
def test_instance_critical_override_false():
|
||||
exc = MountingFailed("mount fail", critical=False)
|
||||
assert exc.critical is False
|
||||
|
||||
|
||||
def test_instance_critical_none_keeps_class_default():
|
||||
# Omitting critical uses class default
|
||||
exc = MountingFailed("mount fail")
|
||||
assert exc.critical is False # class default
|
||||
|
||||
|
||||
def test_aaredb_communication_error_instance_critical_override():
|
||||
exc = AareDBCommunicationError("db down", critical=True, operation="GET")
|
||||
assert exc.critical is True
|
||||
assert exc.operation == "GET"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Routing-root membership
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_automation_errors_are_automation_error():
|
||||
for cls in (
|
||||
LoopCenteringFailed,
|
||||
MountingFailed,
|
||||
UnmountingFailed,
|
||||
AXCFailed,
|
||||
TellCommunicationError,
|
||||
SmargonCommunicationError,
|
||||
AerotechCommunicationError,
|
||||
JFJochCommunicationError,
|
||||
BECCommunicationError,
|
||||
AareDBCommunicationError,
|
||||
BeamlineBusyException,
|
||||
RasterScanException,
|
||||
MagnetPositionSensorErorr,
|
||||
SmartMagnetFaultException,
|
||||
TransformationInvalidException,
|
||||
AutoRasterSampleSkipped,
|
||||
):
|
||||
exc = cls() if cls is not AutoRasterSampleSkipped else cls("skipped")
|
||||
assert isinstance(exc, AutomationError), f"{cls.__name__} should be AutomationError"
|
||||
assert isinstance(exc, AareException), f"{cls.__name__} should be AareException"
|
||||
|
||||
|
||||
def test_auth_errors_are_aare_auth_error():
|
||||
for cls in (AuthenticationException, UserRightsException):
|
||||
exc = cls()
|
||||
assert isinstance(exc, AareAuthError)
|
||||
assert isinstance(exc, AareException)
|
||||
|
||||
|
||||
def test_user_errors_are_aare_user_error():
|
||||
for cls in (ManualMountException, SampleException):
|
||||
exc = cls()
|
||||
assert isinstance(exc, AareUserError)
|
||||
assert isinstance(exc, AareException)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Family-base membership (drives watcher matching)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_tell_family_membership():
|
||||
for cls in (
|
||||
TellCommunicationError,
|
||||
TellConnectionException,
|
||||
CriticalTellException,
|
||||
TellCommandWhileBusyException,
|
||||
WarningTellException,
|
||||
MountingFailed,
|
||||
UnmountingFailed,
|
||||
):
|
||||
exc = cls()
|
||||
assert isinstance(exc, TellException), f"{cls.__name__} should be in Tell family"
|
||||
assert isinstance(exc, AutomationError)
|
||||
|
||||
|
||||
def test_smargon_family_membership():
|
||||
exc = SmargonCommunicationError()
|
||||
assert isinstance(exc, SmargonException)
|
||||
assert isinstance(exc, AutomationError)
|
||||
|
||||
|
||||
def test_aerotech_family_membership():
|
||||
exc = AerotechCommunicationError()
|
||||
assert isinstance(exc, AerotechException)
|
||||
assert isinstance(exc, AutomationError)
|
||||
|
||||
|
||||
def test_jfjoch_family_membership():
|
||||
exc = JFJochCommunicationError()
|
||||
assert isinstance(exc, JFJochException)
|
||||
assert isinstance(exc, AutomationError)
|
||||
|
||||
|
||||
def test_bec_family_membership():
|
||||
exc = BECCommunicationError()
|
||||
assert isinstance(exc, BECException)
|
||||
assert isinstance(exc, AutomationError)
|
||||
|
||||
|
||||
def test_beamline_state_family_membership():
|
||||
for cls in (
|
||||
StateTransitionFailed,
|
||||
MaintenanceStateException,
|
||||
BeamlineBusyException,
|
||||
BeamlineBusyTimeoutException,
|
||||
):
|
||||
exc = cls()
|
||||
assert isinstance(exc, BeamlineStateException)
|
||||
assert isinstance(exc, AutomationError)
|
||||
|
||||
|
||||
def test_data_collection_family_membership():
|
||||
exc = RasterScanException()
|
||||
assert isinstance(exc, DataCollectionException)
|
||||
assert isinstance(exc, AutomationError)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Backward compat: existing __str__/.message contracts preserved
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_message_attribute_preserved():
|
||||
exc = LoopCenteringFailed("custom message")
|
||||
assert exc.message == "custom message"
|
||||
assert str(exc) == "custom message"
|
||||
|
||||
|
||||
def test_default_messages_preserved():
|
||||
assert str(MountingFailed()) == "A sample was not mounted"
|
||||
assert str(UnmountingFailed()) == "A sample was not unmounted"
|
||||
assert str(LoopCenteringFailed()) == "Loop Centering did not detect a sample"
|
||||
@@ -1,84 +0,0 @@
|
||||
from aarecommon.models.aerotech import (
|
||||
AerotechAxisStatus,
|
||||
AerotechRotationScanRequest,
|
||||
AerotechRunEnum,
|
||||
AerotechStatus,
|
||||
AerotechTarget,
|
||||
AxisEnum,
|
||||
TaskEnum,
|
||||
VariableTypeEnum,
|
||||
)
|
||||
|
||||
|
||||
def test_enums():
|
||||
assert TaskEnum.TASK_0.value == 0
|
||||
assert AxisEnum.X.value == "x"
|
||||
assert AerotechRunEnum.START.value == 1
|
||||
assert VariableTypeEnum.REAL.value == 1
|
||||
|
||||
|
||||
def test_aerotech_axis_status():
|
||||
status = AerotechAxisStatus(
|
||||
enabled=True,
|
||||
fault=0,
|
||||
homed=True,
|
||||
is_fault=False,
|
||||
moving=False,
|
||||
position=10.0,
|
||||
status=1,
|
||||
velocity=0.0,
|
||||
)
|
||||
assert status.position == 10.0
|
||||
assert status.enabled is True
|
||||
|
||||
|
||||
def test_aerotech_status_strings():
|
||||
axis_status = AerotechAxisStatus(
|
||||
enabled=True,
|
||||
fault=0,
|
||||
homed=True,
|
||||
is_fault=False,
|
||||
moving=False,
|
||||
position=1.234567,
|
||||
status=1,
|
||||
velocity=0.1,
|
||||
)
|
||||
status = AerotechStatus(state="READY", x=axis_status)
|
||||
|
||||
pretty = status.to_pretty_string()
|
||||
assert "STATE: READY" in pretty
|
||||
assert " X | pos= 1.234567" in pretty
|
||||
assert "Y: unavailable" in pretty
|
||||
|
||||
compact = status.to_compact_string()
|
||||
assert "state=READY" in compact
|
||||
assert "x=1.2346 (H, -)" in compact
|
||||
|
||||
colored = status.to_colored_string()
|
||||
assert "STATE:" in colored
|
||||
assert "READY" in colored
|
||||
assert "pos=" in colored
|
||||
|
||||
assert str(status) == pretty
|
||||
|
||||
|
||||
def test_aerotech_target():
|
||||
target = AerotechTarget(x=10.0, y=20.0)
|
||||
payload = target.to_payload()
|
||||
assert payload == {"x": 10.0, "y": 20.0}
|
||||
assert "z" not in payload
|
||||
|
||||
|
||||
def test_aerotech_rotation_scan_request():
|
||||
request = AerotechRotationScanRequest(
|
||||
rotation_deg=360.0,
|
||||
time_sec=10.0,
|
||||
start_pos_deg=0.0,
|
||||
async_move=True,
|
||||
exp_time_s=0.1,
|
||||
incr_omega_deg=1.0,
|
||||
steps=360,
|
||||
)
|
||||
payload = request.to_payload()
|
||||
assert payload["rotation_deg"] == 360.0
|
||||
assert payload["async"] is True
|
||||
@@ -1,80 +0,0 @@
|
||||
import numpy as np
|
||||
from aarecommon.math.autofocus import focus_measure_blob_size, focus_measure_edges
|
||||
|
||||
|
||||
def test_focus_measure_edges_all_zeros():
|
||||
gray = np.zeros((100, 100), dtype=np.uint8)
|
||||
assert focus_measure_edges(gray) == 0.0
|
||||
|
||||
|
||||
def test_focus_measure_edges_sharp_vs_blurry():
|
||||
# Sharp image (large blocks to survive GaussianBlur)
|
||||
sharp = np.zeros((100, 100), dtype=np.uint8)
|
||||
sharp[:, 0:50] = 255
|
||||
|
||||
# Blurry image (flat)
|
||||
blurry = np.full((100, 100), 128, dtype=np.uint8)
|
||||
|
||||
fm_sharp = focus_measure_edges(sharp, verbose=True)
|
||||
fm_blurry = focus_measure_edges(blurry)
|
||||
|
||||
assert fm_sharp > fm_blurry
|
||||
assert fm_blurry == 0.0
|
||||
|
||||
|
||||
def test_focus_measure_edges_with_mask():
|
||||
gray = np.zeros((100, 100), dtype=np.uint8)
|
||||
gray[40:60, 40:60] = 255
|
||||
|
||||
mask = np.zeros((100, 100), dtype=bool)
|
||||
mask[40:60, 40:60] = True
|
||||
|
||||
fm_with_mask = focus_measure_edges(gray, mask=mask)
|
||||
assert fm_with_mask > 0
|
||||
|
||||
empty_mask = np.zeros((100, 100), dtype=bool)
|
||||
assert focus_measure_edges(gray, mask=empty_mask) == 0.0
|
||||
|
||||
|
||||
def test_focus_measure_edges_verbose(capsys):
|
||||
gray = np.zeros((100, 100), dtype=np.uint8)
|
||||
gray[40:60, 40:60] = 255
|
||||
mask = np.ones((100, 100), dtype=bool)
|
||||
|
||||
focus_measure_edges(gray, mask=mask, verbose=True)
|
||||
captured = capsys.readouterr()
|
||||
assert "focus=" in captured.out
|
||||
|
||||
|
||||
def test_focus_measure_blob_size_all_zeros():
|
||||
gray = np.zeros((100, 100), dtype=np.uint8)
|
||||
assert focus_measure_blob_size(gray) == 0.0
|
||||
|
||||
|
||||
def test_focus_measure_blob_size_sharp_vs_blurry():
|
||||
# Small sharp blob
|
||||
sharp = np.zeros((100, 100), dtype=np.uint8)
|
||||
sharp[50, 50] = 255
|
||||
|
||||
# Larger blurry blob
|
||||
blurry = np.zeros((100, 100), dtype=np.uint8)
|
||||
blurry[45:55, 45:55] = 255
|
||||
|
||||
fm_sharp = focus_measure_blob_size(sharp)
|
||||
fm_blurry = focus_measure_blob_size(blurry)
|
||||
|
||||
assert fm_sharp > fm_blurry
|
||||
|
||||
|
||||
def test_focus_measure_blob_size_with_mask():
|
||||
gray = np.zeros((100, 100), dtype=np.uint8)
|
||||
gray[50, 50] = 255
|
||||
|
||||
mask = np.zeros((100, 100), dtype=bool)
|
||||
mask[50, 50] = True
|
||||
|
||||
fm_with_mask = focus_measure_blob_size(gray, mask=mask)
|
||||
assert fm_with_mask > 0
|
||||
|
||||
empty_mask = np.zeros((100, 100), dtype=bool)
|
||||
assert focus_measure_blob_size(gray, mask=empty_mask) == 0.0
|
||||
@@ -1,26 +0,0 @@
|
||||
import os
|
||||
from unittest.mock import patch
|
||||
|
||||
from aarecommon.config.beamline import mx_beamline
|
||||
from aarecommon.models.beamline import MXBeamline
|
||||
|
||||
|
||||
def test_mx_beamline_default():
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
# If BEAMLINE is not set, it should return SIMULATED
|
||||
assert mx_beamline() == MXBeamline.SIMULATED
|
||||
|
||||
|
||||
def test_mx_beamline_x10sa():
|
||||
with patch.dict(os.environ, {"BEAMLINE": "x10sa"}):
|
||||
assert mx_beamline() == MXBeamline.X10SA
|
||||
|
||||
|
||||
def test_mx_beamline_x06sa():
|
||||
with patch.dict(os.environ, {"BEAMLINE": "X06SA "}):
|
||||
assert mx_beamline() == MXBeamline.X06SA
|
||||
|
||||
|
||||
def test_mx_beamline_invalid():
|
||||
with patch.dict(os.environ, {"BEAMLINE": "INVALID"}):
|
||||
assert mx_beamline() == MXBeamline.SIMULATED
|
||||
@@ -1,149 +0,0 @@
|
||||
import numpy as np
|
||||
import pytest
|
||||
from aarecommon.math.coordinate import (
|
||||
AerotechCoordinate,
|
||||
Coordinate,
|
||||
SmargonCoordinate,
|
||||
positive_coords,
|
||||
)
|
||||
|
||||
|
||||
def test_coordinate_addition():
|
||||
c1 = Coordinate(x=1, y=2, z=3)
|
||||
c2 = Coordinate(x=4, y=5, z=6)
|
||||
c3 = c1 + c2
|
||||
assert c3.x == 5
|
||||
assert c3.y == 7
|
||||
assert c3.z == 9
|
||||
|
||||
|
||||
def test_coordinate_subtraction():
|
||||
c1 = Coordinate(x=5, y=7, z=9)
|
||||
c2 = Coordinate(x=1, y=2, z=3)
|
||||
c3 = c1 - c2
|
||||
assert c3.x == 4
|
||||
assert c3.y == 5
|
||||
assert c3.z == 6
|
||||
|
||||
|
||||
def test_coordinate_multiplication_scalar():
|
||||
c1 = Coordinate(x=1, y=2, z=3)
|
||||
c2 = c1 * 2.0
|
||||
assert c2.x == 2.0
|
||||
assert c2.y == 4.0
|
||||
assert c2.z == 6.0
|
||||
|
||||
|
||||
def test_coordinate_dot_product():
|
||||
c1 = Coordinate(x=1, y=2, z=3)
|
||||
c2 = Coordinate(x=4, y=5, z=6)
|
||||
dot = c1 * c2
|
||||
assert dot == (1 * 4 + 2 * 5 + 3 * 6)
|
||||
|
||||
|
||||
def test_coordinate_division():
|
||||
c1 = Coordinate(x=2, y=4, z=6)
|
||||
c2 = c1 / 2.0
|
||||
assert c2.x == 1.0
|
||||
assert c2.y == 2.0
|
||||
assert c2.z == 3.0
|
||||
|
||||
|
||||
def test_coordinate_division_by_zero():
|
||||
c1 = Coordinate(x=2, y=4, z=6)
|
||||
with pytest.raises(ValueError, match="Cannot divide by zero"):
|
||||
_ = c1 / 0.0
|
||||
|
||||
|
||||
def test_coordinate_normalize():
|
||||
c1 = Coordinate(x=3, y=0, z=4)
|
||||
c2 = c1.normalize()
|
||||
assert c2.x == 0.6
|
||||
assert c2.y == 0.0
|
||||
assert c2.z == 0.8
|
||||
|
||||
|
||||
def test_coordinate_normalize_zero():
|
||||
c1 = Coordinate(x=0, y=0, z=0)
|
||||
with pytest.raises(ValueError, match="Cannot normalize a zero-magnitude vector."):
|
||||
c1.normalize()
|
||||
|
||||
|
||||
def test_coordinate_rotate_x():
|
||||
c1 = Coordinate(x=1, y=1, z=0)
|
||||
# Rotate 90 degrees around X. (1, 1, 0) -> (1, 0, 1)
|
||||
c2 = c1.rotate(90, "x")
|
||||
assert np.isclose(c2.x, 1)
|
||||
assert np.isclose(c2.y, 0)
|
||||
assert np.isclose(c2.z, 1)
|
||||
|
||||
|
||||
def test_coordinate_rotate_y():
|
||||
c1 = Coordinate(x=1, y=0, z=1)
|
||||
# Rotate 90 degrees around Y. (1, 0, 1) -> (1, 0, -1)
|
||||
c2 = c1.rotate(90, "y")
|
||||
assert np.isclose(c2.x, 1)
|
||||
assert np.isclose(c2.y, 0)
|
||||
assert np.isclose(c2.z, -1)
|
||||
|
||||
|
||||
def test_coordinate_rotate_z():
|
||||
c1 = Coordinate(x=1, y=0, z=0)
|
||||
# Rotate 90 degrees around Z. (1, 0, 0) -> (0, 1, 0)
|
||||
c2 = c1.rotate(90, "z")
|
||||
assert np.isclose(c2.x, 0)
|
||||
assert np.isclose(c2.y, 1)
|
||||
assert np.isclose(c2.z, 0)
|
||||
|
||||
|
||||
def test_coordinate_rotate_invalid_axis():
|
||||
c1 = Coordinate(x=1, y=1, z=1)
|
||||
with pytest.raises(ValueError, match="Invalid axis"):
|
||||
c1.rotate(90, "w")
|
||||
|
||||
|
||||
def test_smargon_coordinate_equality():
|
||||
s1 = SmargonCoordinate(sh_mm=Coordinate(x=1, y=1, z=1), phi_deg=10, chi_deg=20)
|
||||
s2 = SmargonCoordinate(sh_mm=Coordinate(x=1.05, y=0.95, z=1.01), phi_deg=10.05, chi_deg=19.95)
|
||||
assert s1 == s2
|
||||
|
||||
s3 = SmargonCoordinate(sh_mm=Coordinate(x=2, y=1, z=1), phi_deg=10, chi_deg=20)
|
||||
assert s1 != s3
|
||||
|
||||
|
||||
def test_aerotech_coordinate_equality():
|
||||
a1 = AerotechCoordinate(at_mm=Coordinate(x=1, y=1, z=1), omega_deg=10)
|
||||
a2 = AerotechCoordinate(at_mm=Coordinate(x=1.005, y=0.995, z=1.001), omega_deg=10.005)
|
||||
assert a1 == a2
|
||||
|
||||
a3 = AerotechCoordinate(at_mm=Coordinate(x=1.1, y=1, z=1), omega_deg=10)
|
||||
assert a1 != a3
|
||||
|
||||
|
||||
def test_positive_coords():
|
||||
c1 = Coordinate(x=1, y=1, z=1)
|
||||
assert positive_coords(c1) == c1
|
||||
|
||||
with pytest.raises(ValueError, match="Coordinates must be positive"):
|
||||
positive_coords(Coordinate(x=-1, y=1, z=1))
|
||||
|
||||
with pytest.raises(ValueError, match="Coordinates must be positive"):
|
||||
positive_coords(Coordinate(x=1, y=-1, z=1))
|
||||
|
||||
|
||||
def test_coordinate_unsupported_ops():
|
||||
c1 = Coordinate(x=1, y=1, z=1)
|
||||
assert c1.__add__(1) == NotImplemented
|
||||
assert c1.__sub__(1) == NotImplemented
|
||||
assert c1.__mul__("string") == NotImplemented
|
||||
assert c1.__truediv__("string") == NotImplemented
|
||||
|
||||
|
||||
def test_smargon_coordinate_eq_not_implemented():
|
||||
s1 = SmargonCoordinate(sh_mm=Coordinate(x=1, y=1, z=1), phi_deg=10, chi_deg=20)
|
||||
assert s1.__eq__(1) == NotImplemented
|
||||
|
||||
|
||||
def test_aerotech_coordinate_eq_not_implemented():
|
||||
a1 = AerotechCoordinate(at_mm=Coordinate(x=1, y=1, z=1), omega_deg=10)
|
||||
assert a1.__eq__(1) == NotImplemented
|
||||
@@ -1,72 +0,0 @@
|
||||
import pytest
|
||||
from aarecommon.models.models import DataCollectionParameters
|
||||
from pydantic import ValidationError
|
||||
|
||||
|
||||
def test_directory_defaults_when_missing():
|
||||
params = DataCollectionParameters()
|
||||
assert params.directory is None
|
||||
|
||||
|
||||
def test_directory_blank_defaults_to_macro_path():
|
||||
params = DataCollectionParameters(directory="")
|
||||
assert params.directory == "{date}/{prefix}"
|
||||
|
||||
|
||||
def test_directory_spaces_are_replaced():
|
||||
params = DataCollectionParameters(directory="my folder/run 1")
|
||||
assert params.directory == "my_folder/run_1"
|
||||
|
||||
|
||||
def test_directory_rejects_invalid_characters():
|
||||
with pytest.raises(ValidationError):
|
||||
DataCollectionParameters(directory="bad|path")
|
||||
|
||||
|
||||
def test_exposure_accepts_values_above_1_after_refactor():
|
||||
params = DataCollectionParameters(exposure=1.5)
|
||||
assert params.exposure == 1.5
|
||||
|
||||
|
||||
def test_cloud_blank_defaults_to_true():
|
||||
params = DataCollectionParameters(cloud="")
|
||||
assert params.cloud is True
|
||||
|
||||
|
||||
def test_directory_accepts_valid_macros():
|
||||
params = DataCollectionParameters(directory="{date}/{prefix}/run")
|
||||
assert params.directory == "{date}/{prefix}/run"
|
||||
|
||||
|
||||
def test_aperture_accepts_float_string():
|
||||
params = DataCollectionParameters(aperture="2.0")
|
||||
assert params.aperture == 2
|
||||
|
||||
|
||||
def test_processingpipeline_accepts_unknown_value_after_refactor():
|
||||
params = DataCollectionParameters(processingpipeline="xia2")
|
||||
assert params.processingpipeline == "xia2"
|
||||
|
||||
|
||||
def test_datacollectionparameters_accepts_legacy_aliases():
|
||||
params = DataCollectionParameters(
|
||||
totalrange=180, cellparameters="10 20 30 90 90 120", userresolution=1.4
|
||||
)
|
||||
assert params.totalangle == 180
|
||||
assert params.unitcell == "10 20 30 90 90 120"
|
||||
assert params.processingresolution == 1.4
|
||||
|
||||
|
||||
def test_datacollectionparameters_accepts_new_fields():
|
||||
params = DataCollectionParameters(
|
||||
totalangle=90,
|
||||
unitcell="11,22,33,90,90,120",
|
||||
processingresolution=1.2,
|
||||
pdbmodel="model.pdb",
|
||||
cloud=False,
|
||||
)
|
||||
assert params.totalangle == 90
|
||||
assert params.unitcell == "11,22,33,90,90,120"
|
||||
assert params.processingresolution == 1.2
|
||||
assert params.pdbmodel == "model.pdb"
|
||||
assert params.cloud is False
|
||||
@@ -1,66 +0,0 @@
|
||||
import pytest
|
||||
from aarecommon.math.diffraction_geometry import DiffractionGeometry
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_dg():
|
||||
return DiffractionGeometry(
|
||||
energy_keV=12.4,
|
||||
dtz_mm=100.0,
|
||||
pixel_size_mm=0.172,
|
||||
beam_center_pxl=(1000.0, 1000.0),
|
||||
detector_size_pxl=(2000, 2000),
|
||||
detector_description="Eiger 16M",
|
||||
detector_serial_number="E-123",
|
||||
poni_rot1_rad=0.0,
|
||||
poni_rot2_rad=0.0,
|
||||
)
|
||||
|
||||
|
||||
def test_detector_max_radius_pxl(sample_dg):
|
||||
# center (1000, 1000), size (2000, 2000)
|
||||
# x0 = 2000-1000 = 1000
|
||||
# x1 = 1000
|
||||
# y0 = 2000-1000 = 1000
|
||||
# y1 = 1000
|
||||
# max = 1000
|
||||
assert sample_dg.detector_max_radius_pxl == 1000.0
|
||||
|
||||
|
||||
def test_detector_radius_mm(sample_dg):
|
||||
# 1000 * 0.172 = 172.0
|
||||
assert sample_dg.detector_radius_mm == 172.0
|
||||
|
||||
|
||||
def test_wavelength_angstrom(sample_dg):
|
||||
# 12.398 / 12.4 = 0.9998387...
|
||||
assert sample_dg.wavelength_angstrom == pytest.approx(0.9998387)
|
||||
|
||||
|
||||
def test_resolution_angstrom(sample_dg):
|
||||
# dtz = 100
|
||||
# radius = 172
|
||||
# theta = atan(172/100) * 0.5 = atan(1.72) * 0.5 = 1.044 * 0.5 = 0.522 rad
|
||||
# res = 0.9998 / (2 * sin(0.522)) = 0.9998 / (2 * 0.498) = 1.003
|
||||
res = sample_dg.resolution_angstrom(100.0)
|
||||
assert res > 0
|
||||
assert res == pytest.approx(1.002469, abs=1e-5)
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
sample_dg.resolution_angstrom(0)
|
||||
|
||||
|
||||
def test_max_resolution_angstrom(sample_dg):
|
||||
assert sample_dg.max_resolution_angstrom == sample_dg.resolution_angstrom(sample_dg.dtz_mm)
|
||||
|
||||
|
||||
def test_calc_dtz_mm(sample_dg):
|
||||
res = sample_dg.resolution_angstrom(100.0)
|
||||
dtz = sample_dg.calc_dtz_mm(res)
|
||||
assert dtz == pytest.approx(100.0)
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
sample_dg.calc_dtz_mm(-1)
|
||||
|
||||
# test x >= 1.0 case: wavelength / (2*res) >= 1.0 -> res <= wavelength / 2
|
||||
assert sample_dg.calc_dtz_mm(0.0001) == 0.0
|
||||
@@ -1,116 +0,0 @@
|
||||
from aarecommon.errors.codes import (
|
||||
AareErrorCode,
|
||||
AuthErrorCode,
|
||||
code_for_exception_class,
|
||||
error_code_help,
|
||||
export_error_codes,
|
||||
export_error_codes_grouped,
|
||||
)
|
||||
|
||||
|
||||
def test_error_code_help_returns_string():
|
||||
help_text = error_code_help(AuthErrorCode.AUTHENTICATION_FAILED)
|
||||
assert isinstance(help_text, str)
|
||||
assert len(help_text) > 0
|
||||
|
||||
|
||||
def test_error_code_help_unknown_code():
|
||||
help_text = error_code_help("UNKNOWN_CODE")
|
||||
assert help_text is None
|
||||
|
||||
|
||||
def test_export_error_codes_contains_known_codes():
|
||||
exported = export_error_codes()
|
||||
assert AuthErrorCode.AUTHENTICATION_FAILED.name in exported
|
||||
|
||||
|
||||
def test_code_for_exception_class_basic():
|
||||
assert code_for_exception_class("TellCommunicationError") == "TELL_COMMUNICATION_ERROR"
|
||||
assert code_for_exception_class("MountingFailed") == "MOUNTING_FAILED"
|
||||
assert code_for_exception_class("LoopCenteringFailed") == "LOOP_CENTERING_FAILED"
|
||||
|
||||
|
||||
def test_code_for_exception_class_acronyms():
|
||||
assert code_for_exception_class("AXCFailed") == "AXC_FAILED"
|
||||
assert code_for_exception_class("AareDBCommunicationError") == "AARE_DB_COMMUNICATION_ERROR"
|
||||
assert code_for_exception_class("JFJochCommunicationError") == "JF_JOCH_COMMUNICATION_ERROR"
|
||||
|
||||
|
||||
def test_code_for_each_concrete_exception_is_in_aare_error_code_enum():
|
||||
"""Every code we'd produce from the rebuilt hierarchy must exist in the
|
||||
AareErrorCode enum -- otherwise clients have no symbol to branch on."""
|
||||
from aarecommon.errors.exception_handler import (
|
||||
AareDBCommunicationError,
|
||||
AerotechCommunicationError,
|
||||
AuthenticationException,
|
||||
AutoRasterSampleSkipped,
|
||||
AXCFailed,
|
||||
BeamlineBusyException,
|
||||
BeamlineBusyTimeoutException,
|
||||
BECCommunicationError,
|
||||
CriticalTellException,
|
||||
DataCollectionException,
|
||||
JFJochCommunicationError,
|
||||
LoopCenteringFailed,
|
||||
MagnetPositionSensorErorr,
|
||||
MaintenanceStateException,
|
||||
ManualMountException,
|
||||
MountingFailed,
|
||||
RasterScanException,
|
||||
SampleException,
|
||||
SmargonCommunicationError,
|
||||
SmartMagnetFaultException,
|
||||
StateTransitionFailed,
|
||||
TellCommandWhileBusyException,
|
||||
TellCommunicationError,
|
||||
TellConnectionException,
|
||||
TransformationInvalidException,
|
||||
UnmountingFailed,
|
||||
UserRightsException,
|
||||
WarningTellException,
|
||||
)
|
||||
|
||||
valid = {c.value for c in AareErrorCode}
|
||||
classes = [
|
||||
TellCommunicationError,
|
||||
TellConnectionException,
|
||||
CriticalTellException,
|
||||
WarningTellException,
|
||||
TellCommandWhileBusyException,
|
||||
MountingFailed,
|
||||
UnmountingFailed,
|
||||
SmargonCommunicationError,
|
||||
AerotechCommunicationError,
|
||||
JFJochCommunicationError,
|
||||
BECCommunicationError,
|
||||
AareDBCommunicationError,
|
||||
StateTransitionFailed,
|
||||
MaintenanceStateException,
|
||||
BeamlineBusyException,
|
||||
BeamlineBusyTimeoutException,
|
||||
DataCollectionException,
|
||||
RasterScanException,
|
||||
LoopCenteringFailed,
|
||||
AXCFailed,
|
||||
AutoRasterSampleSkipped,
|
||||
TransformationInvalidException,
|
||||
MagnetPositionSensorErorr,
|
||||
SmartMagnetFaultException,
|
||||
ManualMountException,
|
||||
SampleException,
|
||||
AuthenticationException,
|
||||
UserRightsException,
|
||||
]
|
||||
missing = []
|
||||
for cls in classes:
|
||||
code = code_for_exception_class(cls.__name__)
|
||||
if code not in valid:
|
||||
missing.append((cls.__name__, code))
|
||||
assert not missing, f"Codes missing from AareErrorCode: {missing}"
|
||||
|
||||
|
||||
def test_export_error_codes_grouped_includes_aare_error_code():
|
||||
exported = export_error_codes_grouped()
|
||||
assert "AareErrorCode" in exported
|
||||
assert "AuthErrorCode" in exported
|
||||
assert "TELL_COMMUNICATION_ERROR" in exported["AareErrorCode"]
|
||||
@@ -1,194 +0,0 @@
|
||||
import pytest
|
||||
from aarecommon.errors.exception_handler import (
|
||||
AareDBCommunicationError,
|
||||
AerotechCommunicationError,
|
||||
AuthenticationException,
|
||||
AuthErrorCode,
|
||||
AXCFailed,
|
||||
BeamlineBusyException,
|
||||
CriticalTellException,
|
||||
DataCollectionException,
|
||||
JFJochCommunicationError,
|
||||
LoopCenteringFailed,
|
||||
MagnetPositionSensorErorr,
|
||||
ManualMountException,
|
||||
MountingFailed,
|
||||
RasterScanException,
|
||||
SampleException,
|
||||
SmargonCommunicationError,
|
||||
SmartMagnetFaultException,
|
||||
TellCommandWhileBusyException,
|
||||
TellCommunicationError,
|
||||
TellConnectionException,
|
||||
TransformationInvalidException,
|
||||
UnmountingFailed,
|
||||
UserRightsException,
|
||||
WarningTellException,
|
||||
)
|
||||
|
||||
|
||||
def test_data_collection_exception_message():
|
||||
exc = DataCollectionException("Custom error")
|
||||
assert str(exc) == "Custom error"
|
||||
|
||||
|
||||
def test_data_collection_exception_default_message():
|
||||
exc = DataCollectionException()
|
||||
assert str(exc) == "Data collection failed"
|
||||
|
||||
|
||||
def test_authentication_exception_properties():
|
||||
exc = AuthenticationException("Failed", status_code=403, code=AuthErrorCode.FORBIDDEN)
|
||||
assert exc.status_code == 403
|
||||
assert exc.code == AuthErrorCode.FORBIDDEN
|
||||
assert "Failed" in str(exc)
|
||||
|
||||
|
||||
def test_tell_communication_error_str():
|
||||
exc = TellCommunicationError("Timeout", endpoint="/state", operation="GET")
|
||||
assert str(exc) == "Timeout"
|
||||
assert exc.endpoint == "/state"
|
||||
assert exc.operation == "GET"
|
||||
|
||||
|
||||
def test_transformation_invalid_exception():
|
||||
exc = TransformationInvalidException()
|
||||
assert "Transformation is not implemented" in str(exc)
|
||||
|
||||
|
||||
def test_raster_scan_exception():
|
||||
exc = RasterScanException("Raster failed")
|
||||
assert "Raster failed" in str(exc)
|
||||
|
||||
|
||||
def test_loop_centering_failed():
|
||||
exc = LoopCenteringFailed()
|
||||
assert "Loop Centering did not detect a sample" in str(exc)
|
||||
|
||||
|
||||
def test_unmounting_failed():
|
||||
exc = UnmountingFailed()
|
||||
assert "A sample was not unmounted" in str(exc)
|
||||
|
||||
|
||||
def test_mounting_failed():
|
||||
exc = MountingFailed()
|
||||
assert "A sample was not mounted" in str(exc)
|
||||
|
||||
|
||||
def test_manual_mount_exception():
|
||||
exc = ManualMountException()
|
||||
assert "Manual mounting failed" in str(exc)
|
||||
|
||||
|
||||
def test_smart_magnet_fault_exception():
|
||||
exc = SmartMagnetFaultException()
|
||||
assert "Smart magnet fault" in str(exc)
|
||||
|
||||
|
||||
def test_tell_command_while_busy_exception():
|
||||
exc = TellCommandWhileBusyException()
|
||||
assert "Tell is busy" in str(exc)
|
||||
|
||||
|
||||
def test_tell_connection_exception():
|
||||
exc = TellConnectionException()
|
||||
assert "Lost connection to Tell" in str(exc)
|
||||
|
||||
|
||||
def test_warning_tell_exception():
|
||||
exc = WarningTellException()
|
||||
assert "Warning error in TELL" in str(exc)
|
||||
|
||||
|
||||
def test_critical_tell_exception():
|
||||
exc = CriticalTellException()
|
||||
assert "Critical error in TELL" in str(exc)
|
||||
|
||||
|
||||
def test_axc_failed():
|
||||
exc = AXCFailed()
|
||||
assert "Auto X-ray centering failed" in str(exc)
|
||||
|
||||
|
||||
def test_beamline_busy_exception():
|
||||
exc = BeamlineBusyException()
|
||||
assert "Beamline is in busy state" in str(exc)
|
||||
|
||||
|
||||
def test_sample_exception():
|
||||
exc = SampleException()
|
||||
assert "Sample not found" in str(exc)
|
||||
|
||||
|
||||
def test_user_rights_exception():
|
||||
exc = UserRightsException(code=AuthErrorCode.NOT_STAFF)
|
||||
assert exc.status_code == 403
|
||||
assert exc.code == AuthErrorCode.NOT_STAFF
|
||||
assert "User does not have rights" in str(exc)
|
||||
|
||||
|
||||
def test_smargon_communication_error():
|
||||
exc = SmargonCommunicationError("Conn error", endpoint="/move", status_code=500)
|
||||
assert str(exc) == "Conn error"
|
||||
assert exc.endpoint == "/move"
|
||||
assert exc.status_code == 500
|
||||
|
||||
|
||||
def test_jfjoch_communication_error():
|
||||
exc = JFJochCommunicationError("JFJoch error", operation="POST")
|
||||
assert str(exc) == "JFJoch error"
|
||||
assert exc.operation == "POST"
|
||||
|
||||
|
||||
def test_aaredb_communication_error():
|
||||
exc = AareDBCommunicationError("DB error")
|
||||
assert "DB error" in str(exc)
|
||||
|
||||
|
||||
def test_aerotech_communication_error():
|
||||
exc = AerotechCommunicationError("Aerotech error")
|
||||
assert "Aerotech error" in str(exc)
|
||||
|
||||
|
||||
def test_magnet_position_sensor_error():
|
||||
exc = MagnetPositionSensorErorr()
|
||||
assert "Magnet position sensor error" in str(exc)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"factory",
|
||||
[
|
||||
lambda: TransformationInvalidException("t"),
|
||||
lambda: LoopCenteringFailed("lc"),
|
||||
lambda: UnmountingFailed("un"),
|
||||
lambda: MountingFailed("mt"),
|
||||
lambda: ManualMountException("mm"),
|
||||
lambda: SmartMagnetFaultException("sm"),
|
||||
lambda: TellCommandWhileBusyException("tb"),
|
||||
lambda: TellConnectionException("tc"),
|
||||
lambda: WarningTellException("wt"),
|
||||
lambda: CriticalTellException("ct"),
|
||||
lambda: AXCFailed("ax"),
|
||||
lambda: BeamlineBusyException("bb"),
|
||||
lambda: SampleException("se"),
|
||||
lambda: AuthenticationException("ae"),
|
||||
lambda: TellCommunicationError("tce", endpoint="/x", operation="GET"),
|
||||
lambda: SmargonCommunicationError("sce", endpoint="/m", status_code=500),
|
||||
lambda: JFJochCommunicationError("jce", operation="POST"),
|
||||
lambda: AareDBCommunicationError("dbce"),
|
||||
lambda: AerotechCommunicationError("ace"),
|
||||
lambda: MagnetPositionSensorErorr(),
|
||||
],
|
||||
)
|
||||
def test_exception_construction_emits_no_logs(factory, caplog):
|
||||
"""Constructing an Aare exception must not emit log records.
|
||||
|
||||
Logging is the responsibility of the catch site (server handler,
|
||||
best-effort wrapper, or callsite that decides to swallow). Emitting
|
||||
on construction double-logs and pollutes the WARNING-vs-ERROR split
|
||||
for best-effort flows."""
|
||||
caplog.clear()
|
||||
with caplog.at_level("DEBUG"):
|
||||
factory()
|
||||
assert caplog.records == []
|
||||
@@ -1,166 +0,0 @@
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
from aarecommon.math.find_xtal import (
|
||||
compute_crystal_score_array,
|
||||
create_quality_filtered_array,
|
||||
get_best_b_factor,
|
||||
get_best_res,
|
||||
get_xtal_size,
|
||||
has_sufficient_low_res_spots,
|
||||
identify_crystal_raster,
|
||||
raster_centre_of_mass,
|
||||
raster_highest_score,
|
||||
rebuild_array_from_scan_results,
|
||||
)
|
||||
from aarecommon.models.models import Coordinate, CrystalSize
|
||||
from aarecommon.models.raster_grid import CenterOfMassModel, RasterGridRequest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_raster_results():
|
||||
results = []
|
||||
for i in range(9):
|
||||
res = MagicMock()
|
||||
res.nx = i % 3
|
||||
res.ny = i // 3
|
||||
res.number = i
|
||||
res.spots_low_res = float(i)
|
||||
res.spots = i + 10
|
||||
res.spots_ice = 1
|
||||
res.index = False
|
||||
res.efficiency = 1.0
|
||||
res.bkg = 5.0
|
||||
res.b = 20.0 + i
|
||||
res.res = 2.0 - i * 0.1
|
||||
results.append(res)
|
||||
return results
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def raster_request():
|
||||
return RasterGridRequest(
|
||||
exp_time_s=0.1, n_x=3, n_y=3, grid_size_mm=Coordinate(x=0.02, y=0.02), smargon_top_left=None
|
||||
)
|
||||
|
||||
|
||||
def test_identify_crystal_raster(mock_raster_results, raster_request):
|
||||
mock_result = MagicMock()
|
||||
mock_result.images = mock_raster_results
|
||||
|
||||
com = identify_crystal_raster(mock_result, raster_request)
|
||||
assert isinstance(com, CenterOfMassModel)
|
||||
# The max spots_low_res is 8.0 at (2, 2)
|
||||
assert com.n_x == 2
|
||||
assert com.n_y == 2
|
||||
assert com.max_image == 8
|
||||
|
||||
|
||||
def test_rebuild_array_from_scan_results(mock_raster_results):
|
||||
arr = rebuild_array_from_scan_results(mock_raster_results, "spots_low_res", array_shape=(3, 3))
|
||||
assert arr.shape == (3, 3)
|
||||
assert arr[0, 0] == 0.0
|
||||
assert arr[2, 2] == 8.0
|
||||
|
||||
|
||||
def test_create_quality_filtered_array(mock_raster_results):
|
||||
# Testing create_quality_filtered_array with a more lenient filter
|
||||
arr = create_quality_filtered_array(
|
||||
mock_raster_results, "spots", min_low_res_spots=0.0, min_background=0.0, array_shape=(3, 3)
|
||||
)
|
||||
# i=8 has nx=2, ny=2 -> row=2, col=2
|
||||
assert arr[2, 2] == 18.0
|
||||
|
||||
|
||||
def test_get_xtal_size(raster_request):
|
||||
# 3x3 array where only center is 1
|
||||
arr = np.zeros((3, 3))
|
||||
arr[1, 1] = 1
|
||||
|
||||
size = CrystalSize(x=0, y=0, z=0)
|
||||
new_size = get_xtal_size(size, arr, raster_request)
|
||||
# row_max-row_min = 0, so size 0? Wait, it should probably be at least 1 grid unit if present.
|
||||
# The code does (row_max - row_min) * grid_size_mm.x * 1000
|
||||
# If row_min=1, row_max=1, then size is 0.
|
||||
# This might be a bug in find_xtal.py if it doesn't account for the pixel itself.
|
||||
assert new_size.x == 0
|
||||
|
||||
|
||||
def test_get_best_b_factor(mock_raster_results):
|
||||
assert get_best_b_factor(mock_raster_results) == 20.0
|
||||
assert get_best_b_factor([]) is None
|
||||
|
||||
|
||||
def test_get_best_res(mock_raster_results):
|
||||
# min res. i=8 -> res = 2.0 - 0.8 = 1.2
|
||||
assert pytest.approx(get_best_res(mock_raster_results)) == 1.2
|
||||
assert get_best_res([]) is None
|
||||
|
||||
|
||||
def test_raster_centre_of_mass(mock_raster_results):
|
||||
arr = np.zeros((3, 3))
|
||||
arr[1, 1] = 10.0
|
||||
com = raster_centre_of_mass(arr, mock_raster_results)
|
||||
assert com.n_x == 1.0
|
||||
assert com.n_y == 1.0
|
||||
|
||||
|
||||
def test_has_sufficient_low_res_spots():
|
||||
arr = np.array([[1, 2], [3, 4]])
|
||||
assert has_sufficient_low_res_spots(arr, 3.0) is True
|
||||
assert has_sufficient_low_res_spots(arr, 5.0) is False
|
||||
assert has_sufficient_low_res_spots(None, 1.0) is False
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_score_results():
|
||||
# 2x2 grid; cell (1, 1) is the strongest crystal across bkg, spots_low_res
|
||||
# and spots_indexed (the three inputs to compute_crystal_score_array).
|
||||
data = [
|
||||
# nx, ny, bkg, spots_low_res, spots_indexed
|
||||
(0, 0, 1.0, 1.0, 0.0),
|
||||
(1, 0, 2.0, 2.0, 0.0),
|
||||
(0, 1, 2.0, 2.0, 0.0),
|
||||
(1, 1, 10.0, 20.0, 5.0),
|
||||
]
|
||||
results = []
|
||||
for i, (nx, ny, bkg, low, idx) in enumerate(data):
|
||||
res = MagicMock()
|
||||
res.nx, res.ny, res.number = nx, ny, i
|
||||
res.bkg, res.spots_low_res, res.spots_indexed = bkg, low, idx
|
||||
results.append(res)
|
||||
return results
|
||||
|
||||
|
||||
def test_compute_crystal_score_array(mock_score_results):
|
||||
score = compute_crystal_score_array(mock_score_results)
|
||||
assert score.shape == (2, 2)
|
||||
# weights sum to 1.0 and each input is min-max normalised to [0, 100]
|
||||
assert score.min() >= 0.0 and score.max() <= 100.0
|
||||
# (1, 1) is max in all three inputs -> 100; weak corner (0, 0) is min in all -> 0
|
||||
assert score[1, 1] == pytest.approx(100.0)
|
||||
assert score[0, 0] == pytest.approx(0.0)
|
||||
assert np.unravel_index(np.argmax(score), score.shape) == (1, 1)
|
||||
|
||||
|
||||
def test_compute_crystal_score_array_weights():
|
||||
# Cell A is the sole max in spots_low_res (weight 0.55); cell B is the sole
|
||||
# max in spots_indexed (weight 0.20). A must outscore B.
|
||||
def _cell(nx, n, low, idx):
|
||||
r = MagicMock()
|
||||
r.nx, r.ny, r.number = nx, 0, n
|
||||
r.bkg, r.spots_low_res, r.spots_indexed = 0.0, low, idx
|
||||
return r
|
||||
|
||||
score = compute_crystal_score_array([_cell(0, 0, 10.0, 0.0), _cell(1, 1, 0.0, 10.0)])
|
||||
assert score[0, 0] > score[1, 0]
|
||||
assert score[0, 0] == pytest.approx(55.0)
|
||||
assert score[1, 0] == pytest.approx(20.0)
|
||||
|
||||
|
||||
def test_raster_highest_score(mock_score_results):
|
||||
com = raster_highest_score(mock_score_results)
|
||||
assert com.n_x == 1.0
|
||||
assert com.n_y == 1.0
|
||||
assert com.max_image == 3
|
||||
@@ -1,60 +0,0 @@
|
||||
import logging
|
||||
import time
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
from aarecommon.config.logger_events import log_timing, merge_log_context
|
||||
|
||||
|
||||
def test_log_timing_success():
|
||||
logger = MagicMock(spec=logging.Logger)
|
||||
|
||||
@log_timing(logger, message_prefix="Test", level=logging.INFO)
|
||||
def sample_func(x):
|
||||
time.sleep(0.01)
|
||||
return x * 2
|
||||
|
||||
result = sample_func(21)
|
||||
|
||||
assert result == 42
|
||||
assert logger.log.call_count == 2
|
||||
# First call: Starting
|
||||
logger.log.assert_any_call(logging.INFO, "Test: Starting sample_func")
|
||||
# Second call: Finished (check that it contains the message and duration_s in extra)
|
||||
args, kwargs = logger.log.call_args_list[1]
|
||||
assert args[0] == logging.INFO
|
||||
assert "Finished sample_func in" in args[1]
|
||||
assert "duration_s" in kwargs["extra"]
|
||||
assert kwargs["extra"]["duration_s"] >= 0.01
|
||||
|
||||
|
||||
def test_log_timing_failure():
|
||||
logger = MagicMock(spec=logging.Logger)
|
||||
|
||||
@log_timing(logger, level=logging.ERROR)
|
||||
def failing_func():
|
||||
time.sleep(0.01)
|
||||
raise ValueError("Something went wrong")
|
||||
|
||||
with pytest.raises(ValueError, match="Something went wrong"):
|
||||
failing_func()
|
||||
|
||||
assert logger.log.call_count == 2
|
||||
# First call: Starting
|
||||
logger.log.assert_any_call(logging.ERROR, "Starting failing_func")
|
||||
# Second call: FAILED
|
||||
args, kwargs = logger.log.call_args_list[1]
|
||||
assert args[0] == logging.ERROR
|
||||
assert "failing_func FAILED after" in args[1]
|
||||
assert "Something went wrong" in args[1]
|
||||
assert "duration_s" in kwargs["extra"]
|
||||
|
||||
|
||||
def test_merge_log_context():
|
||||
ctx1 = {"a": 1, "b": 2}
|
||||
ctx2 = {"b": 3, "c": 4}
|
||||
merged = merge_log_context(ctx1, ctx2, d=5)
|
||||
assert merged == {"a": 1, "b": 3, "c": 4, "d": 5}
|
||||
|
||||
merged_none = merge_log_context(ctx1, None)
|
||||
assert merged_none == {"a": 1, "b": 2}
|
||||
@@ -1,26 +0,0 @@
|
||||
from aarecommon.models.models import MLBoxType, MLOutputModel
|
||||
|
||||
|
||||
def test_add_box_generates_unique_keys():
|
||||
model = MLOutputModel()
|
||||
key1 = model.add_box(MLBoxType.CRYSTAL, (1, 2, 3, 4), 0.8)
|
||||
key2 = model.add_box(MLBoxType.CRYSTAL, (5, 6, 7, 8), 0.9)
|
||||
|
||||
assert key1 == "Crystal"
|
||||
assert key2 == "Crystal_2"
|
||||
|
||||
|
||||
def test_get_best_for_class_returns_highest_confidence():
|
||||
model = MLOutputModel()
|
||||
model.add_box(MLBoxType.PIN, (1, 1, 2, 2), 0.3)
|
||||
model.add_box(MLBoxType.PIN, (3, 3, 4, 4), 0.7)
|
||||
|
||||
best = model.get_best_for_class(MLBoxType.PIN)
|
||||
|
||||
assert best is not None
|
||||
assert best.conf == 0.7
|
||||
|
||||
|
||||
def test_get_best_for_class_returns_none_when_missing():
|
||||
model = MLOutputModel()
|
||||
assert model.get_best_for_class(MLBoxType.CRYSTAL) is None
|
||||
@@ -1,102 +0,0 @@
|
||||
from aarecommon.models.models import (
|
||||
BeamlineStateEnum,
|
||||
BeamMarkCoeffModel,
|
||||
DataCollectionParameters,
|
||||
DewarAddress,
|
||||
MLBoxType,
|
||||
MLOutputModel,
|
||||
SampleShortInfo,
|
||||
)
|
||||
|
||||
|
||||
def test_sample_short_info_methods():
|
||||
info = SampleShortInfo(
|
||||
db_id=1,
|
||||
puck_name="puck1",
|
||||
dewar_name="dew1",
|
||||
sample_name="sample1",
|
||||
run_number=1,
|
||||
aaredb_params=DataCollectionParameters(
|
||||
totalangle=180, processingresolution=1.5, cloud=True
|
||||
),
|
||||
user="group1",
|
||||
pin=3,
|
||||
location=DewarAddress(segment="A", pos=2),
|
||||
)
|
||||
|
||||
addr = info.tell_address()
|
||||
assert addr.puck.segment == "A"
|
||||
assert addr.puck.pos == 2
|
||||
assert addr.pin == 3
|
||||
|
||||
assert info.loc_str() == "A2-3"
|
||||
assert info.loc_str_sort() == "A2-03"
|
||||
assert info.aaredb_params is not None
|
||||
assert info.aaredb_params.totalangle == 180
|
||||
assert info.aaredb_params.processingresolution == 1.5
|
||||
|
||||
info_no_loc = info.model_copy(update={"location": None})
|
||||
assert info_no_loc.loc_str() == "-"
|
||||
assert info_no_loc.loc_str_sort() == ""
|
||||
|
||||
data = {
|
||||
"db_id": 2,
|
||||
"puck_name": "puck2",
|
||||
"dewar_name": "dew2",
|
||||
"sample_name": "sample2",
|
||||
"run_number": 2,
|
||||
"aaredb_params": {"totalrange": 120, "userresolution": 1.8, "cloud": ""},
|
||||
"user": "group2",
|
||||
"pin": 4,
|
||||
"location": {"segment": "B", "pos": 5},
|
||||
}
|
||||
info2 = SampleShortInfo.from_dict(data)
|
||||
assert info2.db_id == 2
|
||||
assert info2.location.segment == "B"
|
||||
assert info2.aaredb_params is not None
|
||||
assert info2.aaredb_params.totalangle == 120
|
||||
assert info2.aaredb_params.processingresolution == 1.8
|
||||
assert info2.aaredb_params.cloud is True
|
||||
|
||||
|
||||
def test_beam_mark_coeff_model_apply():
|
||||
model = BeamMarkCoeffModel(coeff_x=(1.0, 2.0, 5.0), coeff_y=(3.0, 4.0, 6.0))
|
||||
res = model.apply(10.0)
|
||||
assert res.x == 125.0
|
||||
assert res.y == 346.0
|
||||
|
||||
|
||||
def test_ml_output_model_extra_methods():
|
||||
model = MLOutputModel()
|
||||
key = model.add_box(MLBoxType.CRYSTAL, (1, 2, 3, 4), 0.8)
|
||||
|
||||
box_model = model.get_box_model(key)
|
||||
assert box_model.conf == 0.8
|
||||
|
||||
assert model.get_box_tuple(key) == (1, 2, 3, 4)
|
||||
assert model.get_box_tuple("NonExistent") is None
|
||||
|
||||
assert model.get_box_tuple_with_conf(key) == (1, 2, 3, 4, 0.8)
|
||||
assert model.get_box_tuple_with_conf("NonExistent") is None
|
||||
|
||||
tuples = model.get_tuples_for_class(MLBoxType.CRYSTAL)
|
||||
assert len(tuples) == 1
|
||||
assert tuples[0] == (1, 2, 3, 4)
|
||||
|
||||
tuples_conf = model.get_tuples_with_conf_for_class(MLBoxType.CRYSTAL)
|
||||
assert len(tuples_conf) == 1
|
||||
assert tuples_conf[0] == (1, 2, 3, 4, 0.8)
|
||||
|
||||
assert MLOutputModel.get_class_str(MLBoxType.LOOP_ALL) == "Loop_all"
|
||||
assert MLOutputModel.get_class_str(MLBoxType.PIN) == "Pin"
|
||||
assert MLOutputModel.get_class_str(MLBoxType.CRYSTAL) == "Crystal"
|
||||
assert MLOutputModel.get_class_str(MLBoxType.LOOP_FACE) == "Loop_face"
|
||||
assert MLOutputModel.get_class_str(MLBoxType.ICE) == "Ice"
|
||||
assert MLOutputModel.get_class_str(MLBoxType.NEEDLE) == "Needle"
|
||||
assert MLOutputModel.get_class_str(100) == "Unknown"
|
||||
|
||||
|
||||
def test_beamline_state_enum_display_name():
|
||||
assert BeamlineStateEnum.SampleExchange.display_name() == "Sample exchange"
|
||||
assert BeamlineStateEnum.Moving.display_name() == "Moving"
|
||||
assert BeamlineStateEnum.display_name(None) == "-"
|
||||
@@ -1,95 +0,0 @@
|
||||
import pytest
|
||||
from aarecommon.math.raster_grid import grid_to_image_id, image_id_to_grid
|
||||
|
||||
|
||||
def test_grid_to_image_id_single_cell():
|
||||
assert grid_to_image_id(grid_x=1, grid_y=1, number_of_cols=1) == 0
|
||||
|
||||
|
||||
def test_grid_to_image_id_first_row_left_to_right():
|
||||
assert grid_to_image_id(grid_x=1, grid_y=1, number_of_cols=5) == 0
|
||||
assert grid_to_image_id(grid_x=2, grid_y=1, number_of_cols=5) == 1
|
||||
assert grid_to_image_id(grid_x=3, grid_y=1, number_of_cols=5) == 2
|
||||
assert grid_to_image_id(grid_x=4, grid_y=1, number_of_cols=5) == 3
|
||||
assert grid_to_image_id(grid_x=5, grid_y=1, number_of_cols=5) == 4
|
||||
|
||||
|
||||
def test_grid_to_image_id_second_row_right_to_left():
|
||||
assert grid_to_image_id(grid_x=5, grid_y=2, number_of_cols=5) == 5
|
||||
assert grid_to_image_id(grid_x=4, grid_y=2, number_of_cols=5) == 6
|
||||
assert grid_to_image_id(grid_x=3, grid_y=2, number_of_cols=5) == 7
|
||||
assert grid_to_image_id(grid_x=2, grid_y=2, number_of_cols=5) == 8
|
||||
assert grid_to_image_id(grid_x=1, grid_y=2, number_of_cols=5) == 9
|
||||
|
||||
|
||||
def test_image_id_to_grid_first_row_left_to_right():
|
||||
assert image_id_to_grid(0, 5) == (1, 1)
|
||||
assert image_id_to_grid(1, 5) == (2, 1)
|
||||
assert image_id_to_grid(2, 5) == (3, 1)
|
||||
assert image_id_to_grid(3, 5) == (4, 1)
|
||||
assert image_id_to_grid(4, 5) == (5, 1)
|
||||
|
||||
|
||||
def test_image_id_to_grid_second_row_right_to_left():
|
||||
assert image_id_to_grid(5, 5) == (5, 2)
|
||||
assert image_id_to_grid(6, 5) == (4, 2)
|
||||
assert image_id_to_grid(7, 5) == (3, 2)
|
||||
assert image_id_to_grid(8, 5) == (2, 2)
|
||||
assert image_id_to_grid(9, 5) == (1, 2)
|
||||
|
||||
|
||||
def test_round_trip_conversion_for_multiple_grid_shapes():
|
||||
for number_of_cols in (1, 2, 3, 5, 8):
|
||||
for grid_y in range(1, 7):
|
||||
for grid_x in range(1, number_of_cols + 1):
|
||||
image_id = grid_to_image_id(
|
||||
grid_x=grid_x, grid_y=grid_y, number_of_cols=number_of_cols
|
||||
)
|
||||
assert image_id_to_grid(image_id, number_of_cols) == (grid_x, grid_y)
|
||||
|
||||
|
||||
def test_known_3x3_serpentine_mapping():
|
||||
expected = {
|
||||
(1, 1): 0,
|
||||
(2, 1): 1,
|
||||
(3, 1): 2,
|
||||
(3, 2): 3,
|
||||
(2, 2): 4,
|
||||
(1, 2): 5,
|
||||
(1, 3): 6,
|
||||
(2, 3): 7,
|
||||
(3, 3): 8,
|
||||
}
|
||||
for grid_pos, image_id in expected.items():
|
||||
assert grid_to_image_id(grid_pos[0], grid_pos[1], 3) == image_id
|
||||
assert image_id_to_grid(image_id, 3) == grid_pos
|
||||
|
||||
|
||||
def test_grid_to_image_id_rejects_zero_columns():
|
||||
with pytest.raises(ValueError, match="number_of_cols must be >= 1"):
|
||||
grid_to_image_id(grid_x=1, grid_y=1, number_of_cols=0)
|
||||
|
||||
|
||||
def test_grid_to_image_id_rejects_zero_grid_x():
|
||||
with pytest.raises(ValueError, match="grid_x must be >= 1"):
|
||||
grid_to_image_id(grid_x=0, grid_y=1, number_of_cols=5)
|
||||
|
||||
|
||||
def test_grid_to_image_id_rejects_zero_grid_y():
|
||||
with pytest.raises(ValueError, match="grid_y must be >= 1"):
|
||||
grid_to_image_id(grid_x=1, grid_y=0, number_of_cols=5)
|
||||
|
||||
|
||||
def test_grid_to_image_id_rejects_grid_x_larger_than_number_of_cols():
|
||||
with pytest.raises(ValueError, match="grid_x cannot be greater than number_of_cols"):
|
||||
grid_to_image_id(grid_x=6, grid_y=1, number_of_cols=5)
|
||||
|
||||
|
||||
def test_image_id_to_grid_rejects_zero_columns():
|
||||
with pytest.raises(ValueError, match="number_of_cols must be >= 1"):
|
||||
image_id_to_grid(image_id=0, number_of_cols=0)
|
||||
|
||||
|
||||
def test_image_id_to_grid_rejects_negative_image_id():
|
||||
with pytest.raises(ValueError, match="image_id must be >= 0"):
|
||||
image_id_to_grid(image_id=-1, number_of_cols=5)
|
||||
@@ -1,80 +0,0 @@
|
||||
from aarecommon.errors.exception_handler import (
|
||||
LoopCenteringFailed,
|
||||
MountingFailed,
|
||||
SampleException,
|
||||
TellException,
|
||||
)
|
||||
from aarecommon.recurrence_watcher import (
|
||||
RecurrenceWatcher,
|
||||
create_default_watchers,
|
||||
load_watcher_threshold_overrides,
|
||||
redis_key_to_env_var,
|
||||
resolve_exception_class,
|
||||
)
|
||||
|
||||
|
||||
def test_recurrence_watcher_trips_at_threshold_and_resets_on_none():
|
||||
watcher = RecurrenceWatcher(name="tell", observes=TellException, threshold=2)
|
||||
|
||||
assert watcher.observe(MountingFailed) is False
|
||||
assert watcher.observe(MountingFailed) is True
|
||||
assert watcher.streak == 2
|
||||
|
||||
assert watcher.observe(None) is False
|
||||
assert watcher.streak == 0
|
||||
|
||||
|
||||
def test_recurrence_watcher_ignores_non_matching_exception():
|
||||
watcher = RecurrenceWatcher(name="tell", observes=TellException, threshold=2)
|
||||
assert watcher.observe(SampleException) is False
|
||||
assert watcher.streak == 0
|
||||
|
||||
|
||||
def test_load_watcher_threshold_overrides_reads_valid_values_only():
|
||||
values = {
|
||||
"aare:watchers:mx:tell:threshold": "7",
|
||||
"aare:watchers:mx:alc:threshold": b"3",
|
||||
"aare:watchers:mx:smargon:threshold": "not-an-int",
|
||||
}
|
||||
|
||||
overrides = load_watcher_threshold_overrides(
|
||||
beamline="mx", watcher_names=["tell", "alc", "smargon"], get_value=values.get
|
||||
)
|
||||
|
||||
assert overrides == {"tell": 7, "alc": 3}
|
||||
|
||||
|
||||
def test_create_default_watchers_applies_overrides():
|
||||
watchers = {watcher.name: watcher for watcher in create_default_watchers({"alc": 4})}
|
||||
assert watchers["alc"].threshold == 4
|
||||
assert watchers["tell"].threshold > 0
|
||||
|
||||
|
||||
def test_resolve_exception_class_maps_known_classes():
|
||||
assert resolve_exception_class("MountingFailed") is MountingFailed
|
||||
assert resolve_exception_class("LoopCenteringFailed") is LoopCenteringFailed
|
||||
assert resolve_exception_class("UnknownClass") is None
|
||||
|
||||
|
||||
def test_redis_key_to_env_var_converts_colons_and_uppercases():
|
||||
assert redis_key_to_env_var("aare:watchers:default:tell:threshold") == (
|
||||
"AARE_WATCHERS_DEFAULT_TELL_THRESHOLD"
|
||||
)
|
||||
assert redis_key_to_env_var("aare:watchers:mx:alc:threshold") == (
|
||||
"AARE_WATCHERS_MX_ALC_THRESHOLD"
|
||||
)
|
||||
|
||||
|
||||
def test_load_watcher_threshold_overrides_through_env_var_adapter(monkeypatch):
|
||||
monkeypatch.setenv("AARE_WATCHERS_DEFAULT_TELL_THRESHOLD", "9")
|
||||
monkeypatch.setenv("AARE_WATCHERS_DEFAULT_ALC_THRESHOLD", "bad")
|
||||
|
||||
import os
|
||||
|
||||
overrides = load_watcher_threshold_overrides(
|
||||
beamline="default",
|
||||
watcher_names=["tell", "alc", "smargon"],
|
||||
get_value=lambda k: os.getenv(redis_key_to_env_var(k)),
|
||||
)
|
||||
|
||||
assert overrides == {"tell": 9}
|
||||
@@ -1,47 +0,0 @@
|
||||
import pytest
|
||||
from aarecommon.models.models import SampleCameraSettings, ZoomModel
|
||||
|
||||
|
||||
def test_get_camera_settings_exact_match():
|
||||
model = ZoomModel(
|
||||
z={
|
||||
100: SampleCameraSettings(gain=1.0, exposure=0.1),
|
||||
200: SampleCameraSettings(gain=2.0, exposure=0.2),
|
||||
}
|
||||
)
|
||||
|
||||
settings = model.get_camera_settings(100)
|
||||
assert settings.gain == 1.0
|
||||
assert settings.exposure == 0.1
|
||||
|
||||
|
||||
def test_get_camera_settings_interpolates():
|
||||
model = ZoomModel(
|
||||
z={
|
||||
100: SampleCameraSettings(gain=0.0, exposure=0.1),
|
||||
200: SampleCameraSettings(gain=2.0, exposure=0.3),
|
||||
}
|
||||
)
|
||||
|
||||
settings = model.get_camera_settings(150)
|
||||
assert settings.gain == 1.0
|
||||
assert settings.exposure == 0.2
|
||||
|
||||
|
||||
def test_get_camera_settings_raises_when_empty():
|
||||
model = ZoomModel(z={})
|
||||
with pytest.raises(ValueError):
|
||||
model.get_camera_settings(100)
|
||||
|
||||
|
||||
def test_get_camera_settings_below_minimum_returns_first():
|
||||
model = ZoomModel(
|
||||
z={
|
||||
100: SampleCameraSettings(gain=1.0, exposure=0.1),
|
||||
200: SampleCameraSettings(gain=2.0, exposure=0.2),
|
||||
}
|
||||
)
|
||||
|
||||
settings = model.get_camera_settings(50)
|
||||
assert settings.gain == 1.0
|
||||
assert settings.exposure == 0.1
|
||||
Reference in New Issue
Block a user