Skip to content

Commit

Permalink
aspirate while tracking hw testing stuff
Browse files Browse the repository at this point in the history
  • Loading branch information
@caila-marashaj
caila-marashaj committed Jan 15, 2025
1 parent 022e748 commit 8803b47
Show file tree
Hide file tree
Showing 34 changed files with 255 additions and 216 deletions.
105 changes: 71 additions & 34 deletions api/src/opentrons/protocol_api/core/engine/instrument.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,13 @@
from __future__ import annotations

from typing import Optional, TYPE_CHECKING, cast, Union, List, Tuple
from opentrons.types import Location, Mount, NozzleConfigurationType, NozzleMapInterface
from opentrons.types import (
Location,
Mount,
NozzleConfigurationType,
NozzleMapInterface,
MeniscusTracking,
)
from opentrons.hardware_control import SyncHardwareAPI
from opentrons.hardware_control.dev_types import PipetteDict
from opentrons.protocols.api_support.util import FlowRates, find_value_for_api_version
Expand Down Expand Up @@ -134,8 +140,7 @@ def aspirate(
rate: float,
flow_rate: float,
in_place: bool,
is_meniscus: Optional[bool] = None,
is_tracking: Optional[bool] = False,
meniscus_tracking: Optional[MeniscusTracking] = None,
) -> None:
"""Aspirate a given volume of liquid from the specified location.
Args:
Expand All @@ -145,8 +150,7 @@ def aspirate(
rate: Not used in this core.
flow_rate: The flow rate in µL/s to aspirate at.
in_place: whether this is a in-place command.
is_meniscus: whether the aspirate location specified is relative to a liquid meniscus.
is_tracking: whether the z motor is to move with the liquid meniscus.
meniscus_tracking: Optional data about where to aspirate from.
"""
if well_core is None:
if not in_place:
Expand Down Expand Up @@ -176,28 +180,44 @@ def aspirate(
labware_id=labware_id,
well_name=well_name,
absolute_point=location.point,
is_meniscus=is_meniscus,
meniscus_tracking=meniscus_tracking,
)
if well_location.origin == WellOrigin.MENISCUS:
well_location.volumeOffset = "operationVolume"
dynamic_liquid_tracking = False
# caila bookmark
if meniscus_tracking:
if meniscus_tracking.target == "end":
well_location.volumeOffset = "operationVolume"
elif meniscus_tracking.target == "dynamic_meniscus":
dynamic_liquid_tracking = True
pipette_movement_conflict.check_safe_for_pipette_movement(
engine_state=self._engine_client.state,
pipette_id=self._pipette_id,
labware_id=labware_id,
well_name=well_name,
well_location=well_location,
)
self._engine_client.execute_command(
cmd.AspirateParams(
pipetteId=self._pipette_id,
labwareId=labware_id,
wellName=well_name,
wellLocation=well_location,
volume=volume,
flowRate=flow_rate,
is_tracking=is_tracking if is_tracking else False,
if dynamic_liquid_tracking:
self._engine_client.execute_command(
cmd.AspirateWhileTrackingParams(
pipetteId=self._pipette_id,
labwareId=labware_id,
wellName=well_name,
wellLocation=well_location,
volume=volume,
flowRate=flow_rate,
)
)
else:
self._engine_client.execute_command(
cmd.AspirateParams(
pipetteId=self._pipette_id,
labwareId=labware_id,
wellName=well_name,
wellLocation=well_location,
volume=volume,
flowRate=flow_rate,
)
)
)

self._protocol_core.set_last_location(location=location, mount=self.get_mount())

Expand All @@ -210,8 +230,7 @@ def dispense(
flow_rate: float,
in_place: bool,
push_out: Optional[float],
is_meniscus: Optional[bool] = None,
is_tracking: Optional[bool] = False,
meniscus_tracking: Optional[MeniscusTracking] = None,
) -> None:
"""Dispense a given volume of liquid into the specified location.
Args:
Expand All @@ -222,9 +241,9 @@ def dispense(
flow_rate: The flow rate in µL/s to dispense at.
in_place: whether this is a in-place command.
push_out: The amount to push the plunger below bottom position.
is_meniscus: whether the aspirate location specified is relative to a liquid meniscus.
is_tracking: whether the z motor is to move with the liquid meniscus.
meniscus_tracking: Optional data about where to dispense from.
"""
# raise ValueError(f"well location = {location}")
if self._protocol_core.api_version < _DISPENSE_VOLUME_VALIDATION_ADDED_IN:
# In older API versions, when you try to dispense more than you can,
# it gets clamped.
Expand Down Expand Up @@ -273,27 +292,45 @@ def dispense(
labware_id=labware_id,
well_name=well_name,
absolute_point=location.point,
is_meniscus=is_meniscus,
meniscus_tracking=meniscus_tracking,
)
dynamic_liquid_tracking = False
if meniscus_tracking:
if meniscus_tracking.target == "end":
well_location.volumeOffset = "operationVolume"
elif meniscus_tracking.target == "dynamic_meniscus":
dynamic_liquid_tracking = True
pipette_movement_conflict.check_safe_for_pipette_movement(
engine_state=self._engine_client.state,
pipette_id=self._pipette_id,
labware_id=labware_id,
well_name=well_name,
well_location=well_location,
)
self._engine_client.execute_command(
cmd.DispenseParams(
pipetteId=self._pipette_id,
labwareId=labware_id,
wellName=well_name,
wellLocation=well_location,
volume=volume,
flowRate=flow_rate,
pushOut=push_out,
is_tracking=is_tracking if is_tracking else False,
if dynamic_liquid_tracking:
self._engine_client.execute_command(
cmd.DispenseWhileTrackingParams(
pipetteId=self._pipette_id,
labwareId=labware_id,
wellName=well_name,
wellLocation=well_location,
volume=volume,
flowRate=flow_rate,
pushOut=push_out,
)
)
else:
self._engine_client.execute_command(
cmd.DispenseParams(
pipetteId=self._pipette_id,
labwareId=labware_id,
wellName=well_name,
wellLocation=well_location,
volume=volume,
flowRate=flow_rate,
pushOut=push_out,
)
)
)

if isinstance(location, (TrashBin, WasteChute)):
self._protocol_core.set_last_location(location=None, mount=self.get_mount())
Expand Down
12 changes: 4 additions & 8 deletions api/src/opentrons/protocol_api/core/instrument.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -41,8 +41,7 @@ def aspirate(
rate: float,
flow_rate: float,
in_place: bool,
is_meniscus: Optional[bool] = None,
is_tracking: Optional[bool] = False,
meniscus_tracking: Optional[types.MeniscusTracking] = None,
) -> None:
"""Aspirate a given volume of liquid from the specified location.
Args:
Expand All @@ -52,8 +51,7 @@ def aspirate(
rate: The rate for how quickly to aspirate.
flow_rate: The flow rate in µL/s to aspirate at.
in_place: Whether this is in-place.
is_meniscus: whether the aspirate location specified is relative to a liquid meniscus.
is_tracking: whether the z motor is to move with the liquid meniscus.
meniscus_tracking: Optional data about where to aspirate from.
"""
...

Expand All @@ -67,8 +65,7 @@ def dispense(
flow_rate: float,
in_place: bool,
push_out: Optional[float],
is_meniscus: Optional[bool] = None,
is_tracking: Optional[bool] = None,
meniscus_tracking: Optional[types.MeniscusTracking] = None,
) -> None:
"""Dispense a given volume of liquid into the specified location.
Args:
Expand All @@ -79,8 +76,7 @@ def dispense(
flow_rate: The flow rate in µL/s to dispense at.
in_place: Whether this is in-place.
push_out: The amount to push the plunger below bottom position.
is_meniscus: whether the aspirate location specified is relative to a liquid meniscus.
is_tracking: whether the z motor is to move with the liquid meniscus.
meniscus_tracking: Optional data about where to dispense from.
"""
...

Expand Down
12 changes: 4 additions & 8 deletions api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -84,8 +84,7 @@ def aspirate(
rate: float,
flow_rate: float,
in_place: bool,
is_meniscus: Optional[bool] = None,
is_tracking: Optional[bool] = None,
meniscus_tracking: Optional[types.MeniscusTracking] = None,
) -> None:
"""Aspirate a given volume of liquid from the specified location.
Args:
Expand All @@ -95,8 +94,7 @@ def aspirate(
rate: The rate in µL/s to aspirate at.
flow_rate: Not used in this core.
in_place: Whether we should move_to location.
is_meniscus: whether the aspirate location specified is relative to a liquid meniscus.
is_tracking: whether the z motor is to move with the liquid meniscus.
meniscus_tracking: Optional data about where to aspirate from.
"""
if self.get_current_volume() == 0:
# Make sure we're at the top of the labware and clear of any
Expand Down Expand Up @@ -130,8 +128,7 @@ def dispense(
flow_rate: float,
in_place: bool,
push_out: Optional[float],
is_meniscus: Optional[bool] = None,
is_tracking: Optional[bool] = None,
meniscus_tracking: Optional[types.MeniscusTracking] = None,
) -> None:
"""Dispense a given volume of liquid into the specified location.
Args:
Expand All @@ -142,8 +139,7 @@ def dispense(
flow_rate: Not used in this core.
in_place: Whether we should move_to location.
push_out: The amount to push the plunger below bottom position.
is_meniscus: whether the aspirate location specified is relative to a liquid meniscus.
is_tracking: whether the z motor is to move with the liquid meniscus.
meniscus_tracking: Optional data about where to dispense from.
"""
if isinstance(location, (TrashBin, WasteChute)):
raise APIVersionError(
Expand Down
4 changes: 0 additions & 4 deletions api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -95,8 +95,6 @@ def aspirate(
rate: float,
flow_rate: float,
in_place: bool,
is_meniscus: Optional[bool] = None,
is_tracking: Optional[bool] = None,
) -> None:
if self.get_current_volume() == 0:
# Make sure we're at the top of the labware and clear of any
Expand Down Expand Up @@ -138,8 +136,6 @@ def dispense(
flow_rate: float,
in_place: bool,
push_out: Optional[float],
is_meniscus: Optional[bool] = None,
is_tracking: Optional[bool] = None,
) -> None:
if isinstance(location, (TrashBin, WasteChute)):
raise APIVersionError(
Expand Down
Loading

0 comments on commit 8803b47

Please sign in to comment.