feat(api,robot-server,shared-data): Specify tip overlap by version (#…

…15323)

We need to have multiple versions of the tip overlap data for at least
flex pipettes so that we can ship updated overlap data without
inadvertently changing the behavior of protocols on stable API versions.
This PR accomplishes that.


# Details
## separate versioning
The data in shared-data has a version, but that version is separate from
versions in other parts of the stack. This is on purpose, because the
version has to be written in shared-data, and read in protocol engine. I
think it's gross to write API versions in shared data, but we've done it
before and could do it again; the real sticking point was the engine.
The engine, in modern protocols, loads the tip overlap and does all that
data management itself; it would need to select the appropriate version
from the dict. We really don't want API versions there.

Therefore, there's a separate simple versioning scheme that's a little
more like the alternate liquid handling function scheme that we haven't
used in years, but with an actual string key for the version instead of
implicit versioning by positioning in an array.

There will instead be a specific map of API versions to overlap versions
living inside the source of the engine core for the python protocol API
(that's the only place it needs to be - for why, read on).

## engine state management and data lifecycle
There were basically two ways to implement this. The first way is to
save all the data - save the full versioned tip overlap dictionary, and
save which version we've selected. Since I listed this first, you can
guess I didn't do this. Instead, what I did was add a version to the two
engine commands that load pipette data - `load_pipette` and
`configure_for_volume` - and then have the functions those commands call
pull the specified version out of the full config, which means that we
only have to save the specific version in state and state doesn't have
to change. This also means that we don't have to worry about data
storage migration and compatibility.

One huge note here: **the engine argument parsing gives you the MOST
RECENT tip overlap if you don't specify one**. That's in bold because
it's not the way we've implemented similar things. It's necessary
because we want PD protocols to get the new overlap values without
having to change anything.

## python api integration
The python API, however, really wants to keep the old values (until
you're on a certain version - we'll implement that when we actually add
the new values). We do that in a couple ways:
- the old pipette dictionary value that has the tip overlap dict is
still there, using "v0". That means that protocol API versions that use
that dict, which is all those not using the engine core, get the old
values consistently for free
- for protocol api versions that do use the engine core, for now we hard
specify v0.

## json protocol integration

New JSON protocols, any that directly create engine commands rather than
being mapped through python protocol API values, will get the new
overlaps because they won't specify the parameter to the load pipette or
configure for volume engine commands. Older JSON protocols that do adapt
the python protocol API commands will get the python behavior.

# Testing
- This PR gets regression tested - it doesn't really contain any new
features, because it doesn't really contain any new data.
- OT-2 protocol behavior should be adequately covered by the g-code
testing
  - OT-2 calibration should probably be run real quick
- Flex behavior should be adequately tested by running LPC and a
protocol - this applies to all protocols and can be easily tested.
- Flex calibration and LPC doesn't use tips and therefore isn't affected


Closes EXEC-451

api/src/opentrons/hardware_control/dev_types.py

@@ -84,6 +84,7 @@ class PipetteDict(InstrumentDict):
     tip_length: float
     working_volume: float
     tip_overlap: Dict[str, float]
+    versioned_tip_overlap: Dict[str, Dict[str, float]]
     available_volume: float
     return_tip_height: float
     default_aspirate_flow_rates: Dict[str, float]

api/src/opentrons/hardware_control/instruments/ot2/pipette.py

@@ -148,7 +148,7 @@ def __init__(
             self._active_tip_settings.default_blowout_flowrate.default
         )
 
-        self._tip_overlap_lookup = self._liquid_class.tip_overlap_dictionary
+        self._tip_overlap_lookup = self._liquid_class.versioned_tip_overlap_dictionary
 
         if use_old_aspiration_functions:
             self._pipetting_function_version = PIPETTING_FUNCTION_FALLBACK_VERSION
@@ -216,7 +216,7 @@ def pipette_offset(self) -> PipetteOffsetByPipetteMount:
         return self._pipette_offset
 
     @property
-    def tip_overlap(self) -> Dict[str, float]:
+    def tip_overlap(self) -> Dict[str, Dict[str, float]]:
         return self._tip_overlap_lookup
 
     @property
@@ -290,7 +290,7 @@ def reset_state(self) -> None:
             self.active_tip_settings.default_blowout_flowrate.default
         )
 
