Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(api): dynamic liquid tracking papi changes #17291

Closed
wants to merge 2 commits into from
Closed

feat(api): dynamic liquid tracking papi changes #17291

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ea7bdda
fix descriptions
caila-marashaj Jan 16, 2025
1bce5fe
cherry-pick rough draft stuff
caila-marashaj Jan 16, 2025
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
557 changes: 129 additions & 428 deletions api/src/opentrons/protocol_api/core/engine/instrument.py
View file
Open in desktop

Large diffs are not rendered by default.

42 changes: 32 additions & 10 deletions api/src/opentrons/protocol_api/core/instrument.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,10 +13,9 @@
from opentrons.protocol_api._liquid import LiquidClass
from ..disposal_locations import TrashBin, WasteChute
from .well import WellCoreType
from .labware import LabwareCoreType


class AbstractInstrument(ABC, Generic[WellCoreType, LabwareCoreType]):
class AbstractInstrument(ABC, Generic[WellCoreType]):
@abstractmethod
def get_default_speed(self) -> float:
...
Expand All @@ -42,7 +41,7 @@ def aspirate(
rate: float,
flow_rate: float,
in_place: bool,
is_meniscus: Optional[bool] = None,
meniscus_tracking: Optional[types.MeniscusTracking] = None,
) -> None:
"""Aspirate a given volume of liquid from the specified location.
Args:
Expand All @@ -52,6 +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.
meniscus_tracking: Optional data about where to aspirate from.
"""
...

Expand All @@ -65,7 +65,7 @@ def dispense(
flow_rate: float,
in_place: bool,
push_out: Optional[float],
is_meniscus: Optional[bool] = None,
meniscus_tracking: Optional[types.MeniscusTracking] = None,
) -> None:
"""Dispense a given volume of liquid into the specified location.
Args:
Expand All @@ -76,6 +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.
meniscus_tracking: Optional data about where to dispense from.
"""
...

