feat(api): implement labware.set_offset in 2.18 (#14940)

The python protocol api's labware.set_offset() command lets you set a
labware position check offset programmatically. This is useful in
protocols run outside the app (i.e. through Jupyter, through
opentrons_execute).

When we moved to the protocol engine, we didn't support running those
protocols outside the app, and therefore we didn't implement this method
on those new API versions.

Now, we allow that again, so implement that method again.

This relies on dispatching AddLabwareOffset actions at arbitrary times,
because it will set an offset for the location in which the labware is
_currently_ present. This allows the user to set offsets without having
to wrangle with the engine's internal definitions of where something is,
at the cost of the user having to spread their offset calls around their
protocol if they have protocols that load things in one place and later
move them to another.

## Warning
~This won't work the way you think it will right now. Labware offsets
are applied to labware instances by the `LoadLabware` and `MoveLabware`
commands. So if you do an `AddLabwareOffsetAction` after a
`LoadLabware`, the offset will never apply to the labware. We need to
add a new `ReloadLabware` command, but I'm going to do that in a
separate PR and then put this PR on top of that one.~
Fully implemented yay

Closes RSQ-29

api/src/opentrons/protocol_api/core/engine/__init__.py

@@ -10,6 +10,7 @@
 from .well import WellCore
 
 ENGINE_CORE_API_VERSION: Final = APIVersion(2, 14)
+SET_OFFSET_RESTORED_API_VERSION: Final = APIVersion(2, 18)
 
 __all__ = [
     "ENGINE_CORE_API_VERSION",

api/src/opentrons/protocol_api/core/engine/labware.py

@@ -10,9 +10,14 @@
 
 from opentrons.protocol_engine.errors import LabwareNotOnDeckError, ModuleNotOnDeckError
 from opentrons.protocol_engine.clients import SyncClient as ProtocolEngineClient
+from opentrons.protocol_engine.types import (
+    LabwareOffsetCreate,
+    LabwareOffsetVector,
+)
 from opentrons.types import DeckSlotName, Point
 from opentrons.hardware_control.nozzle_manager import NozzleMap
 
+
 from ..labware import AbstractLabware, LabwareLoadParams
 from .well import WellCore
 
@@ -92,8 +97,28 @@ def get_quirks(self) -> List[str]:
         return self._definition.parameters.quirks or []
 
     def set_calibration(self, delta: Point) -> None:
-        raise NotImplementedError(
-            "Setting a labware's calibration after it's been loaded is not supported."
+        """Add a labware offset for this labware at its current location.
+
+        This will override any previous labware offsets for this definition URI and location,
+        even if the other labware offset was for a different specific labware instance.
+        """
+        offset_location = self._engine_client.state.geometry.get_offset_location(
+            self._labware_id
+        )
+        if not offset_location:
+            raise LabwareNotOnDeckError(
+                message=f"Cannot set offset for {self.get_name()} as it is not currently in a deck slot.",
+                details={"kind": "labware-not-in-slot"},
+            )
+
+        request = LabwareOffsetCreate.construct(
+            definitionUri=self.get_uri(),
+            location=offset_location,
+            vector=LabwareOffsetVector(x=delta.x, y=delta.y, z=delta.z),
+        )
+        self._engine_client.add_labware_offset(request)
+        self._engine_client.reload_labware(
+            labware_id=self._labware_id,
         )
 
     def get_calibrated_offset(self) -> Point:

api/src/opentrons/protocol_api/labware.py

@@ -35,7 +35,7 @@
 from ._liquid import Liquid
 from ._types import OffDeckType
 from .core import well_grid
-from .core.engine import ENGINE_CORE_API_VERSION
+from .core.engine import ENGINE_CORE_API_VERSION, SET_OFFSET_RESTORED_API_VERSION
 from .core.labware import AbstractLabware
 from .core.module import AbstractModuleCore
 from .core.core_map import LoadedCoreMap
@@ -594,16 +594,13 @@ def set_offset(self, x: float, y: float, z: float) -> None:
             Instead, use Labware Position Check in the app or on the touchscreen.
 
         """
-        if self._api_version >= ENGINE_CORE_API_VERSION:
-            # TODO(mm, 2023-02-13): See Jira RCORE-535.
-            #
-            # Until that issue is resolved, the only way to simulate or run a
-            # >=ENGINE_CORE_API_VERSION protocol is through the Opentrons App.
-            # Therefore, in >=ENGINE_CORE_API_VERSION protocols,
-            # there's no legitimate way to use this method.
+        if (
+            self._api_version >= ENGINE_CORE_API_VERSION
+            and self._api_version < SET_OFFSET_RESTORED_API_VERSION
+        ):
             raise APIVersionError(
-                "Labware.set_offset() is not supported when apiLevel is 2.14 or higher."
-                " Use a lower apiLevel"
+                "Labware.set_offset() is not supported when apiLevel is 2.14, 2.15, 2.16, or 2.17."
+                " Use apilevel 2.13 or below, or 2.18 or above to set offset,"
                 " or use the Opentrons App's Labware Position Check."
             )
         else:

api/src/opentrons/protocol_engine/clients/sync_client.py

@@ -27,6 +27,7 @@
     Liquid,
     NozzleLayoutConfigurationType,
     AddressableOffsetVector,
+    LabwareOffsetCreate,
 )
 from .transports import ChildThreadTransport
 
@@ -92,6 +93,10 @@ def reset_tips(self, labware_id: str) -> None:
             labware_id=labware_id,
         )
 
+    def add_labware_offset(self, request: LabwareOffsetCreate) -> None:
+        """Add a labware offset."""
+        self._transport.call_method("add_labware_offset", request=request)
+
     def set_pipette_movement_speed(
         self, pipette_id: str, speed: Optional[float]
     ) -> None:

api/src/opentrons/protocol_engine/state/geometry.py

@@ -42,6 +42,7 @@
     AddressableAreaLocation,
     AddressableOffsetVector,
     StagingSlotLocation,
+    LabwareOffsetLocation,
 )
 from .config import Config
 from .labware import LabwareView
@@ -1090,3 +1091,48 @@ def _labware_gripper_offsets(
         return slot_based_offset or self._labware.get_labware_gripper_offsets(
             labware_id=labware_id, slot_name=None
         )
+
+    def get_offset_location(self, labware_id: str) -> Optional[LabwareOffsetLocation]:
+        """Provide the LabwareOffsetLocation specifying the current position of the labware.
+
+        If the labware is in a location that cannot be specified by a LabwareOffsetLocation
+        (for instance, OFF_DECK) then return None.
+        """
+        parent_location = self._labware.get_location(labware_id)
+
+        if isinstance(parent_location, DeckSlotLocation):
+            return LabwareOffsetLocation(
+                slotName=parent_location.slotName, moduleModel=None, definitionUri=None
+            )
+        elif isinstance(parent_location, ModuleLocation):
+            module_model = self._modules.get_requested_model(parent_location.moduleId)
+            module_location = self._modules.get_location(parent_location.moduleId)
+            return LabwareOffsetLocation(
+                slotName=module_location.slotName,
+                moduleModel=module_model,
+                definitionUri=None,
+            )
+        elif isinstance(parent_location, OnLabwareLocation):
+            non_labware_parent_location = self._labware.get_parent_location(labware_id)
+
+            parent_uri = self._labware.get_definition_uri(parent_location.labwareId)
+            if isinstance(non_labware_parent_location, DeckSlotLocation):
+                return LabwareOffsetLocation(
+                    slotName=non_labware_parent_location.slotName,
+                    moduleModel=None,
+                    definitionUri=parent_uri,
+                )
+            elif isinstance(non_labware_parent_location, ModuleLocation):
+                module_model = self._modules.get_requested_model(
+                    non_labware_parent_location.moduleId
+                )
+                module_location = self._modules.get_location(
+                    non_labware_parent_location.moduleId
+                )
+                return LabwareOffsetLocation(
+                    slotName=module_location.slotName,
+                    moduleModel=module_model,
+                    definitionUri=parent_uri,
+                )
+
+        return None

api/tests/opentrons/protocol_api/__init__.py

@@ -1 +1,131 @@
 """Tests for opentrons.protocol_api."""
+from typing import List, overload, Optional
+
+from opentrons.protocols.api_support.types import APIVersion
+from opentrons.protocol_api import (
+    MAX_SUPPORTED_VERSION,
+    MIN_SUPPORTED_VERSION,
+    MIN_SUPPORTED_VERSION_FOR_FLEX,
+)
+
+
+def versions_at_or_above(from_version: APIVersion) -> List[APIVersion]:
+    """Get a list of versions >= the specified one."""
+    return versions_between(
+        low_inclusive_bound=from_version, high_inclusive_bound=MAX_SUPPORTED_VERSION
+    )
+
+
+def versions_at_or_below(
+    from_version: APIVersion, flex_only: bool = False
+) -> List[APIVersion]:
+    """Get a list of versions <= the specified one.
+
+    Since there are different minimum supported versions for Flex and OT-2, specify which you care about
+    with the second argument.
+    """
+    if flex_only:
+        return versions_between(
+            low_inclusive_bound=MIN_SUPPORTED_VERSION_FOR_FLEX,
+            high_inclusive_bound=from_version,
+        )
+    else:
+        return versions_between(
+            low_inclusive_bound=MIN_SUPPORTED_VERSION, high_inclusive_bound=from_version
+        )
+
+
+def versions_above(from_version: APIVersion) -> List[APIVersion]:
+    """Get a list of versions > the specified one."""
+    return versions_between(
+        low_exclusive_bound=from_version, high_inclusive_bound=MAX_SUPPORTED_VERSION
+    )
+
+
+def versions_below(from_version: APIVersion, flex_only: bool) -> List[APIVersion]:
+    """Get a list of versions < the specified one.
+
+    Since there are different minimum supported versions for Flex and OT-2, specify which you care about
+    with the second argument.
+    """
+    if flex_only:
+        return versions_between(
+            low_inclusive_bound=MIN_SUPPORTED_VERSION_FOR_FLEX,
+            high_exclusive_bound=from_version,
+        )
+    else:
+        return versions_between(
+            low_inclusive_bound=MIN_SUPPORTED_VERSION, high_exclusive_bound=from_version
+        )
+
+
+@overload
+def versions_between(
+    *,
+    low_inclusive_bound: APIVersion,
+    high_inclusive_bound: APIVersion,
+) -> List[APIVersion]:
+    ...
+
+
+@overload
+def versions_between(
+    *, low_inclusive_bound: APIVersion, high_exclusive_bound: APIVersion
+) -> List[APIVersion]:
+    ...
+
+
+@overload
+def versions_between(
+    *,
+    high_inclusive_bound: APIVersion,
+    low_exclusive_bound: APIVersion,
+) -> List[APIVersion]:
+    ...
+
+
+@overload
+def versions_between(
+    *, low_exclusive_bound: APIVersion, high_exclusive_bound: APIVersion
+) -> List[APIVersion]:
+    ...
+
+
+def versions_between(
+    low_inclusive_bound: Optional[APIVersion] = None,
+    high_inclusive_bound: Optional[APIVersion] = None,
+    low_exclusive_bound: Optional[APIVersion] = None,
+    high_exclusive_bound: Optional[APIVersion] = None,
+) -> List[APIVersion]:
+    """Build a list of versions based on exclusive and inclusive constraints."""
+    if low_inclusive_bound and high_inclusive_bound:
+        assert (
+            low_inclusive_bound.major == high_inclusive_bound.major
+        ), "You need to change this test when you add a new major version"
+        major = low_inclusive_bound.major
+        start = low_inclusive_bound.minor
+        stop = high_inclusive_bound.minor + 1
+    elif low_inclusive_bound and high_exclusive_bound:
+        assert (
+            low_inclusive_bound.major == high_exclusive_bound.major
+        ), "You need to change this test when you add a new major version"
+        major = low_inclusive_bound.major
+        start = low_inclusive_bound.minor
+        stop = high_exclusive_bound.minor
+    elif low_exclusive_bound and high_inclusive_bound:
+        assert (
+            low_exclusive_bound.major == high_inclusive_bound.major
+        ), "You need to change this test when you add a new major version"
+        major = low_exclusive_bound.major
+        start = low_exclusive_bound.minor + 1
+        stop = high_inclusive_bound.minor + 1
+    elif low_exclusive_bound and high_exclusive_bound:
+        assert (
+            low_exclusive_bound.major == high_exclusive_bound.major
+        ), "You need to change this test when you add a new major version"
+        major = low_exclusive_bound.major
+        start = low_exclusive_bound.minor + 1
+        stop = high_exclusive_bound.minor
+    else:
+        raise ValueError("You must specify one low bound and one high bound")
+    return [APIVersion(major, minor) for minor in range(start, stop)]