refactor(api): Disallow direct access to .state through Protocol En…

…gine state views (#17016)

api/src/opentrons/protocol_api/core/engine/protocol.py

@@ -425,6 +425,8 @@ def load_module(
                 raise InvalidModuleLocationError(deck_slot, model.name)
 
         robot_type = self._engine_client.state.config.robot_type
+        # todo(mm, 2024-12-03): This might be possible to remove:
+        # Protocol Engine will normalize the deck slot itself.
         normalized_deck_slot = deck_slot.to_equivalent_for_robot_type(robot_type)
 
         result = self._engine_client.execute_command_without_recovery(

api/src/opentrons/protocol_engine/commands/load_module.py

@@ -173,7 +173,7 @@ def _ensure_module_location(
             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,
+                self._state_view.labware.get_deck_definition(),
             )
             cutout_id = (
                 self._state_view.addressable_areas.get_cutout_id_by_deck_slot_name(slot)

api/src/opentrons/protocol_engine/execution/tip_handler.py

@@ -324,8 +324,8 @@ async def verify_tip_presence(
         follow_singular_sensor: Optional[InstrumentProbeType] = None,
     ) -> None:
         """See documentation on abstract base class."""
-        nozzle_configuration = (
-            self._state_view.pipettes.state.nozzle_configuration_by_id[pipette_id]
+        nozzle_configuration = self._state_view.pipettes.get_nozzle_configuration(
+            pipette_id=pipette_id
         )
 
         # Configuration metrics by which tip presence checking is ignored

api/src/opentrons/protocol_engine/protocol_engine.py

@@ -427,7 +427,7 @@ async def finish(
             post_run_hardware_state: The state in which to leave the gantry and motors in
                 after the run is over.
         """
-        if self._state_store.commands.state.stopped_by_estop:
+        if self._state_store.commands.get_is_stopped_by_estop():
             # This handles the case where the E-stop was pressed while we were *not* in the middle
             # of some hardware interaction that would raise it as an exception. For example, imagine
             # we were paused between two commands, or imagine we were executing a waitForDuration.

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

@@ -323,7 +323,7 @@ def _validate_addressable_area_for_simulation(
         return cutout_id
 
 
-class AddressableAreaView(HasState[AddressableAreaState]):
+class AddressableAreaView:
     """Read-only addressable area state view."""
 
     _state: AddressableAreaState
@@ -345,8 +345,8 @@ def deck_extents(self) -> Point:
     @cached_property
     def mount_offsets(self) -> Dict[str, Point]:
         """The left and right mount offsets of the robot."""
-        left_offset = self.state.robot_definition["mountOffsets"]["left"]
-        right_offset = self.state.robot_definition["mountOffsets"]["right"]
+        left_offset = self._state.robot_definition["mountOffsets"]["left"]
+        right_offset = self._state.robot_definition["mountOffsets"]["right"]
         return {
             "left": Point(x=left_offset[0], y=left_offset[1], z=left_offset[2]),
             "right": Point(x=right_offset[0], y=right_offset[1], z=right_offset[2]),
@@ -355,10 +355,10 @@ def mount_offsets(self) -> Dict[str, Point]:
     @cached_property
     def padding_offsets(self) -> Dict[str, float]:
         """The padding offsets to be applied to the deck extents of the robot."""
-        rear_offset = self.state.robot_definition["paddingOffsets"]["rear"]
-        front_offset = self.state.robot_definition["paddingOffsets"]["front"]
-        left_side_offset = self.state.robot_definition["paddingOffsets"]["leftSide"]
-        right_side_offset = self.state.robot_definition["paddingOffsets"]["rightSide"]
+        rear_offset = self._state.robot_definition["paddingOffsets"]["rear"]
+        front_offset = self._state.robot_definition["paddingOffsets"]["front"]
+        left_side_offset = self._state.robot_definition["paddingOffsets"]["leftSide"]
+        right_side_offset = self._state.robot_definition["paddingOffsets"]["rightSide"]
         return {
             "rear": rear_offset,
             "front": front_offset,
@@ -420,12 +420,12 @@ def _check_if_area_is_compatible_with_potential_fixtures(
                     _get_conflicting_addressable_areas_error_string(
                         self._state.potential_cutout_fixtures_by_cutout_id[cutout_id],
                         self._state.loaded_addressable_areas_by_name,
-                        self.state.deck_definition,
+                        self._state.deck_definition,
                     )
                 )
                 area_display_name = (
                     deck_configuration_provider.get_addressable_area_display_name(
-                        area_name, self.state.deck_definition
+                        area_name, self._state.deck_definition
                     )
                 )
                 raise IncompatibleAddressableAreaError(
@@ -504,7 +504,7 @@ def get_addressable_area_offsets_from_cutout(
         addressable_area_name: str,
     ) -> Point:
         """Get the offset form cutout fixture of an addressable area."""
-        for addressable_area in self.state.deck_definition["locations"][
+        for addressable_area in self._state.deck_definition["locations"][
             "addressableAreas"
         ]:
             if addressable_area["id"] == addressable_area_name:
@@ -568,7 +568,7 @@ def get_fixture_by_deck_slot_name(
         self, slot_name: DeckSlotName
     ) -> Optional[CutoutFixture]:
         """Get the Cutout Fixture currently loaded where a specific Deck Slot would be."""
-        deck_config = self.state.deck_configuration
+        deck_config = self._state.deck_configuration
         if deck_config:
             slot_cutout_id = DECK_SLOT_TO_CUTOUT_MAP[slot_name]
             slot_cutout_fixture = None
@@ -581,7 +581,7 @@ def get_fixture_by_deck_slot_name(
                 if cutout_id == slot_cutout_id:
                     slot_cutout_fixture = (
                         deck_configuration_provider.get_cutout_fixture(
-                            cutout_fixture_id, self.state.deck_definition
+                            cutout_fixture_id, self._state.deck_definition
                         )
                     )
                     return slot_cutout_fixture
@@ -605,7 +605,7 @@ def get_fixture_serial_from_deck_configuration_by_deck_slot(
         self, slot_name: DeckSlotName
     ) -> Optional[str]:
         """Get the serial number provided by the deck configuration for a Fixture at a given location."""
-        deck_config = self.state.deck_configuration
+        deck_config = self._state.deck_configuration
         if deck_config:
             slot_cutout_id = DECK_SLOT_TO_CUTOUT_MAP[slot_name]
             # This will only ever be one under current assumptions

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

@@ -579,7 +579,7 @@ def _map_finish_exception_to_error_occurrence(
             )
 
 
-class CommandView(HasState[CommandState]):
+class CommandView:
     """Read-only command state view."""
 
     _state: CommandState
@@ -916,7 +916,7 @@ def raise_fatal_command_error(self) -> None:
         fatal error of the overall run coming from anywhere in the Python script,
         including in between commands.
         """
-        failed_command = self.state.failed_command
+        failed_command = self._state.failed_command
         if (
             failed_command
             and failed_command.command.error
@@ -932,12 +932,16 @@ def get_error_recovery_type(self, command_id: str) -> ErrorRecoveryType:
 
         The command ID is assumed to point to a failed command.
         """
-        return self.state.command_error_recovery_types[command_id]
+        return self._state.command_error_recovery_types[command_id]
 
     def get_is_stopped(self) -> bool:
         """Get whether an engine stop has completed."""
         return self._state.run_completed_at is not None
 
+    def get_is_stopped_by_estop(self) -> bool:
+        """Return whether the engine was stopped specifically by an E-stop."""
+        return self._state.stopped_by_estop
+
     def has_been_played(self) -> bool:
         """Get whether engine has started."""
         return self._state.run_started_at is not None

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

@@ -35,7 +35,7 @@ def _handle_state_update(self, state_update: update_types.StateUpdate) -> None:
             self._state.file_ids.extend(state_update.files_added.file_ids)
 
 
-class FileView(HasState[FileState]):
+class FileView:
     """Read-only engine created file state view."""
 
     _state: FileState

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

@@ -244,7 +244,7 @@ def _set_labware_location(self, state_update: update_types.StateUpdate) -> None:
                 self._state.labware_by_id[labware_id].location = new_location
 
 
-class LabwareView(HasState[LabwareState]):
+class LabwareView:
     """Read-only labware state view."""
 
     _state: LabwareState
@@ -268,7 +268,7 @@ def get(self, labware_id: str) -> LoadedLabware:
 
     def get_id_by_module(self, module_id: str) -> str:
         """Return the ID of the labware loaded on the given module."""
-        for labware_id, labware in self.state.labware_by_id.items():
+        for labware_id, labware in self._state.labware_by_id.items():
             if (
                 isinstance(labware.location, ModuleLocation)
                 and labware.location.moduleId == module_id
@@ -281,7 +281,7 @@ def get_id_by_module(self, module_id: str) -> str:
 
     def get_id_by_labware(self, labware_id: str) -> str:
         """Return the ID of the labware loaded on the given labware."""
-        for labware in self.state.labware_by_id.values():
+        for labware in self._state.labware_by_id.values():
             if (
                 isinstance(labware.location, OnLabwareLocation)
                 and labware.location.labwareId == labware_id

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

@@ -54,7 +54,7 @@ def _handle_liquid_class_loaded_update(
         ] = state_update.liquid_class_id
 
 
-class LiquidClassView(HasState[LiquidClassState]):
+class LiquidClassView:
     """Read-only view of the LiquidClassState."""
 
     _state: LiquidClassState

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

@@ -34,7 +34,7 @@ def _add_liquid(self, action: AddLiquidAction) -> None:
         self._state.liquids_by_id[action.liquid.id] = action.liquid
 
 
-class LiquidView(HasState[LiquidState]):
+class LiquidView:
     """Read-only liquid state view."""
 
     _state: LiquidState

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

@@ -632,7 +632,7 @@ def _handle_absorbance_reader_commands(
             )
 
 
-class ModuleView(HasState[ModuleState]):
+class ModuleView:
     """Read-only view of computed module state."""
 
     _state: ModuleState
@@ -860,8 +860,8 @@ def get_nominal_offset_to_child(
         Labware Position Check offset.
         """
         if (
-            self.state.deck_type == DeckType.OT2_STANDARD
-            or self.state.deck_type == DeckType.OT2_SHORT_TRASH
+            self._state.deck_type == DeckType.OT2_STANDARD
+            or self._state.deck_type == DeckType.OT2_SHORT_TRASH
         ):
             definition = self.get_definition(module_id)
             slot = self.get_location(module_id).slotName.id
@@ -908,7 +908,7 @@ def get_nominal_offset_to_child(
                     "Module location invalid for nominal module offset calculation."
                 )
             module_addressable_area = self.ensure_and_convert_module_fixture_location(
-                location, self.state.deck_type, module.model
+                location, self._state.deck_type, module.model
             )
             module_addressable_area_position = (
                 addressable_areas.get_addressable_area_offsets_from_cutout(

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

@@ -350,7 +350,7 @@ def _fluid_stack_log_if_empty(self, pipette_id: str) -> fluid_stack.FluidStack:
         return stack
 
 
-class PipetteView(HasState[PipetteState]):
+class PipetteView:
     """Read-only view of computed pipettes state."""
 
     _state: PipetteState

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

@@ -119,7 +119,7 @@ def _set_used_tips(self, pipette_id: str, well_name: str, labware_id: str) -> No
             wells[well] = TipRackWellState.USED
 
 
-class TipView(HasState[TipState]):
+class TipView:
     """Read-only tip state view."""
 
     _state: TipState

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

@@ -135,7 +135,7 @@ def _handle_well_operated(
                 )
 
 
-class WellView(HasState[WellState]):
+class WellView:
     """Read-only well state view."""
 
     _state: WellState