feat(hardware): read tip update from firmware whenever a notification is received

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

@@ -47,6 +47,7 @@
     moving_axes_in_move_group,
     gripper_jaw_state_from_fw,
 )
+from .tip_presence_manager import TipPresenceManager
 
 try:
     import aionotify  # type: ignore[import]
@@ -86,7 +87,6 @@
     update_motor_position_estimation,
 )
 from opentrons_hardware.hardware_control.limit_switches import get_limit_switches
-from opentrons_hardware.hardware_control.tip_presence import get_tip_ejector_state
 from opentrons_hardware.hardware_control.current_settings import (
     set_run_current,
     set_hold_current,
@@ -127,7 +127,6 @@
     SubSystemState,
     SubSystem,
     TipStateType,
-    FailedTipStateCheck,
     EstopState,
     GripperJawState,
 )
@@ -172,7 +171,6 @@
 from opentrons_shared_data.errors.exceptions import (
     EStopActivatedError,
     EStopNotPresentError,
-    UnmatchedTipPresenceStates,
     PipetteOverpressureError,
     FirmwareUpdateRequiredError,
 )
@@ -316,6 +314,7 @@ def __init__(
                 "or door, likely because not running on linux"
             )
         self._current_settings: Optional[OT3AxisMap[CurrentConfig]] = None
+        self._tip_presence_manager = TipPresenceManager(self._messenger)
 
     async def get_serial_number(self) -> Optional[str]:
         if not self.initialized:
@@ -867,34 +866,6 @@ async def get_limit_switches(self) -> OT3AxisMap[bool]:
         res = await get_limit_switches(self._messenger, motor_nodes)
         return {node_to_axis(node): bool(val) for node, val in res.items()}
 
-    async def check_for_tip_presence(
-        self,
-        mount: OT3Mount,
-        tip_state: TipStateType,
-        expect_multiple_responses: bool = False,
-    ) -> None:
-        """Raise an error if the expected tip state does not match the current state."""
-        res = await self.get_tip_present_state(mount, expect_multiple_responses)
-        if res != tip_state.value:
-            raise FailedTipStateCheck(tip_state, res)
-
-    async def get_tip_present_state(
-        self,
-        mount: OT3Mount,
-        expect_multiple_responses: bool = False,
-    ) -> bool:
-        """Get the state of the tip ejector flag for a given mount."""
-        expected_responses = 2 if expect_multiple_responses else 1
-        node = sensor_node_for_mount(OT3Mount(mount.value))
-        assert node != NodeId.gripper
-        res = await get_tip_ejector_state(self._messenger, node, expected_responses)  # type: ignore[arg-type]
-        vals = list(res.values())
-        if not all([r == vals[0] for r in vals]):
-            states = {int(sensor): res[sensor] for sensor in res}
-            raise UnmatchedTipPresenceStates(states)
-        tip_present_state = bool(vals[0])
-        return tip_present_state
-
     @staticmethod
     def _tip_motor_nodes(axis_current_keys: KeysView[Axis]) -> List[NodeId]:
         return [axis_to_node(Axis.Q)] if Axis.Q in axis_current_keys else []
@@ -1333,3 +1304,21 @@ async def build_estop_detector(self) -> bool:
     def estop_state_machine(self) -> EstopStateMachine:
         """Accessor for the API to get the state machine, if it exists."""
         return self._estop_state_machine
+
+    @property
+    def tip_presence_manager(self) -> TipPresenceManager:
+        return self._tip_presence_manager
+
+    async def update_tip_detector(self, mount: OT3Mount, sensor_count: int) -> None:
+        """Build indiviudal tip detector for a mount."""
+        await self.teardown_tip_detector(mount)
+        await self._tip_presence_manager.build_detector(mount, sensor_count)
+
+    async def teardown_tip_detector(self, mount: OT3Mount) -> None:
+        await self._tip_presence_manager.clear_detector(mount)
+
+    async def get_tip_status(self, mount: OT3Mount) -> TipStateType:
+        return await self.tip_presence_manager.get_tip_status(mount)
+
+    def current_tip_state(self, mount: OT3Mount) -> Optional[bool]:
+        return self.tip_presence_manager.current_tip_state(mount)

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

