Skip to content

Commit 61e4186

Browse files
rclarke0rclarke0
authored
Merge branch 'edge' into abr-error-recording-fixes
2 parents e3eed55 + 630b1b2 commit 61e4186

File tree

14 files changed

+2033
-133
lines changed

14 files changed

+2033
-133
lines changed

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

+231-7
Original file line numberDiff line numberDiff line change
@@ -2,13 +2,17 @@
22

33
from __future__ import annotations
44

5-
from typing import Optional, TYPE_CHECKING, cast, Union, List
5+
from typing import Optional, TYPE_CHECKING, cast, Union, List, Tuple
66
from opentrons.types import Location, Mount, NozzleConfigurationType, NozzleMapInterface
77
from opentrons.hardware_control import SyncHardwareAPI
88
from opentrons.hardware_control.dev_types import PipetteDict
99
from opentrons.protocols.api_support.util import FlowRates, find_value_for_api_version
1010
from opentrons.protocols.api_support.types import APIVersion
11-
from opentrons.protocols.advanced_control.transfers.common import TransferTipPolicyV2
11+
from opentrons.protocols.advanced_control.transfers.common import (
12+
TransferTipPolicyV2,
13+
check_valid_volume_parameters,
14+
expand_for_volume_constraints,
15+
)
1216
from opentrons.protocol_engine import commands as cmd
1317
from opentrons.protocol_engine import (
1418
DeckPoint,
@@ -38,6 +42,7 @@
3842
)
3943
from opentrons.protocol_api._nozzle_layout import NozzleLayout
4044
from . import overlap_versions, pipette_movement_conflict
45+
from . import transfer_components_executor as tx_comps_executor
4146

4247
from .well import WellCore
4348
from ..instrument import AbstractInstrument
@@ -46,6 +51,7 @@
4651
if TYPE_CHECKING:
4752
from .protocol import ProtocolCore
4853
from opentrons.protocol_api._liquid import LiquidClass
54+
from opentrons.protocol_api._liquid_properties import TransferProperties
4955

5056
_DISPENSE_VOLUME_VALIDATION_ADDED_IN = APIVersion(2, 17)
5157

@@ -892,16 +898,230 @@ def load_liquid_class(
892898
)
893899
return result.liquidClassId
894900

