diff --git a/api/src/opentrons/protocol_api/core/engine/protocol.py b/api/src/opentrons/protocol_api/core/engine/protocol.py index 68b86cbfe34..4089dff4b4d 100644 --- a/api/src/opentrons/protocol_api/core/engine/protocol.py +++ b/api/src/opentrons/protocol_api/core/engine/protocol.py @@ -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 @@ -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), @@ -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]: diff --git a/api/src/opentrons/protocol_engine/commands/load_module.py b/api/src/opentrons/protocol_engine/commands/load_module.py index dcaa396a245..5c1d474be4d 100644 --- a/api/src/opentrons/protocol_engine/commands/load_module.py +++ b/api/src/opentrons/protocol_engine/commands/load_module.py @@ -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 @@ -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 @@ -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.""" diff --git a/api/tests/opentrons/protocol_api/core/engine/test_protocol_core.py b/api/tests/opentrons/protocol_api/core/engine/test_protocol_core.py index d5e71f56f46..8f6589b1104 100644 --- a/api/tests/opentrons/protocol_api/core/engine/test_protocol_core.py +++ b/api/tests/opentrons/protocol_api/core/engine/test_protocol_core.py @@ -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( diff --git a/api/tests/opentrons/protocol_engine/commands/test_load_module.py b/api/tests/opentrons/protocol_engine/commands/test_load_module.py index 84be22d4661..65306f34adc 100644 --- a/api/tests/opentrons/protocol_engine/commands/test_load_module.py +++ b/api/tests/opentrons/protocol_engine/commands/test_load_module.py @@ -1,9 +1,11 @@ """Test load module command.""" import pytest +from typing import cast from decoy import Decoy from opentrons.protocol_engine.errors import LocationIsOccupiedError from opentrons.protocol_engine.state import StateView +from opentrons_shared_data.robot.dev_types import RobotType from opentrons.types import DeckSlotName from opentrons.protocol_engine.types import ( DeckSlotLocation, @@ -11,12 +13,30 @@ ModuleDefinition, ) from opentrons.protocol_engine.execution import EquipmentHandler, LoadedModuleData +from opentrons.protocol_engine import ModuleModel as EngineModuleModel +from opentrons.hardware_control.modules import ModuleType from opentrons.protocol_engine.commands.load_module import ( LoadModuleParams, LoadModuleResult, LoadModuleImplementation, ) +from opentrons.hardware_control.modules.types import ( + ModuleModel as HardwareModuleModel, + TemperatureModuleModel, + MagneticModuleModel, + ThermocyclerModuleModel, + HeaterShakerModuleModel, +) +from opentrons_shared_data.deck.dev_types import ( + DeckDefinitionV5, + SlotDefV3, +) +from opentrons_shared_data.deck import load as load_deck +from opentrons.protocols.api_support.deck_type import ( + STANDARD_OT2_DECK, + STANDARD_OT3_DECK, +) async def test_load_module_implementation( @@ -29,19 +49,29 @@ async def test_load_module_implementation( subject = LoadModuleImplementation(equipment=equipment, state_view=state_view) data = LoadModuleParams( - model=ModuleModel.TEMPERATURE_MODULE_V1, - location=DeckSlotLocation(slotName=DeckSlotName.SLOT_1), + model=ModuleModel.TEMPERATURE_MODULE_V2, + location=DeckSlotLocation(slotName=DeckSlotName.SLOT_D1), moduleId="some-id", ) + + deck_def = load_deck(STANDARD_OT3_DECK, 5) + + decoy.when(state_view.addressable_areas.state.deck_definition).then_return(deck_def) + decoy.when( + state_view.addressable_areas.get_cutout_id_by_deck_slot_name( + DeckSlotName.SLOT_D1 + ) + ).then_return("cutout" + DeckSlotName.SLOT_D1.value) + decoy.when( state_view.geometry.ensure_location_not_occupied( - DeckSlotLocation(slotName=DeckSlotName.SLOT_1) + DeckSlotLocation(slotName=DeckSlotName.SLOT_D1) ) ).then_return(DeckSlotLocation(slotName=DeckSlotName.SLOT_2)) decoy.when( await equipment.load_module( - model=ModuleModel.TEMPERATURE_MODULE_V1, + model=ModuleModel.TEMPERATURE_MODULE_V2, location=DeckSlotLocation(slotName=DeckSlotName.SLOT_2), module_id="some-id", ) @@ -73,12 +103,22 @@ async def test_load_module_implementation_mag_block( data = LoadModuleParams( model=ModuleModel.MAGNETIC_BLOCK_V1, - location=DeckSlotLocation(slotName=DeckSlotName.SLOT_1), + location=DeckSlotLocation(slotName=DeckSlotName.SLOT_D1), moduleId="some-id", ) + + deck_def = load_deck(STANDARD_OT3_DECK, 5) + + decoy.when(state_view.addressable_areas.state.deck_definition).then_return(deck_def) + decoy.when( + state_view.addressable_areas.get_cutout_id_by_deck_slot_name( + DeckSlotName.SLOT_D1 + ) + ).then_return("cutout" + DeckSlotName.SLOT_D1.value) + decoy.when( state_view.geometry.ensure_location_not_occupied( - DeckSlotLocation(slotName=DeckSlotName.SLOT_1) + DeckSlotLocation(slotName=DeckSlotName.SLOT_D1) ) ).then_return(DeckSlotLocation(slotName=DeckSlotName.SLOT_2)) @@ -114,16 +154,162 @@ async def test_load_module_raises_if_location_occupied( subject = LoadModuleImplementation(equipment=equipment, state_view=state_view) data = LoadModuleParams( - model=ModuleModel.TEMPERATURE_MODULE_V1, - location=DeckSlotLocation(slotName=DeckSlotName.SLOT_1), + model=ModuleModel.TEMPERATURE_MODULE_V2, + location=DeckSlotLocation(slotName=DeckSlotName.SLOT_D1), moduleId="some-id", ) + deck_def = load_deck(STANDARD_OT3_DECK, 5) + + decoy.when(state_view.addressable_areas.state.deck_definition).then_return(deck_def) + decoy.when( + state_view.addressable_areas.get_cutout_id_by_deck_slot_name( + DeckSlotName.SLOT_D1 + ) + ).then_return("cutout" + DeckSlotName.SLOT_D1.value) + decoy.when( state_view.geometry.ensure_location_not_occupied( - DeckSlotLocation(slotName=DeckSlotName.SLOT_1) + DeckSlotLocation(slotName=DeckSlotName.SLOT_D1) ) ).then_raise(LocationIsOccupiedError("Get your own spot!")) with pytest.raises(LocationIsOccupiedError): await subject.execute(data) + + +@pytest.mark.parametrize( + ( + "requested_model", + "engine_model", + "deck_def", + "slot_name", + "robot_type", + ), + [ + ( + TemperatureModuleModel.TEMPERATURE_V2, + EngineModuleModel.TEMPERATURE_MODULE_V2, + load_deck(STANDARD_OT3_DECK, 5), + DeckSlotName.SLOT_D2, + "OT-3 Standard", + ), + ( + ThermocyclerModuleModel.THERMOCYCLER_V1, + EngineModuleModel.THERMOCYCLER_MODULE_V1, + load_deck(STANDARD_OT2_DECK, 5), + DeckSlotName.SLOT_1, + "OT-2 Standard", + ), + ( + ThermocyclerModuleModel.THERMOCYCLER_V2, + EngineModuleModel.THERMOCYCLER_MODULE_V2, + load_deck(STANDARD_OT3_DECK, 5), + DeckSlotName.SLOT_A2, + "OT-3 Standard", + ), + ( + HeaterShakerModuleModel.HEATER_SHAKER_V1, + EngineModuleModel.HEATER_SHAKER_MODULE_V1, + load_deck(STANDARD_OT3_DECK, 5), + DeckSlotName.SLOT_A2, + "OT-3 Standard", + ), + ], +) +async def test_load_module_raises_wrong_location( + decoy: Decoy, + equipment: EquipmentHandler, + state_view: StateView, + requested_model: HardwareModuleModel, + engine_model: EngineModuleModel, + deck_def: DeckDefinitionV5, + slot_name: DeckSlotName, + robot_type: RobotType, +) -> None: + """It should issue a load module engine command.""" + subject = LoadModuleImplementation(equipment=equipment, state_view=state_view) + + data = LoadModuleParams( + model=engine_model, + location=DeckSlotLocation(slotName=slot_name), + moduleId="some-id", + ) + + decoy.when(state_view.config.robot_type).then_return(robot_type) + + if robot_type == "OT-2 Standard": + decoy.when( + state_view.addressable_areas.get_slot_definition(slot_name.id) + ).then_return(cast(SlotDefV3, {"compatibleModuleTypes": []})) + else: + decoy.when(state_view.addressable_areas.state.deck_definition).then_return( + deck_def + ) + decoy.when( + state_view.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(model=requested_model).value} cannot be loaded into slot {slot_name}", + ): + await subject.execute(data) + + +@pytest.mark.parametrize( + ( + "requested_model", + "engine_model", + "deck_def", + "slot_name", + "robot_type", + ), + [ + ( + MagneticModuleModel.MAGNETIC_V2, + EngineModuleModel.MAGNETIC_MODULE_V2, + load_deck(STANDARD_OT3_DECK, 5), + DeckSlotName.SLOT_A2, + "OT-3 Standard", + ), + ], +) +async def test_load_module_raises_module_fixture_id_does_not_exist( + decoy: Decoy, + equipment: EquipmentHandler, + state_view: StateView, + requested_model: HardwareModuleModel, + engine_model: EngineModuleModel, + 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.""" + subject = LoadModuleImplementation(equipment=equipment, state_view=state_view) + + data = LoadModuleParams( + model=engine_model, + location=DeckSlotLocation(slotName=slot_name), + moduleId="some-id", + ) + + decoy.when(state_view.config.robot_type).then_return(robot_type) + + if robot_type == "OT-2 Standard": + decoy.when( + state_view.addressable_areas.get_slot_definition(slot_name.id) + ).then_return(cast(SlotDefV3, {"compatibleModuleTypes": []})) + else: + decoy.when(state_view.addressable_areas.state.deck_definition).then_return( + deck_def + ) + decoy.when( + state_view.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.", + ): + await subject.execute(data) diff --git a/robot-server/tests/integration/http_api/commands/test_load_module_success.tavern.yaml b/robot-server/tests/integration/http_api/commands/test_load_module_success.tavern.yaml index 8e4e99528a7..c9cc22f8e0d 100644 --- a/robot-server/tests/integration/http_api/commands/test_load_module_success.tavern.yaml +++ b/robot-server/tests/integration/http_api/commands/test_load_module_success.tavern.yaml @@ -49,7 +49,7 @@ stages: params: model: '{model}' location: - slotName: '10' + slotName: '7' response: strict: - json:off