Add attenuator motor squad to i19 access controlled devices

* Add attenuator motor squad class to represent a set of mutually
   cooperating motors which are not co-mounted ( therefore do not
   form a physical stage ); but rather act in concert

* Add initimately related position demand class used as driving
   argument to command motor squad to alter the posture of a beamline's
   transmission system

* Add tests for both new classes:
   AttenuatorMotorPositionDemands
   AttenuatorMotorSquad

* Adds access controlled subdirectory in the i19 device tests
   (to match the src devices file system layout)

* Moves established test class for access controlled I19 shutters
   into the access controlled subdirectory

dodal#1384

Author: CoePaul

src/dodal/devices/i19/access_controlled/attenuator_motor_squad.py

@@ -0,0 +1,65 @@
+from typing import Annotated, Any, Self
+
+from ophyd_async.core import AsyncStatus
+from pydantic import BaseModel, model_validator
+from pydantic.types import PositiveInt, StringConstraints
+
+from dodal.devices.i19.access_controlled.blueapi_device import (
+    OpticsBlueAPIDevice,
+)
+from dodal.devices.i19.access_controlled.hutch_access import ACCESS_DEVICE_NAME
+
+PermittedKeyStr = Annotated[str, StringConstraints(pattern="^[A-Za-z0-9-_]*$")]
+
+
+class AttenuatorMotorPositionDemands(BaseModel):
+    continuous_demands: dict[PermittedKeyStr, float] = {}
+    indexed_demands: dict[PermittedKeyStr, PositiveInt] = {}
+
+    @model_validator(mode="after")
+    def no_keys_clash(self) -> Self:
+        common_key_filter = filter(
+            lambda k: k in self.continuous_demands, self.indexed_demands
+        )
+        common_key_count = sum(1 for _ in common_key_filter)
+        if common_key_count < 1:
+            return self
+        else:
+            ks: str = "key" if common_key_count == 1 else "keys"
+            error_msg = (
+                f"{common_key_count} common {ks} found in distinct motor demands"
+            )
+            raise ValueError(error_msg)
+
+    def restful_format(self) -> dict[PermittedKeyStr, Any]:
+        return self.continuous_demands | self.indexed_demands
+
+
+class AttenuatorMotorSquad(OpticsBlueAPIDevice):
+    """ I19-specific proxy device which requests absorber position changes in the x-ray attenuator.
+
+    Sends REST call to blueapi controlling optics on the I19 cluster.
+     The hutch in use is compared against the hutch which sent the REST call.
+    Only the hutch in use will be permitted to execute a plan (requesting motor moves).
+    As the two hutches are located in series, checking the hutch in use is necessary to \
+    avoid accidentally operating optics devices from one hutch while the other has beam time.
+
+    The name of the hutch that wants to operate the optics device is passed to the \
+    access controlled device upon instantiation of the latter.
+
+    For details see the architecture described in \
+    https://github.com/DiamondLightSource/i19-bluesky/issues/30.
+    """
+
+    @AsyncStatus.wrap
+    async def set(self, value: AttenuatorMotorPositionDemands):
+        request_params = {
+            "name": "operate_motor_squad_plan",
+            "params": {
+                "experiment_hutch": self._get_invoking_hutch(),
+                "access_device": ACCESS_DEVICE_NAME,
+                "attenuator_demands": value.restful_format(),
+            },
+            "instrument_session": self.instrument_session,
+        }
+        await super().set(request_params)

src/dodal/devices/i19/access_controlled/blueapi_device.py

@@ -29,11 +29,18 @@ class OpticsBlueAPIDevice(StandardReadable, Movable[D]):
     https://github.com/DiamondLightSource/i19-bluesky/issues/30.
     """
 
-    def __init__(self, name: str = "") -> None:
+    def __init__(
+        self, hutch: HutchState, instrument_session: str = "", name: str = ""
+    ) -> None:
+        self.hutch_request = hutch
+        self.instrument_session = instrument_session
         self.url = OPTICS_BLUEAPI_URL
         self.headers = HEADERS
         super().__init__(name)
 
+    def _get_invoking_hutch(self) -> str:
+        return self.hutch_request.value
+
     @AsyncStatus.wrap
     async def set(self, value: D):
         """ On set send a POST request to the optics blueapi with the name and \

