Skip to content

Commit

Permalink
add to tip_handler + change arg name
Browse files Browse the repository at this point in the history
  • Loading branch information
caila-marashaj committed Apr 23, 2024
1 parent 0f564ea commit c675501
Show file tree
Hide file tree
Showing 6 changed files with 38 additions and 21 deletions.
4 changes: 2 additions & 2 deletions api/src/opentrons/hardware_control/backends/ot3controller.py
Original file line number Diff line number Diff line change
Expand Up @@ -1524,10 +1524,10 @@ async def teardown_tip_detector(self, mount: OT3Mount) -> None:
async def get_tip_status(
self,
mount: OT3Mount,
ht_operational_sensor: Optional[InstrumentProbeType] = None,
ht_follow_singular_sensor: Optional[InstrumentProbeType] = None,
) -> TipStateType:
return await self.tip_presence_manager.get_tip_status(
mount, ht_operational_sensor
mount, ht_follow_singular_sensor
)

def current_tip_state(self, mount: OT3Mount) -> Optional[bool]:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -783,7 +783,7 @@ def subsystems(self) -> Dict[SubSystem, SubSystemState]:
async def get_tip_status(
self,
mount: OT3Mount,
ht_operational_sensor: Optional[InstrumentProbeType] = None,
ht_follow_singular_sensor: Optional[InstrumentProbeType] = None,
) -> TipStateType:
return TipStateType(self._sim_tip_state[mount])

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -116,21 +116,21 @@ def current_tip_state(self, mount: OT3Mount) -> Optional[bool]:
@staticmethod
def _get_tip_presence(
results: List[tip_types.TipNotification],
ht_operational_sensor: Optional[InstrumentProbeType] = None,
ht_follow_singular_sensor: Optional[InstrumentProbeType] = None,
) -> TipStateType:
"""
We can use ht_operational_sensor used to specify that we only care
We can use ht_follow_singular_sensor used to specify that we only care
about the status of one tip presence sensor on a high throughput
pipette, and the other is allowed to be different.
"""
if ht_operational_sensor:
target_sensor_id = sensor_id_for_instrument(ht_operational_sensor)
if ht_follow_singular_sensor:
target_sensor_id = sensor_id_for_instrument(ht_follow_singular_sensor)
for r in results:
if r.sensor == target_sensor_id:
return TipStateType(r.presence)
# raise an error if requested sensor response isn't found
raise GeneralError(
message=f"Requested status for sensor {ht_operational_sensor} not found."
message=f"Requested status for sensor {ht_follow_singular_sensor} not found."
)
# more than one sensor reported, we have to check if their states match
if len(set(r.presence for r in results)) > 1:
Expand All @@ -142,11 +142,11 @@ def _get_tip_presence(
async def get_tip_status(
self,
mount: OT3Mount,
ht_operational_sensor: Optional[InstrumentProbeType] = None,
ht_follow_singular_sensor: Optional[InstrumentProbeType] = None,
) -> TipStateType:
detector = self.get_detector(mount)
return self._get_tip_presence(
await detector.request_tip_status(), ht_operational_sensor
await detector.request_tip_status(), ht_follow_singular_sensor
)

def get_detector(self, mount: OT3Mount) -> TipDetector:
Expand Down
10 changes: 6 additions & 4 deletions api/src/opentrons/hardware_control/ot3api.py
Original file line number Diff line number Diff line change
Expand Up @@ -2078,7 +2078,7 @@ async def _high_throughput_check_tip(self) -> AsyncIterator[None]:
async def get_tip_presence_status(
self,
mount: Union[top_types.Mount, OT3Mount],
ht_operational_sensor: Optional[InstrumentProbeType] = None,
ht_follow_singular_sensor: Optional[InstrumentProbeType] = None,
) -> TipStateType:
"""
Check tip presence status. If a high throughput pipette is present,
Expand All @@ -2093,18 +2093,20 @@ async def get_tip_presence_status(
):
await stack.enter_async_context(self._high_throughput_check_tip())
result = await self._backend.get_tip_status(
real_mount, ht_operational_sensor
real_mount, ht_follow_singular_sensor
)
return result

async def verify_tip_presence(
self,
mount: Union[top_types.Mount, OT3Mount],
expected: TipStateType,
ht_operational_sensor: Optional[InstrumentProbeType] = None,
ht_follow_singular_sensor: Optional[InstrumentProbeType] = None,
) -> None:
real_mount = OT3Mount.from_mount(mount)
status = await self.get_tip_presence_status(real_mount, ht_operational_sensor)
status = await self.get_tip_presence_status(
real_mount, ht_follow_singular_sensor
)
if status != expected:
raise FailedTipStateCheck(expected, status.value)

Expand Down
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
"""Flex-specific extensions to instrument configuration."""
from typing import Union
from typing import Union, Optional
from typing_extensions import Protocol

from .types import MountArgType
Expand All @@ -9,6 +9,7 @@
)
from opentrons.hardware_control.types import (
TipStateType,
InstrumentProbeType,
)
from opentrons.hardware_control.instruments.ot3.instrument_calibration import (
PipetteOffsetSummary,
Expand Down Expand Up @@ -42,7 +43,10 @@ async def get_tip_presence_status(
...

async def verify_tip_presence(
self, mount: MountArgType, expected: TipStateType
self,
mount: MountArgType,
expected: TipStateType,
follow_singular_sensor: Optional[InstrumentProbeType] = None,
) -> None:
"""Check tip presence status and raise if it does not match `expected`."""
...
21 changes: 16 additions & 5 deletions api/src/opentrons/protocol_engine/execution/tip_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
from typing_extensions import Protocol as TypingProtocol

from opentrons.hardware_control import HardwareControlAPI
from opentrons.hardware_control.types import FailedTipStateCheck
from opentrons.hardware_control.types import FailedTipStateCheck, InstrumentProbeType
from opentrons_shared_data.errors.exceptions import (
CommandPreconditionViolated,
CommandParameterLimitViolated,
Expand Down Expand Up @@ -74,7 +74,10 @@ async def get_tip_presence(self, pipette_id: str) -> TipPresenceStatus:
"""Get tip presence status on the pipette."""

async def verify_tip_presence(
self, pipette_id: str, expected: TipPresenceStatus
self,
pipette_id: str,
expected: TipPresenceStatus,
follow_singular_sensor: Optional[InstrumentProbeType] = None,
) -> None:
"""Verify the expected tip presence status."""

Expand Down Expand Up @@ -237,7 +240,10 @@ async def get_tip_presence(self, pipette_id: str) -> TipPresenceStatus:
return TipPresenceStatus.UNKNOWN

async def verify_tip_presence(
self, pipette_id: str, expected: TipPresenceStatus
self,
pipette_id: str,
expected: TipPresenceStatus,
follow_singular_sensor: Optional[InstrumentProbeType] = None,
) -> None:
"""Verify the expecterd tip presence status of the pipette.
Expand All @@ -247,7 +253,9 @@ async def verify_tip_presence(
try:
ot3api = ensure_ot3_hardware(hardware_api=self._hardware_api)
hw_mount = self._state_view.pipettes.get_mount(pipette_id).to_hw_mount()
await ot3api.verify_tip_presence(hw_mount, expected.to_hw_state())
await ot3api.verify_tip_presence(
hw_mount, expected.to_hw_state(), follow_singular_sensor
)
except HardwareNotSupportedError:
# Tip presence sensing is not supported on the OT2
pass
Expand Down Expand Up @@ -332,7 +340,10 @@ async def add_tip(self, pipette_id: str, tip: TipGeometry) -> None:
assert False, "TipHandler.add_tip should not be used with virtual pipettes"

async def verify_tip_presence(
self, pipette_id: str, expected: TipPresenceStatus
self,
pipette_id: str,
expected: TipPresenceStatus,
follow_singular_sensor: Optional[InstrumentProbeType] = None,
) -> None:
"""Verify tip presence.
Expand Down

0 comments on commit c675501

Please sign in to comment.