@@ -189,6 +189,10 @@ def _sanitize_attached_instrument(
         self._present_nodes = nodes
         self._current_settings: Optional[OT3AxisMap[CurrentConfig]] = None
         self._sim_jaw_state = GripperJawState.HOMED_READY
+        self._sim_tip_state: Dict[OT3Mount, Optional[bool]] = {
+            mount: False if self._attached_instruments[mount] else None
+            for mount in [OT3Mount.LEFT, OT3Mount.RIGHT]
+        }
 
     async def get_serial_number(self) -> Optional[str]:
         return "simulator"
@@ -391,21 +395,6 @@ async def gripper_hold_jaw(
         self._encoder_position[NodeId.gripper_g] = encoder_position_um / 1000.0
         self._sim_jaw_state = GripperJawState.HOLDING
 
-    async def check_for_tip_presence(
-        self,
-        mount: OT3Mount,
-        tip_state: TipStateType,
-        expect_multiple_responses: bool = False,
-    ) -> None:
-        """Raise an error if the given state doesn't match the physical state."""
-        pass
-
-    async def get_tip_present_state(
-        self, mount: OT3Mount, expect_multiple_responses: bool = False
-    ) -> bool:
-        """Get the state of the tip ejector flag for a given mount."""
-        pass
-
     async def get_jaw_state(self) -> GripperJawState:
         """Get the state of the gripper jaw."""
         return self._sim_jaw_state
@@ -747,3 +736,15 @@ def subsystems(self) -> Dict[SubSystem, SubSystemState]:
     def estop_state_machine(self) -> EstopStateMachine:
         """Return an estop state machine locked in the "disengaged" state."""
         return self._estop_state_machine
+
+    async def get_tip_status(self, mount: OT3Mount) -> TipStateType:
+        return TipStateType(self._sim_tip_state[mount])
+
+    def current_tip_state(self, mount: OT3Mount) -> Optional[bool]:
+        return self._sim_tip_state[mount]
+
+    async def update_tip_detector(self, mount: OT3Mount, sensor_count: int) -> None:
+        pass
+
+    async def teardown_tip_detector(self, mount: OT3Mount) -> None:
+        pass

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

@@ -0,0 +1,146 @@
+import logging
+from functools import partial
+from typing import cast, Callable, Optional, List, Set
+from typing_extensions import TypedDict, Literal
+
+from opentrons.hardware_control.types import TipStateType, OT3Mount
+
+from opentrons_hardware.drivers.can_bus import CanMessenger
+from opentrons_hardware.firmware_bindings.constants import NodeId
+from opentrons_hardware.hardware_control.tip_presence import (
+    TipDetector,
+    types as tip_types,
+)
+from opentrons_shared_data.errors.exceptions import (
+    TipDetectorNotFound,
+    UnmatchedTipPresenceStates,
+)
+
+log = logging.getLogger(__name__)
+
+TipListener = Callable[[OT3Mount, bool], None]
+PipetteMountKeys = Literal["left", "right"]
+
+
+class TipDetectorByMount(TypedDict):
+    left: Optional[TipDetector]
+    right: Optional[TipDetector]
+
+
+class UnsubMethodByMount(TypedDict):
+    left: Optional[Callable[[], None]]
+    right: Optional[Callable[[], None]]
+
+
+class TipUpdateByMount(TypedDict):
+    left: Optional[bool]
+    right: Optional[bool]
+
+
+def _mount_to_node(mount: OT3Mount) -> NodeId:
+    return {
+        OT3Mount.LEFT: NodeId.pipette_left,
+        OT3Mount.RIGHT: NodeId.pipette_right,
+    }[mount]
+
+
+class TipPresenceManager:
+    """Handle tip change notification coming from CAN."""
+
+    _listeners: Set[TipListener]
+    _detectors: TipDetectorByMount
+    _unsub_methods: UnsubMethodByMount
+    _last_state: TipUpdateByMount
+
+    def __init__(
+        self,
+        can_messenger: CanMessenger,
+        listeners: Set[TipListener] = set(),
+    ) -> None:
+        self._messenger = can_messenger
+        self._listeners = listeners
+        self._detectors = TipDetectorByMount(left=None, right=None)
+        self._unsub_methods = UnsubMethodByMount(left=None, right=None)
+        self._last_state = TipUpdateByMount(left=None, right=None)
+
+    @staticmethod
+    def _get_key(mount: OT3Mount) -> PipetteMountKeys:
+        assert mount != OT3Mount.GRIPPER
+        return cast(PipetteMountKeys, mount.name.lower())
+
+    async def clear_detector(self, mount: OT3Mount) -> None:
+        """Clean up and remove tip detector."""
+
+        def _unsubscribe() -> None:
+            """Unsubscribe from detector."""
+            unsub = self._unsub_methods[self._get_key(mount)]
+            if unsub:
+                unsub()
+                self.set_unsub(mount, None)
+
+        detector = self.get_detector(mount)
+        if detector:
+            _unsubscribe()
+            detector.cleanup()
+            self.set_detector(mount, None)
+
+    async def build_detector(self, mount: OT3Mount, sensor_count: int) -> None:
+        assert self.get_detector(mount) is None
+        # set up and subscribe to the detector
+        d = TipDetector(self._messenger, _mount_to_node(mount), sensor_count)
+        # listens to the detector so we can immediately notify listeners
+        # the most up-to-date tip state
+        unsub = d.add_subscriber(partial(self._handle_tip_update, mount))
+        self.set_unsub(mount, unsub)
+        self.set_detector(mount, d)
+
+    def _handle_tip_update(
+        self, mount: OT3Mount, update: tip_types.TipNotification
+    ) -> None:
+        """Callback for detector."""
+        self._last_state[self._get_key(mount)] = update.presence
+
+        for listener in self._listeners:
+            listener(mount, update.presence)
+
+    def current_tip_state(self, mount: OT3Mount) -> Optional[bool]:
+        state = self._last_state[self._get_key(mount)]
+        if state is None:
+            log.warning("Tip state for {mount} is unknown")
+        return state
+
+    @staticmethod
+    def _get_tip_presence(results: List[tip_types.TipNotification]) -> TipStateType:
+        # more than one sensor reported, we have to check if their states match
+        if len(set(r.presence for r in results)) > 1:
+            raise UnmatchedTipPresenceStates(
+                {int(r.sensor): int(r.presence) for r in results}
+            )
+        return TipStateType(results[0].presence)
+
+    async def get_tip_status(self, mount: OT3Mount) -> TipStateType:
+        detector = self.get_detector(mount)
+        return self._get_tip_presence(await detector.request_tip_status())
+
+    def get_detector(self, mount: OT3Mount) -> TipDetector:
+        detector = self._detectors[self._get_key(mount)]
+        if not detector:
+            raise TipDetectorNotFound(
+                message=f"Tip detector not set up for {mount} mount",
+                detail={"mount": str(mount)},
+            )
+        return detector
+
+    def set_detector(self, mount: OT3Mount, detector: Optional[TipDetector]) -> None:
+        self._detectors[self._get_key(mount)] = detector
+
+    def set_unsub(self, mount: OT3Mount, unsub: Optional[Callable[[], None]]) -> None:
+        self._unsub_methods[self._get_key(mount)] = unsub
+
+    def add_listener(self, listener: TipListener) -> Callable[[], None]:
+        self._listeners.add(listener)
+
+        def remove() -> None:
+            self._listeners.discard(listener)
+
+        return remove