-        self._tip_overlap_lookup = self.liquid_class.tip_overlap_dictionary
+        self._tip_overlap_lookup = self.liquid_class.versioned_tip_overlap_dictionary
         self._nozzle_manager = (
             nozzle_manager.NozzleConfigurationManager.build_from_config(self._config)
         )
@@ -571,7 +571,8 @@ def as_dict(self) -> "Pipette.DictType":
                 "default_dispense_flow_rates": self.dispense_flow_rates_lookup,
                 "tip_length": self.current_tip_length,
                 "return_tip_height": self.active_tip_settings.default_return_tip_height,
-                "tip_overlap": self.tip_overlap,
+                "tip_overlap": self.tip_overlap["v0"],
+                "versioned_tip_overlap": self.tip_overlap,
                 "back_compat_names": self._config.pipette_backcompat_names,
                 "supported_tips": self.liquid_class.supported_tips,
             }

api/src/opentrons/hardware_control/instruments/ot2/pipette_handler.py

@@ -212,6 +212,7 @@ def get_attached_instrument(self, mount: MountType) -> PipetteDict:
                 "blow_out_flow_rate",
                 "working_volume",
                 "tip_overlap",
+                "versioned_tip_overlap",
                 "available_volume",
                 "return_tip_height",
                 "default_aspirate_flow_rates",

api/src/opentrons/hardware_control/instruments/ot3/pipette.py

@@ -133,7 +133,7 @@ def __init__(
         )
         self._flow_acceleration = self._active_tip_settings.default_flow_acceleration
 
-        self._tip_overlap_lookup = self._liquid_class.tip_overlap_dictionary
+        self._tip_overlap_lookup = self._liquid_class.versioned_tip_overlap_dictionary
 
         if use_old_aspiration_functions:
             self._pipetting_function_version = PIPETTING_FUNCTION_FALLBACK_VERSION
@@ -161,7 +161,7 @@ def backlash_distance(self) -> float:
         return self._backlash_distance
 
     @property
-    def tip_overlap(self) -> Dict[str, float]:
+    def tip_overlap(self) -> Dict[str, Dict[str, float]]:
         return self._tip_overlap_lookup
 
     @property
@@ -254,7 +254,7 @@ def reset_state(self) -> None:
         )
         self._flow_acceleration = self._active_tip_settings.default_flow_acceleration
 
-        self._tip_overlap_lookup = self.liquid_class.tip_overlap_dictionary
+        self._tip_overlap_lookup = self.liquid_class.versioned_tip_overlap_dictionary
         self._nozzle_manager = (
             nozzle_manager.NozzleConfigurationManager.build_from_config(self._config)
         )
@@ -560,7 +560,8 @@ def as_dict(self) -> "Pipette.DictType":
                 "default_flow_acceleration": self.active_tip_settings.default_flow_acceleration,
                 "tip_length": self.current_tip_length,
                 "return_tip_height": self.active_tip_settings.default_return_tip_height,
-                "tip_overlap": self.tip_overlap,
+                "tip_overlap": self.tip_overlap["v0"],
+                "versioned_tip_overlap": self.tip_overlap,
                 "back_compat_names": self._config.pipette_backcompat_names,
                 "supported_tips": self.liquid_class.supported_tips,
             }
@@ -655,7 +656,7 @@ def set_tip_type(self, tip_type: pip_types.PipetteTipType) -> None:
         self._flow_acceleration = self._active_tip_settings.default_flow_acceleration
 
         self._fallback_tip_length = self._active_tip_settings.default_tip_length
