refactor(api): Delete dead PipetteStore code and type nozzle maps as non-Optional (#16481)

Author: SyntaxColoring

api/src/opentrons/protocol_engine/state/pipettes.py

@@ -96,14 +96,17 @@ class StaticPipetteConfig:
     nozzle_offset_z: float
     pipette_bounding_box_offsets: PipetteBoundingBoxOffsets
     bounding_nozzle_offsets: BoundingNozzlesOffsets
-    default_nozzle_map: NozzleMap
+    default_nozzle_map: NozzleMap  # todo(mm, 2024-10-14): unused, remove?
     lld_settings: Optional[Dict[str, Dict[str, float]]]
 
 
 @dataclasses.dataclass
 class PipetteState:
     """Basic pipette data state and getter methods."""
 
+    # todo(mm, 2024-10-14): It's getting difficult to ensure that all of these
+    # attributes are populated at the appropriate times. Refactor to a
+    # single dict-of-many-things instead of many dicts-of-single-things.
     pipettes_by_id: Dict[str, LoadedPipette]
     aspirated_volume_by_id: Dict[str, Optional[float]]
     current_location: Optional[CurrentPipetteLocation]
@@ -112,7 +115,7 @@ class PipetteState:
     movement_speed_by_id: Dict[str, Optional[float]]
     static_config_by_id: Dict[str, StaticPipetteConfig]
     flow_rates_by_id: Dict[str, FlowRates]
-    nozzle_configuration_by_id: Dict[str, Optional[NozzleMap]]
+    nozzle_configuration_by_id: Dict[str, NozzleMap]
     liquid_presence_detection_by_id: Dict[str, bool]
 
 
@@ -167,11 +170,6 @@ def _set_load_pipette(self, state_update: update_types.StateUpdate) -> None:
             self._state.aspirated_volume_by_id[pipette_id] = None
             self._state.movement_speed_by_id[pipette_id] = None
             self._state.attached_tip_by_id[pipette_id] = None
-            static_config = self._state.static_config_by_id.get(pipette_id)
-            if static_config:
-                self._state.nozzle_configuration_by_id[
-                    pipette_id
-                ] = static_config.default_nozzle_map
 
     def _update_tip_state(self, state_update: update_types.StateUpdate) -> None:
         if state_update.pipette_tip_state != update_types.NO_CHANGE:
@@ -632,31 +630,23 @@ def get_plunger_axis(self, pipette_id: str) -> MotorAxis:
 
     def get_nozzle_layout_type(self, pipette_id: str) -> NozzleConfigurationType:
         """Get the current set nozzle layout configuration."""
-        nozzle_map_for_pipette = self._state.nozzle_configuration_by_id.get(pipette_id)
-        if nozzle_map_for_pipette:
-            return nozzle_map_for_pipette.configuration
-        else:
-            return NozzleConfigurationType.FULL
+        nozzle_map_for_pipette = self._state.nozzle_configuration_by_id[pipette_id]
+        return nozzle_map_for_pipette.configuration
 
     def get_is_partially_configured(self, pipette_id: str) -> bool:
         """Determine if the provided pipette is partially configured."""
         return self.get_nozzle_layout_type(pipette_id) != NozzleConfigurationType.FULL
 
-    def get_primary_nozzle(self, pipette_id: str) -> Optional[str]:
+    def get_primary_nozzle(self, pipette_id: str) -> str:
         """Get the primary nozzle, if any, related to the given pipette's nozzle configuration."""
-        nozzle_map = self._state.nozzle_configuration_by_id.get(pipette_id)
-        return nozzle_map.starting_nozzle if nozzle_map else None
+        nozzle_map = self._state.nozzle_configuration_by_id[pipette_id]
+        return nozzle_map.starting_nozzle
 
     def _get_critical_point_offset_without_tip(
         self, pipette_id: str, critical_point: Optional[CriticalPoint]
     ) -> Point:
         """Get the offset of the specified critical point from pipette's mount position."""
-        nozzle_map = self._state.nozzle_configuration_by_id.get(pipette_id)
-        # Nozzle map is unavailable only when there's no pipette loaded
-        # so this is merely for satisfying the type checker
-        assert (
-            nozzle_map is not None
-        ), "Error getting critical point offset. Nozzle map not found."
+        nozzle_map = self._state.nozzle_configuration_by_id[pipette_id]
         match critical_point:
             case CriticalPoint.INSTRUMENT_XY_CENTER:
                 return nozzle_map.instrument_xy_center_offset

api/src/opentrons/protocol_engine/state/update_types.py

@@ -66,9 +66,10 @@ class AddressableArea:
 
 @dataclasses.dataclass
 class PipetteLocationUpdate:
-    """Represents an update to perform on a pipette's location."""
+    """An update to a pipette's location."""
 
     pipette_id: str
+    """The ID of the already-loaded pipette."""
 
     new_location: Well | AddressableArea | None | NoChangeType
     """The pipette's new logical location.
@@ -82,19 +83,30 @@ class PipetteLocationUpdate:
 
 @dataclasses.dataclass
 class LabwareLocationUpdate:
-    """Represents an update to perform on a labware's location."""
+    """An update to a labware's location."""
 
     labware_id: str
+    """The ID of the already-loaded labware."""
 
     new_location: LabwareLocation
-    """The labware's new logical location."""
+    """The labware's new location."""
 
     offset_id: typing.Optional[str]
+    """The ID of the labware's new offset, for its new location."""
 
 
 @dataclasses.dataclass
-class LoadedLabwareUpdate(LabwareLocationUpdate):
-    """Update loaded labware."""
+class LoadedLabwareUpdate:
+    """An update that loads a new labware."""
+
+    labware_id: str
+    """The unique ID of the new labware."""
+
+    new_location: LabwareLocation
+    """The labware's initial location."""
+
+    offset_id: typing.Optional[str]
+    """The ID of the labware's offset."""
 
     display_name: typing.Optional[str]
 
@@ -103,20 +115,30 @@ class LoadedLabwareUpdate(LabwareLocationUpdate):
 
 @dataclasses.dataclass
 class LoadPipetteUpdate:
-    """Update loaded pipette."""
+    """An update that loads a new pipette.
+
+    NOTE: Currently, if this is provided, a PipetteConfigUpdate must always be
+    provided alongside it to fully initialize everything.
+    """
 
     pipette_id: str
+    """The unique ID of the new pipette."""
+
     pipette_name: PipetteNameType
     mount: MountType
     liquid_presence_detection: typing.Optional[bool]
 
 
 @dataclasses.dataclass
 class PipetteConfigUpdate:
-    """Update pipette config."""
+    """An update to a pipette's config."""
 
     pipette_id: str
+    """The ID of the already-loaded pipette."""
+
+    # todo(mm, 2024-10-14): Does serial_number belong in LoadPipetteUpdate?
     serial_number: str
+
     config: pipette_data_provider.LoadedStaticPipetteData
 
 
@@ -237,7 +259,7 @@ def set_labware_location(
         new_location: LabwareLocation,
         new_offset_id: str | None,
     ) -> None:
-        """Set labware location."""
+        """Set a labware's location. See `LabwareLocationUpdate`."""
         self.labware_location = LabwareLocationUpdate(
             labware_id=labware_id,
             new_location=new_location,
@@ -252,7 +274,7 @@ def set_loaded_labware(
         display_name: typing.Optional[str],
         location: LabwareLocation,
     ) -> None:
-        """Add loaded labware to state."""
+        """Add a new labware to state. See `LoadedLabwareUpdate`."""
         self.loaded_labware = LoadedLabwareUpdate(
             definition=definition,
             labware_id=labware_id,
@@ -268,7 +290,7 @@ def set_load_pipette(
         mount: MountType,
         liquid_presence_detection: typing.Optional[bool],
     ) -> None:
-        """Add loaded pipette to state."""
+        """Add a new pipette to state. See `LoadPipetteUpdate`."""
         self.loaded_pipette = LoadPipetteUpdate(
             pipette_id=pipette_id,
             pipette_name=pipette_name,
@@ -282,29 +304,29 @@ def update_pipette_config(
         config: pipette_data_provider.LoadedStaticPipetteData,
         serial_number: str,
     ) -> None:
-        """Update pipette config."""
+        """Update a pipette's config. See `PipetteConfigUpdate`."""
         self.pipette_config = PipetteConfigUpdate(
             pipette_id=pipette_id, config=config, serial_number=serial_number
         )
 
     def update_pipette_nozzle(self, pipette_id: str, nozzle_map: NozzleMap) -> None:
-        """Update pipette nozzle map."""
+        """Update a pipette's nozzle map. See `PipetteNozzleMapUpdate`."""
         self.pipette_nozzle_map = PipetteNozzleMapUpdate(
             pipette_id=pipette_id, nozzle_map=nozzle_map
         )
 
     def update_pipette_tip_state(
         self, pipette_id: str, tip_geometry: typing.Optional[TipGeometry]
     ) -> None:
-        """Update tip state."""
+        """Update a pipette's tip state. See `PipetteTipStateUpdate`."""
         self.pipette_tip_state = PipetteTipStateUpdate(
             pipette_id=pipette_id, tip_geometry=tip_geometry
         )
 
     def mark_tips_as_used(
         self, pipette_id: str, labware_id: str, well_name: str
     ) -> None:
-        """Mark tips in a tip rack as used. See `MarkTipsUsedState`."""
+        """Mark tips in a tip rack as used. See `TipsUsedUpdate`."""
         self.tips_used = TipsUsedUpdate(
             pipette_id=pipette_id, labware_id=labware_id, well_name=well_name
         )

api/tests/opentrons/protocol_api/core/engine/test_instrument_core.py

@@ -1,5 +1,5 @@
 """Test for the ProtocolEngine-based instrument API core."""
-from typing import cast, Optional, Union
+from typing import cast, Optional
 
 from opentrons_shared_data.errors.exceptions import PipetteLiquidNotFoundError
 import pytest
@@ -1227,17 +1227,14 @@ def test_configure_nozzle_layout(
     argnames=["pipette_channels", "nozzle_layout", "primary_nozzle", "expected_result"],
     argvalues=[
         (96, NozzleConfigurationType.FULL, "A1", True),
-        (96, NozzleConfigurationType.FULL, None, True),
         (96, NozzleConfigurationType.ROW, "A1", True),
         (96, NozzleConfigurationType.COLUMN, "A1", True),
         (96, NozzleConfigurationType.COLUMN, "A12", True),
         (96, NozzleConfigurationType.SINGLE, "H12", True),
         (96, NozzleConfigurationType.SINGLE, "A1", True),
         (8, NozzleConfigurationType.FULL, "A1", True),
-        (8, NozzleConfigurationType.FULL, None, True),
         (8, NozzleConfigurationType.SINGLE, "H1", True),
         (8, NozzleConfigurationType.SINGLE, "A1", True),
-        (1, NozzleConfigurationType.FULL, None, True),
     ],
 )
 def test_is_tip_tracking_available(
@@ -1246,7 +1243,7 @@ def test_is_tip_tracking_available(
     subject: InstrumentCore,
     pipette_channels: int,
     nozzle_layout: NozzleConfigurationType,
-    primary_nozzle: Union[str, None],
+    primary_nozzle: str,
     expected_result: bool,
 ) -> None:
     """It should return whether tip tracking is available based on nozzle configuration."""

api/tests/opentrons/protocol_engine/state/test_pipette_store.py

@@ -191,25 +191,52 @@ def test_location_state_update(subject: PipetteStore) -> None:
     )
 
 
-def test_handles_load_pipette(subject: PipetteStore) -> None:
+def test_handles_load_pipette(
+    subject: PipetteStore,
+    supported_tip_fixture: pipette_definition.SupportedTipsDefinition,
+) -> None:
     """It should add the pipette data to the state."""
-    command = create_load_pipette_command(
+    dummy_command = create_succeeded_command()
+
+    load_pipette_update = update_types.LoadPipetteUpdate(
         pipette_id="pipette-id",
         pipette_name=PipetteNameType.P300_SINGLE,
         mount=MountType.LEFT,
+        liquid_presence_detection=None,
+    )
+
+    config = LoadedStaticPipetteData(
+        model="pipette-model",
+        display_name="pipette name",
+        min_volume=1.23,
+        max_volume=4.56,
+        channels=7,
+        flow_rates=FlowRates(
+            default_aspirate={"a": 1},
+            default_dispense={"b": 2},
+            default_blow_out={"c": 3},
+        ),
+        tip_configuration_lookup_table={4: supported_tip_fixture},
+        nominal_tip_overlap={"default": 5},
+        home_position=8.9,
+        nozzle_offset_z=10.11,
+        nozzle_map=get_default_nozzle_map(PipetteNameType.P300_SINGLE),
+        back_left_corner_offset=Point(x=1, y=2, z=3),
+        front_right_corner_offset=Point(x=4, y=5, z=6),
+        pipette_lld_settings={},
+    )
+    config_update = update_types.PipetteConfigUpdate(
+        pipette_id="pipette-id",
+        config=config,
+        serial_number="pipette-serial",
     )
 
     subject.handle_action(
         SucceedCommandAction(
             private_result=None,
-            command=command,
+            command=dummy_command,
             state_update=update_types.StateUpdate(
-                loaded_pipette=update_types.LoadPipetteUpdate(
-                    pipette_id="pipette-id",
-                    pipette_name=PipetteNameType.P300_SINGLE,
-                    mount=MountType.LEFT,
-                    liquid_presence_detection=None,
-                )
+                loaded_pipette=load_pipette_update, pipette_config=config_update
             ),
         )
     )

api/tests/opentrons/protocol_engine/state/test_pipette_view.py

@@ -65,7 +65,7 @@ def get_pipette_view(
     movement_speed_by_id: Optional[Dict[str, Optional[float]]] = None,
     static_config_by_id: Optional[Dict[str, StaticPipetteConfig]] = None,
     flow_rates_by_id: Optional[Dict[str, FlowRates]] = None,
-    nozzle_layout_by_id: Optional[Dict[str, Optional[NozzleMap]]] = None,
+    nozzle_layout_by_id: Optional[Dict[str, NozzleMap]] = None,
     liquid_presence_detection_by_id: Optional[Dict[str, bool]] = None,
 ) -> PipetteView:
     """Get a pipette view test subject with the specified state."""

api/tests/opentrons/protocol_engine/state/test_tip_state.py

@@ -1371,24 +1371,24 @@ def _reconfigure_nozzle_layout(start: str, back_l: str, front_r: str) -> NozzleM
         return nozzle_map
 
     map = _reconfigure_nozzle_layout("A1", "A1", "A1")
-    for x in range(96):
+    for _ in range(96):
         _get_next_and_pickup(map)
     assert _get_next_and_pickup(map) is None
 
     subject.handle_action(actions.ResetTipsAction(labware_id="cool-labware"))
     map = _reconfigure_nozzle_layout("A12", "A12", "A12")
-    for x in range(96):
+    for _ in range(96):
         _get_next_and_pickup(map)
     assert _get_next_and_pickup(map) is None
 
     subject.handle_action(actions.ResetTipsAction(labware_id="cool-labware"))
     map = _reconfigure_nozzle_layout("H1", "H1", "H1")
-    for x in range(96):
+    for _ in range(96):
         _get_next_and_pickup(map)
     assert _get_next_and_pickup(map) is None
 
     subject.handle_action(actions.ResetTipsAction(labware_id="cool-labware"))
     map = _reconfigure_nozzle_layout("H12", "H12", "H12")
-    for x in range(96):
+    for _ in range(96):
         _get_next_and_pickup(map)
     assert _get_next_and_pickup(map) is None