Skip to content

Commit

Permalink
refactor(api): Relocate module location validation to engine (#14960)
Browse files Browse the repository at this point in the history 
Prevents the engine from accepting load module commands in locations for
modules that match the deck configuraiton but would not have passed
protocol core validation.
  • Loading branch information
@CaseyBatten
CaseyBatten authored Apr 19, 2024
1 parent 8e1794f commit 15bfd98
Show file tree
Hide file tree
Showing 5 changed files with 227 additions and 198 deletions.
26 changes: 0 additions & 26 deletions api/src/opentrons/protocol_api/core/engine/protocol.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@

from opentrons.protocol_engine.commands import LoadModuleResult
from opentrons_shared_data.deck.dev_types import DeckDefinitionV5, SlotDefV3
from opentrons.protocol_engine.resources import deck_configuration_provider
from opentrons_shared_data.labware.labware_definition import LabwareDefinition
from opentrons_shared_data.labware.dev_types import LabwareDefinition as LabwareDefDict
from opentrons_shared_data.pipette.dev_types import PipetteNameType
Expand Down Expand Up @@ -410,7 +409,6 @@ def load_module(

robot_type = self._engine_client.state.config.robot_type
normalized_deck_slot = deck_slot.to_equivalent_for_robot_type(robot_type)
self._ensure_module_location(normalized_deck_slot, module_type)

result = self._engine_client.load_module(
model=EngineModuleModel(model),
Expand Down Expand Up @@ -623,30 +621,6 @@ def get_staging_slot_definitions(self) -> Dict[str, SlotDefV3]:
self._engine_client.state.addressable_areas.get_staging_slot_definitions()
)

def _ensure_module_location(
self, slot: DeckSlotName, module_type: ModuleType
) -> None:
if self._engine_client.state.config.robot_type == "OT-2 Standard":
slot_def = self.get_slot_definition(slot)
compatible_modules = slot_def["compatibleModuleTypes"]
if module_type.value not in compatible_modules:
raise ValueError(
f"A {module_type.value} cannot be loaded into slot {slot}"
)
else:
cutout_fixture_id = ModuleType.to_module_fixture_id(module_type)
module_fixture = deck_configuration_provider.get_cutout_fixture(
cutout_fixture_id,
self._engine_client.state.addressable_areas.state.deck_definition,
)
cutout_id = self._engine_client.state.addressable_areas.get_cutout_id_by_deck_slot_name(
slot
)
if cutout_id not in module_fixture["mayMountTo"]:
raise ValueError(
f"A {module_type.value} cannot be loaded into slot {slot}"
)

def get_slot_item(
self, slot_name: Union[DeckSlotName, StagingSlotName]
) -> Union[LabwareCore, ModuleCore, NonConnectedModuleCore, None]:
Expand Down
31 changes: 31 additions & 0 deletions api/src/opentrons/protocol_engine/commands/load_module.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,9 +7,13 @@
from .command import AbstractCommandImpl, BaseCommand, BaseCommandCreate
from ..types import (
DeckSlotLocation,
ModuleType,
ModuleModel,
ModuleDefinition,
)
from opentrons.types import DeckSlotName

from opentrons.protocol_engine.resources import deck_configuration_provider

if TYPE_CHECKING:
from ..state import StateView
Expand Down Expand Up @@ -108,6 +112,9 @@ def __init__(

async def execute(self, params: LoadModuleParams) -> LoadModuleResult:
"""Check that the requested module is attached and assign its identifier."""
module_type = params.model.as_type()
self._ensure_module_location(params.location.slotName, module_type)

if self._state_view.config.robot_type == "OT-2 Standard":
self._state_view.addressable_areas.raise_if_area_not_in_deck_configuration(
params.location.slotName.id
Expand Down Expand Up @@ -146,6 +153,30 @@ async def execute(self, params: LoadModuleParams) -> LoadModuleResult:
definition=loaded_module.definition,
)

def _ensure_module_location(
self, slot: DeckSlotName, module_type: ModuleType
) -> None:
if self._state_view.config.robot_type == "OT-2 Standard":
slot_def = self._state_view.addressable_areas.get_slot_definition(slot.id)
compatible_modules = slot_def["compatibleModuleTypes"]
if module_type.value not in compatible_modules:
raise ValueError(
f"A {module_type.value} cannot be loaded into slot {slot}"
)
else:
cutout_fixture_id = ModuleType.to_module_fixture_id(module_type)
module_fixture = deck_configuration_provider.get_cutout_fixture(
cutout_fixture_id,
self._state_view.addressable_areas.state.deck_definition,
)
cutout_id = (
self._state_view.addressable_areas.get_cutout_id_by_deck_slot_name(slot)
)
if cutout_id not in module_fixture["mayMountTo"]:
raise ValueError(
f"A {module_type.value} cannot be loaded into slot {slot}"
)


class LoadModule(BaseCommand[LoadModuleParams, LoadModuleResult]):
"""The model for a load module command."""
Expand Down
162 changes: 0 additions & 162 deletions api/tests/opentrons/protocol_api/core/engine/test_protocol_core.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1264,168 +1264,6 @@ def test_load_module(
assert subject.get_labware_on_module(result) is None


@pytest.mark.parametrize(
(
"requested_model",
"engine_model",
"expected_core_cls",
"deck_def",
"slot_name",
"robot_type",
),
[
(
TemperatureModuleModel.TEMPERATURE_V2,
EngineModuleModel.TEMPERATURE_MODULE_V2,
TemperatureModuleCore,
lazy_fixture("ot3_standard_deck_def"),
DeckSlotName.SLOT_D2,
"OT-3 Standard",
),
(
ThermocyclerModuleModel.THERMOCYCLER_V1,
EngineModuleModel.THERMOCYCLER_MODULE_V1,
ThermocyclerModuleCore,
lazy_fixture("ot2_standard_deck_def"),
DeckSlotName.SLOT_1,
"OT-2 Standard",
),
(
ThermocyclerModuleModel.THERMOCYCLER_V2,
EngineModuleModel.THERMOCYCLER_MODULE_V2,
ThermocyclerModuleCore,
lazy_fixture("ot3_standard_deck_def"),
DeckSlotName.SLOT_A2,
"OT-3 Standard",
),
(
HeaterShakerModuleModel.HEATER_SHAKER_V1,
EngineModuleModel.HEATER_SHAKER_MODULE_V1,
HeaterShakerModuleCore,
lazy_fixture("ot3_standard_deck_def"),
DeckSlotName.SLOT_A2,
"OT-3 Standard",
),
],
)
def test_load_module_raises_wrong_location(
decoy: Decoy,
mock_engine_client: EngineClient,
mock_sync_hardware_api: SyncHardwareAPI,
requested_model: ModuleModel,
engine_model: EngineModuleModel,
expected_core_cls: Type[ModuleCore],
subject: ProtocolCore,
deck_def: DeckDefinitionV5,
slot_name: DeckSlotName,
robot_type: RobotType,
) -> None:
"""It should issue a load module engine command."""
mock_hw_mod_1 = decoy.mock(cls=AbstractModule)
mock_hw_mod_2 = decoy.mock(cls=AbstractModule)

decoy.when(mock_hw_mod_1.device_info).then_return({"serial": "abc123"})
decoy.when(mock_hw_mod_2.device_info).then_return({"serial": "xyz789"})
decoy.when(mock_sync_hardware_api.attached_modules).then_return(
[mock_hw_mod_1, mock_hw_mod_2]
)

decoy.when(mock_engine_client.state.config.robot_type).then_return(robot_type)

if robot_type == "OT-2 Standard":
decoy.when(subject.get_slot_definition(slot_name)).then_return(
cast(SlotDefV3, {"compatibleModuleTypes": []})
)
else:
decoy.when(
mock_engine_client.state.addressable_areas.state.deck_definition
).then_return(deck_def)
decoy.when(
mock_engine_client.state.addressable_areas.get_cutout_id_by_deck_slot_name(
slot_name
)
).then_return("cutout" + slot_name.value)

with pytest.raises(
ValueError,
match=f"A {ModuleType.from_model(requested_model).value} cannot be loaded into slot {slot_name}",
):
subject.load_module(
model=requested_model,
deck_slot=slot_name,
configuration=None,
)


@pytest.mark.parametrize(
(
"requested_model",
"engine_model",
"expected_core_cls",
"deck_def",
"slot_name",
"robot_type",
),
[
(
MagneticModuleModel.MAGNETIC_V2,
EngineModuleModel.MAGNETIC_MODULE_V2,
MagneticModuleCore,
lazy_fixture("ot3_standard_deck_def"),
DeckSlotName.SLOT_A2,
"OT-3 Standard",
),
],
)
def test_load_module_raises_module_fixture_id_does_not_exist(
decoy: Decoy,
mock_engine_client: EngineClient,
mock_sync_hardware_api: SyncHardwareAPI,
requested_model: ModuleModel,
engine_model: EngineModuleModel,
expected_core_cls: Type[ModuleCore],
subject: ProtocolCore,
deck_def: DeckDefinitionV5,
slot_name: DeckSlotName,
robot_type: RobotType,
) -> None:
"""It should issue a load module engine command and raise an error for unmatched fixtures."""
mock_hw_mod_1 = decoy.mock(cls=AbstractModule)
mock_hw_mod_2 = decoy.mock(cls=AbstractModule)

decoy.when(mock_hw_mod_1.device_info).then_return({"serial": "abc123"})
decoy.when(mock_hw_mod_2.device_info).then_return({"serial": "xyz789"})
decoy.when(mock_sync_hardware_api.attached_modules).then_return(
[mock_hw_mod_1, mock_hw_mod_2]
)

decoy.when(mock_engine_client.state.config.robot_type).then_return(robot_type)

if robot_type == "OT-2 Standard":
decoy.when(subject.get_slot_definition(slot_name)).then_return(
cast(SlotDefV3, {"compatibleModuleTypes": []})
)
else:
decoy.when(
mock_engine_client.state.addressable_areas.state.deck_definition
).then_return(deck_def)
decoy.when(
mock_engine_client.state.addressable_areas.get_cutout_id_by_deck_slot_name(
slot_name
)
).then_return("cutout" + slot_name.value)

with pytest.raises(
ValueError,
match=f"Module Type {ModuleType.from_model(requested_model).value} does not have a related fixture ID.",
):
subject.load_module(
model=requested_model,
deck_slot=slot_name,
configuration=None,
)


# APIv2.15 because we're expecting a fixed trash.
@pytest.mark.parametrize("api_version", [APIVersion(2, 15)])
def test_load_mag_block(
Expand Down
Loading

0 comments on commit 15bfd98

Please sign in to comment.