src/dodal/devices/i19/access_controlled/shutter.py

@@ -36,16 +36,14 @@ def __init__(
         # see https://github.com/DiamondLightSource/blueapi/issues/1187
         with self.add_children_as_readables(StandardReadableFormat.HINTED_SIGNAL):
             self.shutter_status = epics_signal_r(ShutterState, f"{prefix}STA")
-        self.hutch_request = hutch
-        self.instrument_session = instrument_session
-        super().__init__(name)
+        super().__init__(hutch=hutch, instrument_session=instrument_session, name=name)
 
     @AsyncStatus.wrap
     async def set(self, value: ShutterDemand):
         request_params = {
             "name": "operate_shutter_plan",
             "params": {
-                "experiment_hutch": self.hutch_request.value,
+                "experiment_hutch": self._get_invoking_hutch(),
                 "access_device": ACCESS_DEVICE_NAME,
                 "shutter_demand": value.value,
             },

tests/devices/i19/access_controlled/__init__.py

@@ -0,0 +1,128 @@
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from aiohttp.client import ClientConnectionError
+from bluesky.run_engine import RunEngine
+
+from dodal.devices.i19.access_controlled.attenuator_motor_squad import (
+    AttenuatorMotorPositionDemands,
+    AttenuatorMotorSquad,
+)
+from dodal.devices.i19.access_controlled.blueapi_device import HutchState
+
+
+def given_position_demands() -> AttenuatorMotorPositionDemands:
+    position_demand = MagicMock()
+    restful_payload = {"x": 54.3, "y": 72.1, "w": 4}
+    position_demand.restful_format = MagicMock(return_value=restful_payload)
+    return position_demand
+
+
+def given_an_unhappy_restful_response() -> AsyncMock:
+    unhappy_response = AsyncMock()
+    unhappy_response.ok = False
+    unhappy_response.json.return_value = {"task_id": "alas_not"}
+    return unhappy_response
+
+
+async def given_a_squad_of_attenuator_motors(
+    hutch: HutchState,
+) -> AttenuatorMotorSquad:
+    motor_squad = AttenuatorMotorSquad(
+        hutch, instrument_session="cm54321-0", name="attenuator_motor_squad"
+    )
+    await motor_squad.connect(mock=True)
+
+    motor_squad.url = "http://test-blueapi.url"
+    return motor_squad
+
+
+@pytest.fixture
+async def eh1_motor_squad(re: RunEngine) -> AttenuatorMotorSquad:
+    return await given_a_squad_of_attenuator_motors(HutchState.EH1)
+
+
+@pytest.fixture
+async def eh2_motor_squad(re: RunEngine) -> AttenuatorMotorSquad:
+    return await given_a_squad_of_attenuator_motors(HutchState.EH2)
+
+
+@pytest.mark.parametrize("invoking_hutch", [HutchState.EH1, HutchState.EH2])
+async def test_that_motor_squad_can_be_instantiated(invoking_hutch):
+    motor_squad: AttenuatorMotorSquad = await given_a_squad_of_attenuator_motors(
+        invoking_hutch
+    )
+    assert isinstance(motor_squad, AttenuatorMotorSquad)
+
+
+@pytest.mark.parametrize("invoking_hutch", [HutchState.EH1, HutchState.EH2])
+@patch(
+    "dodal.devices.i19.access_controlled.attenuator_motor_squad.OpticsBlueAPIDevice.set",
+    new_callable=AsyncMock,
+)
+async def test_when_motor_squad_is_set_that_expected_request_params_are_passed(
+    internal_setter, invoking_hutch
+):
+    motors: AttenuatorMotorSquad = await given_a_squad_of_attenuator_motors(
+        invoking_hutch
+    )
+    position_demands: AttenuatorMotorPositionDemands = given_position_demands()
+    await motors.set(position_demands)  # when motor position demand is set
+
+    expected_hutch: str = invoking_hutch.value
+    expected_device: str = "access_control"
+    expected_request_name: str = "operate_motor_squad_plan"
+    expected_parameters: dict = {
+        "experiment_hutch": expected_hutch,
+        "access_device": expected_device,
+        "attenuator_demands": {"x": 54.3, "y": 72.1, "w": 4},
+    }
+    expected_instrument_session: str = "cm54321-0"
+    expected_request: dict = {
+        "name": expected_request_name,
+        "params": expected_parameters,
+        "instrument_session": expected_instrument_session,
+    }
+    internal_setter.assert_called_once_with(expected_request)
+
+
+@pytest.mark.parametrize("invoking_hutch", [HutchState.EH1, HutchState.EH2])
+@patch("dodal.devices.i19.access_controlled.blueapi_device.ClientSession.post")
+async def test_when_rest_post_unsuccessful_that_error_raised(
+    unsuccessful_post, invoking_hutch
+):
+    motors: AttenuatorMotorSquad = await given_a_squad_of_attenuator_motors(
+        invoking_hutch
+    )
+    with pytest.raises(ClientConnectionError):
+        restful_response: AsyncMock = given_an_unhappy_restful_response()
+        unsuccessful_post.return_value.__aenter__.return_value = restful_response
+
+        postion_demands = given_position_demands()
+        await motors.set(postion_demands)
+
+
+@pytest.mark.parametrize("invoking_hutch", [HutchState.EH1, HutchState.EH2])
+@patch("dodal.devices.i19.access_controlled.blueapi_device.LOGGER")
+@patch("dodal.devices.i19.access_controlled.blueapi_device.ClientSession.put")
+@patch("dodal.devices.i19.access_controlled.blueapi_device.ClientSession.post")
+async def test_that_error_is_logged_when_response_to_position_demand_set_indicates_failure(
+    restful_post, restful_put, logger, invoking_hutch
+):
+    response_to_post: AsyncMock = AsyncMock()
+    response_to_post.ok = True
+    response_to_post.json.return_value = {"task_id": "0123"}
+    restful_post.return_value.__aenter__.return_value = response_to_post
+
+    response_to_put = AsyncMock()
+    response_to_put.ok = False
+    restful_put.return_value.__aenter__.return_value = response_to_put
+
+    motors: AttenuatorMotorSquad = await given_a_squad_of_attenuator_motors(
+        invoking_hutch
+    )
+    position_demands: AttenuatorMotorPositionDemands = given_position_demands()
+
+    logger.error.assert_not_called()
+    await motors.set(position_demands)
+    logger.error.assert_called_once()

tests/devices/i19/access_controlled/test_attenuator_motor_squad.py

@@ -0,0 +1,150 @@
+import pytest
+
+from dodal.devices.i19.access_controlled.attenuator_motor_squad import (
+    AttenuatorMotorPositionDemands,
+)
+
+
+def test_that_attenuator_position_demand_can_be_created_with_only_one_wedge():
+    wedge_position_demands = {"x": 0.5}
+    wheel_position_demands = {}
+    position_demand = AttenuatorMotorPositionDemands(
+        continuous_demands=wedge_position_demands,
+        indexed_demands=wheel_position_demands,
+    )
+    assert position_demand is not None
+
+
+def test_that_attenuator_position_demand_with_only_one_wedge_provides_expected_rest_format():
+    wedge_position_demands = {"y": 14.9}
+    wheel_position_demands = {}
+    position_demand = AttenuatorMotorPositionDemands(
+        continuous_demands=wedge_position_demands,
+        indexed_demands=wheel_position_demands,
+    )
+    restful_payload = position_demand.restful_format()
+    assert restful_payload["y"] == 14.9
+
+
+def test_that_attenuator_position_demand_can_be_created_with_only_one_wheel():
+    wedge_position_demands = {}
+    wheel_position_demands = {"w": 2}
+    position_demand = AttenuatorMotorPositionDemands(
+        continuous_demands=wedge_position_demands,
+        indexed_demands=wheel_position_demands,
+    )
+    assert position_demand is not None
+
+
+def test_that_attenuator_position_demand_with_only_one_wheel_provides_expected_rest_format():
+    wedge_position_demands = {}
+    wheel_position_demands = {"w": 6}
+    position_demand = AttenuatorMotorPositionDemands(
+        continuous_demands=wedge_position_demands,
+        indexed_demands=wheel_position_demands,
+    )
+    restful_payload = position_demand.restful_format()
+    assert restful_payload["w"] == 6
+
+
+def test_that_empty_attenuator_position_demand_can_be_created():
+    wedge_position_demands = {}
+    wheel_position_demands = {}
+    position_demand = AttenuatorMotorPositionDemands(
+        continuous_demands=wedge_position_demands,
+        indexed_demands=wheel_position_demands,
+    )
+    assert position_demand is not None
+
+
+def test_that_empty_attenuator_position_demand_provides_empty_rest_format():
+    wedge_position_demands = {}
+    wheel_position_demands = {}
+    position_demand = AttenuatorMotorPositionDemands(
+        continuous_demands=wedge_position_demands,
+        indexed_demands=wheel_position_demands,
+    )
+    restful_payload = position_demand.restful_format()
+    assert restful_payload == {}
+
+
+def test_that_attenuator_position_demand_triplet_can_be_created():
+    standard_wedge_position_demand = {"x": 25.9, "y": 5.0}
+    standard_wheel_position_demand = {"w": 4}
+    position_demand = AttenuatorMotorPositionDemands(
+        continuous_demands=standard_wedge_position_demand,
+        indexed_demands=standard_wheel_position_demand,
+    )
+    assert position_demand is not None
+
+
+def test_that_attenuator_position_demand_triplet_provides_expected_rest_format():
+    wedge_position_demands = {"x": 0.1, "y": 90.1}
+    wheel_position_demands = {"w": 6}
+    position_demand = AttenuatorMotorPositionDemands(
+        continuous_demands=wedge_position_demands,
+        indexed_demands=wheel_position_demands,
+    )
+    restful_payload = position_demand.restful_format()
+    assert restful_payload == {"x": 0.1, "y": 90.1, "w": 6}
+
+
+# Happy path tests above
+
+# Unhappy path tests below
+
+
+def test_that_attenuator_position_raises_error_when_discrete_and_continuous_demands_overload_axis_label():
+    wedge_position_demands = {"x": 0.1, "v": 90.1}
+    wheel_position_demands = {"w": 6, "v": 7}
+    preamble: str = (
+        "1 validation error for AttenuatorMotorPositionDemands\n  Value error,"
+    )
+    anticipated_message: str = (
+        f"{preamble} 1 common key found in distinct motor demands"
+    )
+    with pytest.raises(expected_exception=ValueError, match=anticipated_message):
+        AttenuatorMotorPositionDemands(
+            continuous_demands=wedge_position_demands,
+            indexed_demands=wheel_position_demands,
+        )
+
+
+def test_that_attenuator_position_creation_raises_error_when_continuous_position_demand_is_none():
+    wedge_position_demands = {"x": None, "y": 90.1}
+    wheel_position_demands = {}
+    with pytest.raises(expected_exception=ValueError):
+        AttenuatorMotorPositionDemands(
+            continuous_demands=wedge_position_demands,
+            indexed_demands=wheel_position_demands,
+        )
+
+
+def test_that_attenuator_position_creation_raises_error_when_indexed_position_demand_is_none():
+    wedge_position_demands = {"x": 14.88, "y": 90.1}
+    wheel_position_demands = {"w": None, "v": 3}
+    with pytest.raises(expected_exception=ValueError):
+        AttenuatorMotorPositionDemands(
+            continuous_demands=wedge_position_demands,
+            indexed_demands=wheel_position_demands,
+        )
+
+
+def test_that_attenuator_position_creation_raises_error_when_continuous_position_key_is_none():
+    wedge_position_demands = {"x": 32.65, None: 80.1}
+    wheel_position_demands = {"w": 8}
+    with pytest.raises(expected_exception=ValueError):
+        AttenuatorMotorPositionDemands(
+            continuous_demands=wedge_position_demands,
+            indexed_demands=wheel_position_demands,
+        )
+
+
+def test_that_attenuator_position_creation_raises_error_when_indexed_position_key_is_none():
+    wedge_position_demands = {"x": 24.08, "y": 71.4}
+    wheel_position_demands = {"w": 1, None: 2}
+    with pytest.raises(expected_exception=ValueError):
+        AttenuatorMotorPositionDemands(
+            continuous_demands=wedge_position_demands,
+            indexed_demands=wheel_position_demands,
+        )