Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 4 additions & 18 deletions api/src/opentrons/protocol_api/core/engine/instrument.py
Original file line number Diff line number Diff line change
Expand Up @@ -392,12 +392,7 @@ def dispense(
)
)

if isinstance(location, (TrashBin, WasteChute)):
self._protocol_core.set_last_location(location=None, mount=self.get_mount())
else:
self._protocol_core.set_last_location(
location=location, mount=self.get_mount()
)
self._protocol_core.set_last_location(location=location, mount=self.get_mount())

def blow_out(
self,
Expand Down Expand Up @@ -473,12 +468,7 @@ def blow_out(
)
)

if isinstance(location, (TrashBin, WasteChute)):
self._protocol_core.set_last_location(location=None, mount=self.get_mount())
else:
self._protocol_core.set_last_location(
location=location, mount=self.get_mount()
)
self._protocol_core.set_last_location(location=location, mount=self.get_mount())

def touch_tip(
self,
Expand Down Expand Up @@ -806,12 +796,8 @@ def move_to(
speed=speed,
)
)
if isinstance(location, (TrashBin, WasteChute)):
self._protocol_core.set_last_location(location=None, mount=self.get_mount())
else:
self._protocol_core.set_last_location(
location=location, mount=self.get_mount()
)

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

def resin_tip_seal(
self, location: Location, well_core: WellCore, in_place: Optional[bool] = False
Expand Down
6 changes: 3 additions & 3 deletions api/src/opentrons/protocol_api/core/engine/protocol.py
Original file line number Diff line number Diff line change
Expand Up @@ -112,7 +112,7 @@ def __init__(
self._engine_client = engine_client
self._api_version = api_version
self._sync_hardware = sync_hardware
self._last_location: Optional[Location] = None
self._last_location: Optional[Union[Location, TrashBin, WasteChute]] = None
self._last_mount: Optional[Mount] = None
self._labware_cores_by_id: Dict[str, LabwareCore] = {}
self._module_cores_by_id: Dict[
Expand Down Expand Up @@ -892,7 +892,7 @@ def door_closed(self) -> bool:
def get_last_location(
self,
mount: Optional[Mount] = None,
) -> Optional[Location]:
) -> Optional[Union[Location, TrashBin, WasteChute]]:
"""Get the last accessed location."""
if mount is None or mount == self._last_mount:
return self._last_location
Expand All @@ -901,7 +901,7 @@ def get_last_location(

def set_last_location(
self,
location: Optional[Location],
location: Optional[Union[Location, TrashBin, WasteChute]],
mount: Optional[Mount] = None,
) -> None:
"""Set the last accessed location."""
Expand Down
2 changes: 1 addition & 1 deletion api/src/opentrons/protocol_api/core/protocol.py
Original file line number Diff line number Diff line change
Expand Up @@ -216,7 +216,7 @@ def door_closed(self) -> bool:
def get_last_location(
self,
mount: Optional[Mount] = None,
) -> Optional[Location]:
) -> Optional[Union[Location, TrashBin, WasteChute]]:
...

@abstractmethod
Expand Down
75 changes: 56 additions & 19 deletions api/src/opentrons/protocol_api/instrument_context.py
Original file line number Diff line number Diff line change
Expand Up @@ -258,7 +258,7 @@ def aspirate(
"knows where it is."
) from e

if isinstance(target, (TrashBin, WasteChute)):
if isinstance(target, validation.DisposalTarget):
raise ValueError(
"Trash Bin and Waste Chute are not acceptable location parameters for Aspirate commands."
)
Expand Down Expand Up @@ -399,6 +399,10 @@ def dispense(

.. versionchanged:: 2.17
Behavior of the ``volume`` parameter.

.. versionchanged:: 2.24
``location`` is no longer required if the pipette just moved to, dispensed, or blew out
into a trash bin or waste chute.
"""
if self.api_version < APIVersion(2, 15) and push_out:
raise APIVersionError(
Expand Down Expand Up @@ -432,24 +436,24 @@ def dispense(

flow_rate = self._core.get_dispense_flow_rate(rate)

if isinstance(target, (TrashBin, WasteChute)):
if isinstance(target, validation.DisposalTarget):
with publisher.publish_context(
broker=self.broker,
command=cmds.dispense_in_disposal_location(
instrument=self,
volume=c_vol,
location=target,
location=target.location,
rate=rate,
flow_rate=flow_rate,
),
):
self._core.dispense(
volume=c_vol,
rate=rate,
location=target,
location=target.location,
well_core=None,
flow_rate=flow_rate,
in_place=False,
in_place=target.in_place,
push_out=push_out,
meniscus_tracking=None,
)
Expand Down Expand Up @@ -656,6 +660,10 @@ def blow_out(
without first calling a method that takes a location, like
:py:meth:`.aspirate` or :py:meth:`dispense`.
:returns: This instance.

.. versionchanged:: 2.24
``location`` is no longer required if the pipette just moved to, dispensed, or blew out
into a trash bin or waste chute.
"""
well: Optional[labware.Well] = None
move_to_location: types.Location
Expand Down Expand Up @@ -696,17 +704,17 @@ def blow_out(
well = target.well
elif isinstance(target, validation.PointTarget):
move_to_location = target.location
elif isinstance(target, (TrashBin, WasteChute)):
elif isinstance(target, validation.DisposalTarget):
with publisher.publish_context(
broker=self.broker,
command=cmds.blow_out_in_disposal_location(
instrument=self, location=target
instrument=self, location=target.location
),
):
self._core.blow_out(
location=target,
location=target.location,
well_core=None,
in_place=False,
in_place=target.in_place,
)
return self

Expand Down Expand Up @@ -793,8 +801,12 @@ def touch_tip( # noqa: C901
# If location is a valid well, move to the well first
if location is None:
last_location = self._protocol_core.get_last_location()
if not last_location:
raise RuntimeError("No valid current location cache present")
if last_location is None or isinstance(
last_location, (TrashBin, WasteChute)
):
raise RuntimeError(
f"Cached location of {last_location} is not valid for touch tip."
)
parent_labware, well = last_location.labware.get_parent_labware_and_well()
if not well or not parent_labware:
raise RuntimeError(
Expand Down Expand Up @@ -896,10 +908,10 @@ def air_gap( # noqa: C901

.. versionchanged:: 2.22
No longer implemented as an aspirate.

.. versionchanged:: 2.24
Adds the ``rate`` and ``flow_rate`` parameter. You can only define one or the other. If
both are unspecified then ``rate`` is by default set to 1.0.
Can air gap over a trash bin or waste chute.
"""
if not self._core.has_tip():
raise UnexpectedTipRemovalError("air_gap", self.name, self.mount)
Expand All @@ -923,10 +935,23 @@ def air_gap( # noqa: C901

if height is None:
height = 5
loc = self._protocol_core.get_last_location()
if not loc or not loc.labware.is_well:
raise RuntimeError("No previous Well cached to perform air gap")
target = loc.labware.as_well().top(height)
last_location = self._protocol_core.get_last_location()
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Did you mean to use _get_last_location_by_api_version() that you defined below?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If we were to make this change we'd call get_last_location using a mount argument, which we have not been doing here. That may be more technically correct, but this is what has been in the code and changing it without a version check would potentially break or change the behavior of older protocols, and changing that is slightly out of scope for this PR.

if self.api_version < APIVersion(2, 24) and isinstance(
last_location, (TrashBin, WasteChute)
):
last_location = None
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Did you mean to use _get_last_location_by_api_version() that you defined below?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

See above comment.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(Sorry, I don't know why Github sent out a duplicate comment here!)

if last_location is None or (
isinstance(last_location, types.Location)
and not last_location.labware.is_well
):
raise RuntimeError(
f"Cached location of {last_location} is not valid for air gap."
)
target: Union[types.Location, TrashBin, WasteChute]
if isinstance(last_location, types.Location):
target = last_location.labware.as_well().top(height)
else:
target = last_location.top(height)
self.move_to(target, publish=False)
if self.api_version >= _AIR_GAP_TRACKING_ADDED_IN:
self._core.prepare_to_aspirate()
Expand Down Expand Up @@ -2583,14 +2608,22 @@ def well_bottom_clearance(self) -> "Clearances":
"""
return self._well_bottom_clearances

def _get_last_location_by_api_version(self) -> Optional[types.Location]:
def _get_last_location_by_api_version(
self,
) -> Optional[Union[types.Location, TrashBin, WasteChute]]:
"""Get the last location accessed by this pipette, if any.

In pre-engine Protocol API versions, this call omits the pipette mount.
Between 2.14 (first engine PAPI version) and 2.23 this only returns None or a Location object.
This is to preserve pre-existing, potentially buggy behavior.
"""
if self._api_version >= ENGINE_CORE_API_VERSION:
if self._api_version >= APIVersion(2, 24):
return self._protocol_core.get_last_location(mount=self._core.get_mount())
elif self._api_version >= ENGINE_CORE_API_VERSION:
last_location = self._protocol_core.get_last_location(
mount=self._core.get_mount()
)
return last_location if isinstance(last_location, types.Location) else None
else:
return self._protocol_core.get_last_location()
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What does this return (before version 2.14)? Could this be a trashbin/wastechute?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This calls get_last_location without a mount argument. We can get away without checking the type because that core will be the legacy core and that will not return a trash bin or waste chute.


Expand Down Expand Up @@ -2646,7 +2679,11 @@ def configure_for_volume(self, volume: float) -> None:
actual_value=str(volume),
)
last_location = self._get_last_location_by_api_version()
if last_location and isinstance(last_location.labware, labware.Well):
if (
last_location
and isinstance(last_location, types.Location)
and isinstance(last_location.labware, labware.Well)
):
self.move_to(last_location.labware.top())
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So the pipette should not move during configure_for_volume() if the pipette is in the trash/wastechute?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The main concern for configure_for_volume is accidentally aspirating liquid if the configuration is done while the tip is submerged in liquid. This is not a valid concern if the tip is over a trash bin or waste chute.

self._core.configure_for_volume(volume)

Expand Down
13 changes: 10 additions & 3 deletions api/src/opentrons/protocol_api/protocol_context.py
Original file line number Diff line number Diff line change
Expand Up @@ -1190,9 +1190,16 @@ def home(self) -> None:
self._core.home()

@property
def location_cache(self) -> Optional[Location]:
"""The cache used by the robot to determine where it last was."""
return self._core.get_last_location()
def location_cache(self) -> Optional[Union[Location, TrashBin, WasteChute]]:
"""The cache used by the robot to determine where it last was.

.. versionchanged:: 2.24
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We've hidden this reference entry, but this is a good addition for internal documentation.

Can return a ``TrashBin`` or ``WasteChute`` object.
"""
last_loc = self._core.get_last_location()
if isinstance(last_loc, Location) or self._api_version >= APIVersion(2, 24):
return last_loc
return None

@location_cache.setter
def location_cache(self, loc: Optional[Location]) -> None:
Expand Down
21 changes: 14 additions & 7 deletions api/src/opentrons/protocol_api/validation.py
Original file line number Diff line number Diff line change
Expand Up @@ -545,6 +545,11 @@ class PointTarget(NamedTuple):
in_place: bool


class DisposalTarget(NamedTuple):
location: Union[TrashBin, WasteChute]
in_place: bool


class NoLocationError(ValueError):
"""Error representing that no location was supplied."""

Expand All @@ -553,12 +558,12 @@ class LocationTypeError(TypeError):
"""Error representing that the location supplied is of different expected type."""


ValidTarget = Union[WellTarget, PointTarget, TrashBin, WasteChute]
ValidTarget = Union[WellTarget, PointTarget, DisposalTarget]


def validate_location(
location: Union[Location, Well, TrashBin, WasteChute, None],
last_location: Optional[Location],
location: Optional[Union[Location, Well, TrashBin, WasteChute]],
last_location: Optional[Union[Location, TrashBin, WasteChute]],
) -> ValidTarget:
"""Validate a given location for a liquid handling command.

Expand All @@ -569,9 +574,11 @@ def validate_location(
Returns:
A `WellTarget` if the input location represents a well.
A `PointTarget` if the input location is an x, y, z coordinate.
A `TrashBin` if the input location is a trash bin
A `WasteChute` if the input location is a waste chute

Raises:
NoLocationError: The is no input location and no cached loaction.
NoLocationError: There is no input location and no cached location.
LocationTypeError: The location supplied is of unexpected type.
"""
from .labware import Well
Expand All @@ -586,11 +593,11 @@ def validate_location(
f"location should be a Well, Location, TrashBin or WasteChute, but it is {location}"
)

if isinstance(target_location, (TrashBin, WasteChute)):
return target_location

in_place = target_location == last_location

if isinstance(target_location, (TrashBin, WasteChute)):
return DisposalTarget(location=target_location, in_place=in_place)

if isinstance(target_location, Well):
return WellTarget(well=target_location, location=None, in_place=in_place)

Expand Down
Loading