Expand Down Expand Up @@ -312,15 +313,27 @@ def configure_nozzle_layout(
...

@abstractmethod
def transfer_liquid(
def load_liquid_class(
self,
liquid_class: LiquidClass,
pipette_load_name: str,
tiprack_uri: str,
) -> str:
"""Load the liquid class properties of given pipette and tiprack into the engine.

Returns: ID of the liquid class record
"""
...

@abstractmethod
def transfer_liquid(
self,
liquid_class_id: str,
volume: float,
source: List[Tuple[types.Location, WellCoreType]],
dest: List[Tuple[types.Location, WellCoreType]],
source: List[WellCoreType],
dest: List[WellCoreType],
new_tip: TransferTipPolicyV2,
tip_racks: List[Tuple[types.Location, LabwareCoreType]],
trash_location: Union[types.Location, TrashBin, WasteChute],
trash_location: Union[WellCoreType, types.Location, TrashBin, WasteChute],
) -> None:
"""Transfer a liquid from source to dest according to liquid class properties."""
...
Expand Down Expand Up @@ -354,9 +367,18 @@ def liquid_probe_without_recovery(
"""Do a liquid probe to find the level of the liquid in the well."""
...

@abstractmethod
def liquid_probe_testing_data(
self,
well_core: types.WellCore,
loc: types.Location,
operation_volume: float,
) -> Tuple[float, float, float]:
...

@abstractmethod
def nozzle_configuration_valid_for_lld(self) -> bool:
"""Check if the nozzle configuration currently supports LLD."""


InstrumentCoreType = TypeVar("InstrumentCoreType", bound=AbstractInstrument[Any, Any])
InstrumentCoreType = TypeVar("InstrumentCoreType", bound=AbstractInstrument[Any])
30 changes: 20 additions & 10 deletions api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
from __future__ import annotations

import logging
from typing import TYPE_CHECKING, Optional, Union, List, Tuple
from typing import TYPE_CHECKING, Optional, Union, List

from opentrons import types
from opentrons.hardware_control import CriticalPoint
Expand All @@ -25,7 +25,6 @@
from ...disposal_locations import TrashBin, WasteChute
from ..instrument import AbstractInstrument
from .legacy_well_core import LegacyWellCore
from .legacy_labware_core import LegacyLabwareCore
from .legacy_module_core import LegacyThermocyclerCore, LegacyHeaterShakerCore

if TYPE_CHECKING:
Expand All @@ -38,7 +37,7 @@
"""In PAPIv2.1 and below, tips are always dropped 10 mm from the bottom of the well."""


class LegacyInstrumentCore(AbstractInstrument[LegacyWellCore, LegacyLabwareCore]):
class LegacyInstrumentCore(AbstractInstrument[LegacyWellCore]):
"""Implementation of the InstrumentContext interface."""

def __init__(
Expand Down Expand Up @@ -85,7 +84,7 @@ def aspirate(
rate: float,
flow_rate: float,
in_place: bool,
is_meniscus: Optional[bool] = None,
meniscus_tracking: Optional[types.MeniscusTracking] = None,
) -> None:
"""Aspirate a given volume of liquid from the specified location.
Args:
Expand All @@ -95,6 +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.
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 @@ -128,7 +128,7 @@ def dispense(
flow_rate: float,
in_place: bool,
push_out: Optional[float],
is_meniscus: Optional[bool] = None,
meniscus_tracking: Optional[types.MeniscusTracking] = None,
) -> None:
"""Dispense a given volume of liquid into the specified location.
Args:
Expand All @@ -139,6 +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.
meniscus_tracking: Optional data about where to dispense from.
"""
if isinstance(location, (TrashBin, WasteChute)):
raise APIVersionError(
Expand Down Expand Up @@ -557,15 +558,24 @@ def configure_nozzle_layout(
"""This will never be called because it was added in API 2.16."""
pass

def transfer_liquid(
def load_liquid_class(
self,
liquid_class: LiquidClass,
pipette_load_name: str,
tiprack_uri: str,
) -> str:
"""This will never be called because it was added in .."""
# TODO(spp, 2024-11-20): update the docstring and error to include API version
assert False, "load_liquid_class is not supported in legacy context"

def transfer_liquid(
self,
liquid_class_id: str,
volume: float,
source: List[Tuple[types.Location, LegacyWellCore]],
dest: List[Tuple[types.Location, LegacyWellCore]],
source: List[LegacyWellCore],
dest: List[LegacyWellCore],
new_tip: TransferTipPolicyV2,
tip_racks: List[Tuple[types.Location, LegacyLabwareCore]],
trash_location: Union[types.Location, TrashBin, WasteChute],
trash_location: Union[LegacyWellCore, types.Location, TrashBin, WasteChute],
) -> None:
"""This will never be called because it was added in .."""
# TODO(spp, 2024-11-20): update the docstring and error to include API version
Expand Down
28 changes: 16 additions & 12 deletions api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
from __future__ import annotations

import logging
from typing import TYPE_CHECKING, Optional, Union, List, Tuple
from typing import TYPE_CHECKING, Optional, Union, List

from opentrons import types
from opentrons.hardware_control.dev_types import PipetteDict
Expand All @@ -23,7 +23,6 @@
UnexpectedTipAttachError,
)

from ..legacy.legacy_labware_core import LegacyLabwareCore
from ...disposal_locations import TrashBin, WasteChute
from opentrons.protocol_api._nozzle_layout import NozzleLayout
from opentrons.protocol_api._liquid import LiquidClass
Expand All @@ -43,9 +42,7 @@
"""In PAPIv2.1 and below, tips are always dropped 10 mm from the bottom of the well."""


class LegacyInstrumentCoreSimulator(
AbstractInstrument[LegacyWellCore, LegacyLabwareCore]
):
class LegacyInstrumentCoreSimulator(AbstractInstrument[LegacyWellCore]):
"""A simulation of an instrument context."""

def __init__(
Expand Down Expand Up @@ -98,7 +95,6 @@ def aspirate(
rate: float,
flow_rate: float,
in_place: bool,
is_meniscus: 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 @@ -140,7 +136,6 @@ def dispense(
flow_rate: float,
in_place: bool,
push_out: Optional[float],
is_meniscus: Optional[bool] = None,
) -> None:
if isinstance(location, (TrashBin, WasteChute)):
raise APIVersionError(
Expand Down Expand Up @@ -477,15 +472,24 @@ def configure_nozzle_layout(
"""This will never be called because it was added in API 2.15."""
pass

def transfer_liquid(
def load_liquid_class(
self,
liquid_class: LiquidClass,
pipette_load_name: str,
tiprack_uri: str,
) -> str:
"""This will never be called because it was added in .."""
# TODO(spp, 2024-11-20): update the docstring and error to include API version
assert False, "load_liquid_class is not supported in legacy context"

def transfer_liquid(
self,
liquid_class_id: str,
volume: float,
source: List[Tuple[types.Location, LegacyWellCore]],
dest: List[Tuple[types.Location, LegacyWellCore]],
source: List[LegacyWellCore],
dest: List[LegacyWellCore],
new_tip: TransferTipPolicyV2,
tip_racks: List[Tuple[types.Location, LegacyLabwareCore]],
trash_location: Union[types.Location, TrashBin, WasteChute],
trash_location: Union[LegacyWellCore, types.Location, TrashBin, WasteChute],
) -> None:
"""Transfer a liquid from source to dest according to liquid class properties."""
# TODO(spp, 2024-11-20): update the docstring and error to include API version
Expand Down
Loading
Loading