refactor(api): hardware controller use error codes (#13318)

api/src/opentrons/hardware_control/__init__.py

@@ -14,7 +14,6 @@
 from .pause_manager import PauseManager
 from .backends import Controller, Simulator
 from .types import CriticalPoint, ExecutionState
-from .errors import ExecutionCancelledError, NoTipAttachedError, TipAttachedError
 from .constants import DROP_TIP_RELEASE_DISTANCE
 from .thread_manager import ThreadManager
 from .execution_manager import ExecutionManager
@@ -48,13 +47,10 @@
     "SynchronousAdapter",
     "HardwareControlAPI",
     "CriticalPoint",
-    "NoTipAttachedError",
-    "TipAttachedError",
     "DROP_TIP_RELEASE_DISTANCE",
     "ThreadManager",
     "ExecutionManager",
     "ExecutionState",
-    "ExecutionCancelledError",
     "ThreadedAsyncLock",
     "ThreadedAsyncForbidden",
     "ThreadManagedHardware",

api/src/opentrons/hardware_control/api.py

@@ -20,6 +20,10 @@
     cast,
 )
 
+from opentrons_shared_data.errors.exceptions import (
+    PositionUnknownError,
+    UnsupportedHardwareCommand,
+)
 from opentrons_shared_data.pipette import (
     pipette_load_name_conversions as pipette_load_name,
 )
@@ -54,10 +58,6 @@
     SubSystem,
     SubSystemState,
 )