901+
# TODO: update with getNextTip implementation
902+
def get_next_tip(self) -> None:
903+
"""Get the next tip to pick up."""
904+
895905
def transfer_liquid(
896906
self,
897-
liquid_class_id: str,
907+
liquid_class: LiquidClass,
898908
volume: float,
899-
source: List[WellCore],
900-
dest: List[WellCore],
909+
source: List[Tuple[Location, WellCore]],
910+
dest: List[Tuple[Location, WellCore]],
901911
new_tip: TransferTipPolicyV2,
902-
trash_location: Union[WellCore, Location, TrashBin, WasteChute],
912+
tiprack_uri: str,
913+
trash_location: Union[Location, TrashBin, WasteChute],
903914
) -> None:
904-
"""Execute transfer using liquid class properties."""
915+
"""Execute transfer using liquid class properties.
916+
917+
Args:
918+
liquid_class: The liquid class to use for transfer properties.
919+
volume: Volume to transfer per well.
920+
source: List of source wells, with each well represented as a tuple of
921+
types.Location and WellCore.
922+
types.Location is only necessary for saving the last accessed location.
923+
dest: List of destination wells, with each well represented as a tuple of
924+
types.Location and WellCore.
925+
types.Location is only necessary for saving the last accessed location.
926+
new_tip: Whether the transfer should use a new tip 'once', 'never', 'always',
927+
or 'per source'.
928+
tiprack_uri: The URI of the tiprack that the transfer settings are for.
929+
tip_drop_location: Location where the tip will be dropped (if appropriate).
930+
"""
931+
# This function is WIP
932+
# TODO: use the ID returned by load_liquid_class in command annotations
933+
self.load_liquid_class(
934+
liquid_class=liquid_class,
935+
pipette_load_name=self.get_pipette_name(), # TODO: update this to use load name instead
936+
tiprack_uri=tiprack_uri,
937+
)
938+
transfer_props = liquid_class.get_for(
939+
# update this to fetch load name instead
940+
pipette=self.get_pipette_name(),
941+
tiprack=tiprack_uri,
942+
)
943+
aspirate_props = transfer_props.aspirate
944+
945+
check_valid_volume_parameters(
946+
disposal_volume=0, # No disposal volume for 1-to-1 transfer
947+
air_gap=aspirate_props.retract.air_gap_by_volume.get_for_volume(volume),
948+
max_volume=self.get_max_volume(),
949+
)
950+
source_dest_per_volume_step = expand_for_volume_constraints(
951+
volumes=[volume for _ in range(len(source))],
952+
targets=zip(source, dest),
953+
max_volume=self.get_max_volume(),
954+
)
955+
if new_tip == TransferTipPolicyV2.ONCE:
956+
# TODO: update this once getNextTip is implemented
957+
self.get_next_tip()
958+
for step_volume, (src, dest) in source_dest_per_volume_step: # type: ignore[assignment]
959+
if new_tip == TransferTipPolicyV2.ALWAYS:
960+
# TODO: update this once getNextTip is implemented
961+
self.get_next_tip()
962+
963+
# TODO: add aspirate and dispense
964+
965+
if new_tip == TransferTipPolicyV2.ALWAYS:
966+
if isinstance(trash_location, (TrashBin, WasteChute)):
967+
self.drop_tip_in_disposal_location(
968+
disposal_location=trash_location,
969+
home_after=False,
970+
alternate_tip_drop=True,
971+
)
972+
elif isinstance(trash_location, Location):
973+
self.drop_tip(
974+
location=trash_location,
975+
well_core=trash_location.labware.as_well()._core, # type: ignore[arg-type]
976+
home_after=False,
977+
alternate_drop_location=True,
978+
)
979+
980+
def aspirate_liquid_class(
981+
self,
982+
volume: float,
983+
source: Tuple[Location, WellCore],
984+
transfer_properties: TransferProperties,
985+
transfer_type: tx_comps_executor.TransferType,
986+
tip_contents: List[tx_comps_executor.LiquidAndAirGapPair],
987+
) -> tx_comps_executor.LiquidAndAirGapPair:
988+
"""Execute aspiration steps.
989+
990+
1. Submerge
991+
2. Mix
992+
3. pre-wet
993+
4. Aspirate
994+
5. Delay- wait inside the liquid
995+
6. Aspirate retract
996+
997+
Return: The last liquid and air gap pair in tip.
998+
"""
999+
aspirate_props = transfer_properties.aspirate
1000+
source_loc, source_well = source
1001+
aspirate_point = (
1002+
tx_comps_executor.absolute_point_from_position_reference_and_offset(
1003+
well=source_well,
1004+
position_reference=aspirate_props.position_reference,
1005+
offset=aspirate_props.offset,
1006+
)
1007+
)
1008+
aspirate_location = Location(aspirate_point, labware=source_loc.labware)
1009+
if len(tip_contents) > 0:
1010+
last_liquid_and_airgap_in_tip = tip_contents[-1]
1011+
else:
1012+
last_liquid_and_airgap_in_tip = tx_comps_executor.LiquidAndAirGapPair(
1013+
liquid=0,
1014+
air_gap=0,
1015+
)
1016+
components_executor = tx_comps_executor.TransferComponentsExecutor(
1017+
instrument_core=self,
1018+
transfer_properties=transfer_properties,
1019+
target_location=aspirate_location,
1020+
target_well=source_well,
1021+
transfer_type=transfer_type,
1022+
tip_state=tx_comps_executor.TipState(
1023+
last_liquid_and_air_gap_in_tip=last_liquid_and_airgap_in_tip
1024+
),
1025+
)
1026+
components_executor.submerge(submerge_properties=aspirate_props.submerge)
1027+
# TODO: when aspirating for consolidation, do not perform mix
1028+
components_executor.mix(
1029+
mix_properties=aspirate_props.mix, last_dispense_push_out=False
1030+
)
1031+
# TODO: when aspirating for consolidation, do not preform pre-wet
1032+
components_executor.pre_wet(
1033+
volume=volume,
1034+
)
1035+
components_executor.aspirate_and_wait(volume=volume)
1036+
components_executor.retract_after_aspiration(volume=volume)
1037+
return components_executor.tip_state.last_liquid_and_air_gap_in_tip
1038+
1039+
def dispense_liquid_class(
1040+
self,
1041+
volume: float,
1042+
dest: Tuple[Location, WellCore],
1043+
source: Optional[Tuple[Location, WellCore]],
1044+
transfer_properties: TransferProperties,
1045+
transfer_type: tx_comps_executor.TransferType,
1046+
tip_contents: List[tx_comps_executor.LiquidAndAirGapPair],
1047+
trash_location: Union[Location, TrashBin, WasteChute],
1048+
) -> tx_comps_executor.LiquidAndAirGapPair:
1049+
"""Execute single-dispense steps.
1050+
1. Move pipette to the ‘submerge’ position with normal speed.
1051+
- The pipette will move in an arc- move to max z height of labware
1052+
(if asp & disp are in same labware)
1053+
or max z height of all labware (if asp & disp are in separate labware)
1054+
2. Air gap removal:
1055+
- If dispense location is above the meniscus, DO NOT remove air gap
1056+
(it will be dispensed along with rest of the liquid later).
1057+
All other scenarios, remove the air gap by doing a dispense
1058+
- Flow rate = min(dispenseFlowRate, (airGapByVolume)/sec)
1059+
- Use the post-dispense delay
1060+
4. Move to the dispense position at the specified ‘submerge’ speed
1061+
(even if we might not be moving into the liquid)
1062+
- Do a delay (submerge delay)
1063+
6. Dispense:
1064+
- Dispense at the specified flow rate.
1065+
- Do a push out as specified ONLY IF there is no mix following the dispense AND the tip is empty.
1066+
Volume for push out is the volume being dispensed. So if we are dispensing 50uL, use pushOutByVolume[50] as push out volume.
1067+
7. Delay
1068+
8. Mix using the same flow rate and delays as specified for asp+disp,
1069+
with the volume and the number of repetitions specified. Use the delays in asp & disp.
1070+
- If the dispense position is outside the liquid, then raise error if mix is enabled.
1071+
- If the user wants to perform a mix then they should specify a dispense position that’s inside the liquid OR do mix() on the wells after transfer.
1072+
- Do push out at the last dispense.
1073+
9. Retract
1074+
1075+
Return:
1076+
The last liquid and air gap pair in tip.
1077+
"""
1078+
dispense_props = transfer_properties.dispense
1079+
dest_loc, dest_well = dest
1080+
dispense_point = (
1081+
tx_comps_executor.absolute_point_from_position_reference_and_offset(
1082+
well=dest_well,
1083+
position_reference=dispense_props.position_reference,
1084+
offset=dispense_props.offset,
1085+
)
1086+
)
1087+
dispense_location = Location(dispense_point, labware=dest_loc.labware)
1088+
if len(tip_contents) > 0:
1089+
last_liquid_and_airgap_in_tip = tip_contents[-1]
1090+
else:
1091+
last_liquid_and_airgap_in_tip = tx_comps_executor.LiquidAndAirGapPair(
1092+
liquid=0,
1093+
air_gap=0,
1094+
)
1095+
components_executor = tx_comps_executor.TransferComponentsExecutor(
1096+
instrument_core=self,
1097+
transfer_properties=transfer_properties,
1098+
target_location=dispense_location,
1099+
target_well=dest_well,
1100+
transfer_type=transfer_type,
1101+
tip_state=tx_comps_executor.TipState(
1102+
last_liquid_and_air_gap_in_tip=last_liquid_and_airgap_in_tip
1103+
),
1104+
)
1105+
components_executor.submerge(submerge_properties=dispense_props.submerge)
1106+
if dispense_props.mix.enabled:
1107+
push_out_vol = 0.0
1108+
else:
1109+
# TODO: if distributing, do a push out only at the last dispense
1110+
push_out_vol = dispense_props.push_out_by_volume.get_for_volume(volume)
1111+
components_executor.dispense_and_wait(
1112+
volume=volume,
1113+
push_out_override=push_out_vol,
1114+
)
1115+
components_executor.mix(
1116+
mix_properties=dispense_props.mix,
1117+
last_dispense_push_out=True,
1118+
)
1119+
components_executor.retract_after_dispensing(
1120+
trash_location=trash_location,
1121+
source_location=source[0] if source else None,
1122+
source_well=source[1] if source else None,
1123+
)
1124+
return components_executor.tip_state.last_liquid_and_air_gap_in_tip
9051125

9061126
def retract(self) -> None:
9071127
"""Retract this instrument to the top of the gantry."""
@@ -994,3 +1214,7 @@ def nozzle_configuration_valid_for_lld(self) -> bool:
9941214
return self._engine_client.state.pipettes.get_nozzle_configuration_supports_lld(
9951215
self.pipette_id
9961216
)
1217+
1218+
def delay(self, seconds: float) -> None:
1219+
"""Call a protocol delay."""
1220+
self._protocol_core.delay(seconds=seconds, msg=None)

0 commit comments

Comments
 (0)