refactor(api): hardware controller use error codes

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,
@@ -602,8 +602,8 @@ async def home(self, axes: Optional[List[Axis]] = None) -> None:
         # 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."
+            raise UnsupportedHardwareCommand(
+                message=f"At least one axis in {axes} is not supported on the OT2."
             )
         self._reset_last_mount()
         # Initialize/update current_position
@@ -645,12 +645,14 @@ async def current_position(
             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."
+            raise PositionUnknownError(
+                message=f"Current position of {str(mount)} pipette is unknown, please home."
             )
 
         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 +732,9 @@ 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."
+        )
 
     async def move_rel(
         self,
@@ -748,8 +752,8 @@ 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"
+        mhe = PositionUnknownError(
+            message="Cannot make a relative move because absolute position is unknown"
         )
         if not self._current_position:
             if fail_on_not_homed:
@@ -937,7 +941,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/ot3controller.py

@@ -132,8 +132,6 @@
 from opentrons.hardware_control.errors import (
     InvalidPipetteName,
     InvalidPipetteModel,
-    FirmwareUpdateRequired,
-    OverPressureDetected,
 )
 from opentrons_hardware.hardware_control.motion import (
     MoveStopCondition,
@@ -169,6 +167,8 @@
     EStopActivatedError,
     EStopNotPresentError,
     UnmatchedTipPresenceStates,
+    PipetteOverpressureError,
+    FirmwareUpdateRequiredError,
 )
 
 from .subsystem_manager import SubsystemManager
@@ -187,12 +187,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)
@@ -343,6 +346,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,
@@ -768,7 +775,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.
@@ -789,7 +796,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
@@ -1113,8 +1120,7 @@ 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(str(mount))
         else:
             yield
 

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

api/src/opentrons/hardware_control/errors.py

@@ -1,114 +1,43 @@
-from opentrons_shared_data.errors.exceptions import PipetteOverpressureError
-from .types import OT3Mount
+from typing import Optional, Dict, Any
+from opentrons_shared_data.errors.exceptions import (
+    MotionPlanningFailureError,
+    InvalidInstrumentData,
+    RobotInUseError,
+)
 
 
-class OutOfBoundsMove(RuntimeError):
-    def __init__(self, message: str):
-        self.message = message
-        super().__init__()
+class OutOfBoundsMove(MotionPlanningFailureError):
+    def __init__(self, message: str, detail: Dict[str, Any]):
+        super().__init__(message=message, detail=detail)
 
-    def __str__(self) -> str:
-        return f"OutOfBoundsMove: {self.message}"
 