-from .errors import (
-    MustHomeError,
-    NotSupportedByHardware,
-)
 from . import modules
 from .robot_calibration import (
     RobotCalibrationProvider,
@@ -601,10 +601,13 @@ async def home(self, axes: Optional[List[Axis]] = None) -> None:
         # No internal code passes OT3 axes as arguments on an OT2. But a user/ client
         # can still explicitly specify an OT3 axis even when working on an OT2.
         # Adding this check in order to prevent misuse of axes types.
-        if axes and any(axis not in Axis.ot2_axes() for axis in axes):
-            raise NotSupportedByHardware(
-                f"At least one axis in {axes} is not supported on the OT2."
-            )
+        if axes:
+            unsupported = list(axis not in Axis.ot2_axes() for axis in axes)
+            if any(unsupported):
+                raise UnsupportedHardwareCommand(
+                    message=f"At least one axis in {axes} is not supported on the OT2.",
+                    detail={"unsupported_axes": unsupported},
+                )
         self._reset_last_mount()
         # Initialize/update current_position
         checked_axes = axes or [ax for ax in Axis.ot2_axes()]
@@ -641,16 +644,24 @@ async def current_position(
         plunger_ax = Axis.of_plunger(mount)
         position_axes = [Axis.X, Axis.Y, z_ax, plunger_ax]
 
-        if fail_on_not_homed and (
-            not self._backend.is_homed([ot2_axis_to_string(a) for a in position_axes])
-            or not self._current_position
-        ):
-            raise MustHomeError(
-                f"Current position of {str(mount)} pipette is unknown, please home."
-            )
-
+        if fail_on_not_homed:
+            if not self._current_position:
+                raise PositionUnknownError(
+                    message=f"Current position of {str(mount)} pipette is unknown,"
+                    " please home.",
+                    detail={"mount": str(mount), "missing_axes": position_axes},
+                )
+            axes_str = [ot2_axis_to_string(a) for a in position_axes]
+            if not self._backend.is_homed(axes_str):
+                unhomed = self._backend._unhomed_axes(axes_str)
+                raise PositionUnknownError(
+                    message=f"{str(mount)} pipette axes ({unhomed}) must be homed.",
+                    detail={"mount": str(mount), "unhomed_axes": unhomed},
+                )
         elif not self._current_position and not refresh:
-            raise MustHomeError("Current position is unknown; please home motors.")
+            raise PositionUnknownError(
+                message="Current position is unknown; please home motors."
+            )
         async with self._motion_lock:
             if refresh:
                 smoothie_pos = await self._backend.update_position()
@@ -730,7 +741,10 @@ async def move_axes(
         The effector of the x,y axis is the center of the carriage.
         The effector of the pipette mount axis are the mount critical points but only in z.
         """
-        raise NotSupportedByHardware("move_axes is not supported on the OT-2.")
+        raise UnsupportedHardwareCommand(
+            message="move_axes is not supported on the OT-2.",
+            detail={"axes_commanded": list(position.keys())},
+        )
 
     async def move_rel(
         self,
@@ -748,23 +762,32 @@ async def move_rel(
         # TODO: Remove the fail_on_not_homed and make this the behavior all the time.
         # Having the optional arg makes the bug stick around in existing code and we
         # really want to fix it when we're not gearing up for a release.
-        mhe = MustHomeError(
-            "Cannot make a relative move because absolute position is unknown"
-        )
         if not self._current_position:
             if fail_on_not_homed:
-                raise mhe
+                raise PositionUnknownError(
+                    message="Cannot make a relative move because absolute position"
+                    " is unknown.",
+                    detail={
+                        "mount": str(mount),
+                        "fail_on_not_homed": fail_on_not_homed,
+                    },
+                )
             else:
                 await self.home()
 
         target_position = target_position_from_relative(
             mount, delta, self._current_position
         )
+
         axes_moving = [Axis.X, Axis.Y, Axis.by_mount(mount)]
-        if fail_on_not_homed and not self._backend.is_homed(
-            [ot2_axis_to_string(axis) for axis in axes_moving if axis is not None]
-        ):
-            raise mhe
+        axes_str = [ot2_axis_to_string(a) for a in axes_moving]
+        if fail_on_not_homed and not self._backend.is_homed(axes_str):
+            unhomed = self._backend._unhomed_axes(axes_str)
+            raise PositionUnknownError(
+                message=f"{str(mount)} pipette axes ({unhomed}) must be homed.",
+                detail={"mount": str(mount), "unhomed_axes": unhomed},
+            )
+
         await self._cache_and_maybe_retract_mount(mount)
         await self._move(
             target_position,
@@ -937,7 +960,7 @@ async def prepare_for_aspirate(
         Prepare the pipette for aspiration.
         """
         instrument = self.get_pipette(mount)
-        self.ready_for_tip_action(instrument, HardwareAction.PREPARE_ASPIRATE)
+        self.ready_for_tip_action(instrument, HardwareAction.PREPARE_ASPIRATE, mount)
 
         if instrument.current_volume == 0:
             speed = self.plunger_speed(

api/src/opentrons/hardware_control/backends/controller.py

@@ -145,11 +145,15 @@ async def update_position(self) -> Dict[str, float]:
         await self._smoothie_driver.update_position()
         return self._smoothie_driver.position
 
+    def _unhomed_axes(self, axes: Sequence[str]) -> List[str]:
+        return list(
+            axis
+            for axis in axes
+            if not self._smoothie_driver.homed_flags.get(axis, False)
+        )
+
     def is_homed(self, axes: Sequence[str]) -> bool:
-        for axis in axes:
-            if not self._smoothie_driver.homed_flags.get(axis, False):
-                return False
-        return True
+        return not any(self._unhomed_axes(axes))
 
     async def move(
         self,

api/src/opentrons/hardware_control/backends/ot3controller.py

@@ -132,8 +132,6 @@
 from opentrons.hardware_control.errors import (
     InvalidPipetteName,
     InvalidPipetteModel,
-    FirmwareUpdateRequired,
-    OverPressureDetected,
 )
 from opentrons_hardware.hardware_control.motion import (
     MoveStopCondition,
@@ -173,6 +171,8 @@
     EStopActivatedError,
     EStopNotPresentError,
     UnmatchedTipPresenceStates,
+    PipetteOverpressureError,
+    FirmwareUpdateRequiredError,
 )
 
 from .subsystem_manager import SubsystemManager
@@ -191,12 +191,15 @@
 
 
 def requires_update(func: Wrapped) -> Wrapped:
-    """Decorator that raises FirmwareUpdateRequired if the update_required flag is set."""
+    """Decorator that raises FirmwareUpdateRequiredError if the update_required flag is set."""
 
     @wraps(func)
     async def wrapper(self: Any, *args: Any, **kwargs: Any) -> Any:
         if self.update_required and self.initialized:
-            raise FirmwareUpdateRequired()
+            raise FirmwareUpdateRequiredError(
+                func.__name__,
+                self.subsystems_to_update,
+            )
         return await func(self, *args, **kwargs)
 
     return cast(Wrapped, wrapper)
@@ -347,6 +350,10 @@ def eeprom_data(self) -> EEPROMData:
     def update_required(self) -> bool:
         return self._subsystem_manager.update_required and self._check_updates
 
+    @property
+    def subsystems_to_update(self) -> List[SubSystem]:
+        return self._subsystem_manager.subsystems_to_update
+
     @staticmethod
     def _build_system_hardware(
         can_messenger: CanMessenger,
@@ -776,7 +783,7 @@ def _build_attached_pip(
         attached: ohc_tool_types.PipetteInformation, mount: OT3Mount
     ) -> AttachedPipette:
         if attached.name == FirmwarePipetteName.unknown:
-            raise InvalidPipetteName(name=attached.name_int, mount=mount)
+            raise InvalidPipetteName(name=attached.name_int, mount=mount.name)
         try:
             # TODO (lc 12-8-2022) We should return model as an int rather than
             # a string.
@@ -797,7 +804,7 @@ def _build_attached_pip(
             }
         except KeyError:
             raise InvalidPipetteModel(
-                name=attached.name.name, model=attached.model, mount=mount
+                name=attached.name.name, model=attached.model, mount=mount.name
             )
 
     @staticmethod
@@ -1115,6 +1122,7 @@ def _axis_map_to_present_nodes(
 
     @asynccontextmanager
     async def _monitor_overpressure(self, mounts: List[NodeId]) -> AsyncIterator[None]:
+        msg = "The pressure sensor on the {} mount has exceeded operational limits."
         if ff.overpressure_detection_enabled() and mounts:
             tools_with_id = map_pipette_type_to_sensor_id(
                 mounts, self._subsystem_manager.device_info
@@ -1140,8 +1148,10 @@ def _pop_queue() -> Optional[Tuple[NodeId, ErrorCode]]:
                     q_msg = _pop_queue()
                     if q_msg:
                         mount = Axis.to_ot3_mount(node_to_axis(q_msg[0]))
-                        raise OverPressureDetected(mount.name)
-
+                        raise PipetteOverpressureError(
+                            message=msg.format(str(mount)),
+                            detail={"mount": mount},
+                        )
         else:
             yield
 

api/src/opentrons/hardware_control/backends/simulator.py

@@ -187,11 +187,15 @@ def module_controls(self, module_controls: AttachedModulesControl) -> None:
     async def update_position(self) -> Dict[str, float]:
         return self._position
 
+    def _unhomed_axes(self, axes: Sequence[str]) -> List[str]:
+        return list(
+            axis
+            for axis in axes
+            if not self._smoothie_driver.homed_flags.get(axis, False)
+        )
+
     def is_homed(self, axes: Sequence[str]) -> bool:
-        for axis in axes:
-            if not self._smoothie_driver.homed_flags.get(axis, False):
-                return False
-        return True
+        return not any(self._unhomed_axes(axes))
 
     @ensure_yield
     async def move(

api/src/opentrons/hardware_control/backends/subsystem_manager.py

@@ -13,6 +13,7 @@
     Callable,
     AsyncIterator,
     Union,
+    List,
 )
 
 from opentrons_hardware.hardware_control import network, tools
@@ -144,6 +145,10 @@ def _next_version(target: FirmwareTarget, current: int) -> int:
     def update_required(self) -> bool:
         return bool(self._updates_required)
 
+    @property
+    def subsystems_to_update(self) -> List[SubSystem]:
+        return [target_to_subsystem(t) for t in self._updates_required.keys()]
+
     async def start(self) -> None:
         await self._probe_network_and_cache_fw_updates(
             self._expected_core_targets, True