-        self._tip_overlap_lookup = self.liquid_class.tip_overlap_dictionary
+        self._tip_overlap_lookup = self.liquid_class.versioned_tip_overlap_dictionary
         self._working_volume = min(tip_type.value, self.liquid_class.max_volume)
 
     def get_pick_up_configuration_for_tip_count(

api/src/opentrons/hardware_control/instruments/ot3/pipette_handler.py

@@ -228,6 +228,7 @@ def get_attached_instrument(self, mount: OT3Mount) -> PipetteDict:
                 "blow_out_flow_rate",
                 "working_volume",
                 "tip_overlap",
+                "versioned_tip_overlap",
                 "available_volume",
                 "return_tip_height",
                 "default_aspirate_flow_rates",

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

@@ -745,7 +745,7 @@ def set_flow_rate(
 
     def configure_for_volume(self, volume: float) -> None:
         self._engine_client.configure_for_volume(
-            pipette_id=self._pipette_id, volume=volume
+            pipette_id=self._pipette_id, volume=volume, tip_overlap_version="v0"
         )
 
     def prepare_to_aspirate(self) -> None:

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

@@ -498,7 +498,9 @@ def load_instrument(
             An instrument core configured to use the requested instrument.
         """
         engine_mount = MountType[mount.name]
-        load_result = self._engine_client.load_pipette(instrument_name, engine_mount)
+        load_result = self._engine_client.load_pipette(
+            instrument_name, engine_mount, tip_overlap_version="v0"
+        )
 
         return InstrumentCore(
             pipette_id=load_result.pipetteId,

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

@@ -172,10 +172,15 @@ def load_pipette(
         self,
         pipette_name: PipetteNameType,
         mount: MountType,
+        tip_overlap_version: Optional[str] = None,
     ) -> commands.LoadPipetteResult:
         """Execute a LoadPipette command and return the result."""
         request = commands.LoadPipetteCreate(
-            params=commands.LoadPipetteParams(mount=mount, pipetteName=pipette_name)
+            params=commands.LoadPipetteParams(
+                mount=mount,
+                pipetteName=pipette_name,
+                tipOverlapNotAfterVersion=tip_overlap_version,
+            )
         )
         result = self._transport.execute_command(request=request)
 
@@ -376,12 +381,14 @@ def drop_tip_in_place(
         return cast(commands.DropTipInPlaceResult, result)
 
     def configure_for_volume(
-        self, pipette_id: str, volume: float
+        self, pipette_id: str, volume: float, tip_overlap_version: Optional[str] = None
     ) -> commands.ConfigureForVolumeResult:
         """Execute a ConfigureForVolume command."""
         request = commands.ConfigureForVolumeCreate(
             params=commands.ConfigureForVolumeParams(
-                pipetteId=pipette_id, volume=volume
+                pipetteId=pipette_id,
+                volume=volume,
+                tipOverlapNotAfterVersion=tip_overlap_version,
             )
         )
         result = self._transport.execute_command(request=request)

api/src/opentrons/protocol_engine/commands/configure_for_volume.py

@@ -28,6 +28,13 @@ class ConfigureForVolumeParams(PipetteIdMixin):
         "than a pipette-specific maximum volume.",
         ge=0,
     )
+    tipOverlapNotAfterVersion: Optional[str] = Field(
+        None,
+        description="A version of tip overlap data to not exceed. The highest-versioned "
+        "tip overlap data that does not exceed this version will be used. Versions are "
+        "expressed as vN where N is an integer, counting up from v0. If None, the current "
+        "highest version will be used.",
+    )
 
 
 class ConfigureForVolumePrivateResult(PipetteConfigUpdateResultMixin):
@@ -61,6 +68,7 @@ async def execute(
         pipette_result = await self._equipment.configure_for_volume(
             pipette_id=params.pipetteId,
             volume=params.volume,
+            tip_overlap_version=params.tipOverlapNotAfterVersion,
         )
 
         return ConfigureForVolumeResult(), ConfigureForVolumePrivateResult(

api/src/opentrons/protocol_engine/commands/load_pipette.py

@@ -52,6 +52,13 @@ class LoadPipetteParams(BaseModel):
         description="An optional ID to assign to this pipette. If None, an ID "
         "will be generated.",
     )
+    tipOverlapNotAfterVersion: Optional[str] = Field(
+        None,
+        description="A version of tip overlap data to not exceed. The highest-versioned "
+        "tip overlap data that does not exceed this version will be used. Versions are "
+        "expressed as vN where N is an integer, counting up from v0. If None, the current "
+        "highest version will be used.",
+    )
 
 
 class LoadPipetteResult(BaseModel):
@@ -112,6 +119,7 @@ async def execute(
             pipette_name=params.pipetteName,
             mount=params.mount,
             pipette_id=params.pipetteId,
+            tip_overlap_version=params.tipOverlapNotAfterVersion,
         )
 
         return LoadPipetteResult(

api/src/opentrons/protocol_engine/execution/equipment.py

@@ -203,6 +203,7 @@ async def load_pipette(
         pipette_name: PipetteNameType,
         mount: MountType,
         pipette_id: Optional[str],
+        tip_overlap_version: Optional[str],
     ) -> LoadedPipetteData:
         """Ensure the requested pipette is attached.
 
@@ -211,6 +212,8 @@ async def load_pipette(
             mount: The mount on which pipette must be attached.
             pipette_id: An optional identifier to assign the pipette. If None, an
                 identifier will be generated.
+            tip_overlap_version: An optional specifier for the version of tip overlap data to use.
+                If None, defaults to v0. Does not need to be format checked - this function does it.
 
         Returns:
             A LoadedPipetteData object.
@@ -225,6 +228,11 @@ async def load_pipette(
             if isinstance(pipette_name, PipetteNameType)
             else pipette_name
         )
+        sanitized_overlap_version = (
+            pipette_data_provider.validate_and_default_tip_overlap_version(
+                tip_overlap_version
+            )
+        )
 
         pipette_id = pipette_id or self._model_utils.generate_id()
         if not use_virtual_pipettes:
@@ -257,14 +265,16 @@ async def load_pipette(
 
             serial_number = pipette_dict["pipette_id"]
             static_pipette_config = pipette_data_provider.get_pipette_static_config(
-                pipette_dict
+                pipette_dict=pipette_dict, tip_overlap_version=sanitized_overlap_version
             )
 
         else:
             serial_number = self._model_utils.generate_id(prefix="fake-serial-number-")
             static_pipette_config = (
                 self._virtual_pipette_data_provider.get_virtual_pipette_static_config(
-                    pipette_name_value, pipette_id
+                    pipette_name=pipette_name_value,
+                    pipette_id=pipette_id,
+                    tip_overlap_version=sanitized_overlap_version,
                 )
             )
         serial = serial_number or ""
@@ -367,9 +377,7 @@ async def load_module(
         )
 
     async def configure_for_volume(
-        self,
-        pipette_id: str,
-        volume: float,
+        self, pipette_id: str, volume: float, tip_overlap_version: Optional[str]
     ) -> LoadedConfigureForVolumeData:
         """Ensure the requested volume can be configured for the given pipette.
 
@@ -381,6 +389,11 @@ async def configure_for_volume(
             A LoadedConfiguredVolumeData object.
         """
         use_virtual_pipettes = self._state_store.config.use_virtual_pipettes
+        sanitized_overlap_version = (
+            pipette_data_provider.validate_and_default_tip_overlap_version(
+                tip_overlap_version
+            )
+        )
 
         if not use_virtual_pipettes:
             mount = self._state_store.pipettes.get_mount(pipette_id).to_hw_mount()
@@ -390,7 +403,7 @@ async def configure_for_volume(
 
             serial_number = pipette_dict["pipette_id"]
             static_pipette_config = pipette_data_provider.get_pipette_static_config(
-                pipette_dict
+                pipette_dict=pipette_dict, tip_overlap_version=sanitized_overlap_version
             )
 
         else:
@@ -401,7 +414,9 @@ async def configure_for_volume(
 
             serial_number = self._model_utils.generate_id(prefix="fake-serial-number-")
             static_pipette_config = self._virtual_pipette_data_provider.get_virtual_pipette_static_config_by_model_string(
-                model, pipette_id
+                pipette_model_string=model,
+                pipette_id=pipette_id,
+                tip_overlap_version=sanitized_overlap_version,
             )
 
         return LoadedConfigureForVolumeData(