-    def __repr__(self) -> str:
-        return f"<{str(self.__class__)}: {self.message}>"
-
-
-class ExecutionCancelledError(RuntimeError):
-    pass
-
-
-class MustHomeError(RuntimeError):
-    pass
-
-
-class NoTipAttachedError(RuntimeError):
-    pass
-
-
-class TipAttachedError(RuntimeError):
-    pass
-
-
-class InvalidMoveError(ValueError):
-    pass
-
-
-class NotSupportedByHardware(ValueError):
-    """Error raised when attempting to use arguments and values not supported by the specific hardware."""
-
-
-class GripperNotAttachedError(Exception):
-    """An error raised if a gripper is accessed that is not attached."""
-
-    pass
-
-
-class AxisNotPresentError(Exception):
-    """An error raised if an axis that is not present."""
-
-    pass
-
-
-class FirmwareUpdateRequired(RuntimeError):
-    """An error raised when the firmware of the submodules needs to be updated."""
-
-    pass
-
-
-class FirmwareUpdateFailed(RuntimeError):
-    """An error raised when a firmware update fails."""
-
-    pass
-
-
-class OverPressureDetected(PipetteOverpressureError):
-    """An error raised when the pressure sensor max value is exceeded."""
-
-    def __init__(self, mount: str) -> None:
-        return super().__init__(
-            message=f"The pressure sensor on the {mount} mount has exceeded operational limits.",
-            detail={"mount": mount},
+class InvalidCriticalPoint(MotionPlanningFailureError):
+    def __init__(self, cp_name: str, instr: str, message: Optional[str] = None):
+        super().__init__(
+            message=(message or f"Critical point {cp_name} is invalid for a {instr}."),
+            detail={"instrument": instr, "critical point": cp_name},
         )
 
 
-class InvalidPipetteName(KeyError):
+class InvalidPipetteName(InvalidInstrumentData):
     """Raised for an invalid pipette."""
 
-    def __init__(self, name: int, mount: OT3Mount) -> None:
-        self.name = name
-        self.mount = mount
-
-    def __repr__(self) -> str:
-        return f"<{self.__class__.__name__}: name={self.name} mount={self.mount}>"
-
-    def __str__(self) -> str:
-        return f"{self.__class__.__name__}: Pipette name key {self.name} on mount {self.mount.name} is not valid"
+    def __init__(self, name: int, mount: str) -> None:
+        super().__init__(
+            message=f"Invalid pipette name key {name} on mount {mount}",
+            detail={"mount": mount, "name": name},
+        )
 
 
-class InvalidPipetteModel(KeyError):
+class InvalidPipetteModel(InvalidInstrumentData):
     """Raised for a pipette with an unknown model."""
 
-    def __init__(self, name: str, model: str, mount: OT3Mount) -> None:
-        self.name = name
-        self.model = model
-        self.mount = mount
+    def __init__(self, name: str, model: str, mount: str) -> None:
+        super().__init__(detail={"mount": mount, "name": name, "model": model})
 
-    def __repr__(self) -> str:
-        return f"<{self.__class__.__name__}: name={self.name}, model={self.model}, mount={self.mount}>"
 
-    def __str__(self) -> str:
-        return f"{self.__class__.__name__}: {self.name} on {self.mount.name} has an unknown model {self.model}"
-
-
-class UpdateOngoingError(RuntimeError):
+class UpdateOngoingError(RobotInUseError):
     """Error when an update is already happening."""
 
-    def __init__(self, msg: str) -> None:
-        self.msg = msg
-
-    def __repr__(self) -> str:
-        return f"<{self.__class__.__name__}: msg={self.msg}>"
-
-    def __str__(self) -> str:
-        return self.msg
+    def __init__(self, message: str) -> None:
+        super().__init__(message=message)

api/src/opentrons/hardware_control/execution_manager.py

@@ -2,7 +2,7 @@
 import functools
 from typing import Set, TypeVar, Type, cast, Callable, Any, Awaitable, overload
 from .types import ExecutionState
-from .errors import ExecutionCancelledError
+from opentrons_shared_data.errors.exceptions import ExecutionCancelledError
 
 TaskContents = TypeVar("TaskContents")
 

api/src/opentrons/hardware_control/instruments/ot2/pipette.py

@@ -45,7 +45,7 @@
     CriticalPoint,
     BoardRevision,
 )
-from opentrons.hardware_control.errors import InvalidMoveError
+from opentrons.hardware_control.errors import InvalidCriticalPoint
 
 
 from opentrons_shared_data.pipette.dev_types import (
@@ -331,9 +331,7 @@ def critical_point(self, cp_override: Optional[CriticalPoint] = None) -> Point:
             CriticalPoint.GRIPPER_FRONT_CALIBRATION_PIN,
             CriticalPoint.GRIPPER_REAR_CALIBRATION_PIN,
         ]:
-            raise InvalidMoveError(
-                f"Critical point {cp_override.name} is not valid for a pipette"
-            )
+            raise InvalidCriticalPoint(cp_override.name, "pipette")
 
         if not self.has_tip or cp_override == CriticalPoint.NOZZLE:
             cp_type = CriticalPoint.NOZZLE