From f3cb886a42b57cc04e830b2fd1fc8abfa51371bb Mon Sep 17 00:00:00 2001 From: Rick Wierenga Date: Fri, 27 Sep 2024 11:12:27 -0700 Subject: [PATCH] hamilton liquid classes make kwargs (#248) --- CHANGELOG.md | 2 + .../liquid_handling/backends/hamilton/STAR.py | 280 +++++------------- .../backends/hamilton/STAR_tests.py | 54 +++- .../backends/hamilton/vantage.py | 206 ++++--------- .../backends/hamilton/vantage_tests.py | 37 ++- .../liquid_classes/hamilton/base.py | 48 ++- .../liquid_classes/hamilton/star.py | 40 +-- .../liquid_classes/hamilton/vantage.py | 41 +-- 8 files changed, 255 insertions(+), 453 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 80d182aea6..668deaba36 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,6 +38,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). - Move `LiquidHandlerChatterboxBackend` from `liquid_handling.backends.chatterbox_backend` to `liquid_handling.backends.chatterbox` (https://github.com/PyLabRobot/pylabrobot/pull/242) - Changed `pedestal_size_z=-5` to `pedestal_size_z=-4.74` for `PLT_CAR_L5AC_A00` (https://github.com/PyLabRobot/pylabrobot/pull/255) - rename `homogenization_` parameters in `STAR` to `mix_` (https://github.com/PyLabRobot/pylabrobot/pull/261) +- Hamilton liquid classes are no longer automatically inferred on the backends (`STAR`/`Vantage`). Instead, they create kwargs with `make_(asp|disp)(96)?_kwargs` (https://github.com/PyLabRobot/pylabrobot/pull/248) + - This also applies to volume correction curves, which are now the users' responsibility. ### Added diff --git a/pylabrobot/liquid_handling/backends/hamilton/STAR.py b/pylabrobot/liquid_handling/backends/hamilton/STAR.py index 72d01135f1..505195d509 100644 --- a/pylabrobot/liquid_handling/backends/hamilton/STAR.py +++ b/pylabrobot/liquid_handling/backends/hamilton/STAR.py @@ -14,8 +14,7 @@ from pylabrobot import audio from pylabrobot.liquid_handling.backends.hamilton.base import HamiltonLiquidHandler from pylabrobot.liquid_handling.errors import ChannelizedError -from pylabrobot.liquid_handling.liquid_classes.hamilton import ( - HamiltonLiquidClass, get_star_liquid_class) +from pylabrobot.liquid_handling.liquid_classes.hamilton import HamiltonLiquidClass from pylabrobot.liquid_handling.standard import ( Pickup, PickupTipRack, @@ -38,7 +37,6 @@ NoTipError ) from pylabrobot.resources.hamilton.hamilton_decks import STAR_SIZE_X, STARLET_SIZE_X -from pylabrobot.resources.liquid import Liquid from pylabrobot.resources.ml_star import HamiltonTip, TipDropMethod, TipPickupMethod, TipSize from pylabrobot.resources.resource_holder import get_child_location from pylabrobot.utils.linalg import matrix_vector_multiply_3x3 @@ -1438,8 +1436,6 @@ async def aspirate( self, ops: List[Aspiration], use_channels: List[int], - jet: Optional[List[bool]] = None, - blow_out: Optional[List[bool]] = None, lld_search_height: Optional[List[float]] = None, clot_detection_height: Optional[List[float]] = None, pull_out_distance_transport_air: Optional[List[float]] = None, @@ -1489,9 +1485,6 @@ async def aspirate( Args: ops: The aspiration operations to perform. use_channels: The channels to use for the operations. - jet: whether to search for a jet liquid class. Only used on dispense. Default is False. - blow_out: whether to blow out air. Only used on dispense. Note that in the VENUS Liquid - Editor, this is called "empty". Default is False. lld_search_height: The height to start searching for the liquid level when using LLD. clot_detection_height: Unknown, but probably the height to search for clots when doing LLD. @@ -1540,46 +1533,20 @@ async def aspirate( starting an aspiration. min_z_endpos: The minimum height to move to, this is the end of aspiration. - hamilton_liquid_classes: Override the default liquid classes. See - pylabrobot/liquid_handling/liquid_classes/hamilton/star.py liquid_surface_no_lld: Liquid surface at function without LLD [mm]. Must be between 0 and 360. Defaults to well bottom + liquid height. Should use absolute z. """ + if hamilton_liquid_classes is not None: + raise NotImplementedError("Hamilton liquid classes are deprecated.") + x_positions, y_positions, channels_involved = \ self._ops_to_fw_positions(ops, use_channels) n = len(ops) - if jet is None: - jet = [False] * n - if blow_out is None: - blow_out = [False] * n - - if hamilton_liquid_classes is None: - hamilton_liquid_classes = [] - for i, op in enumerate(ops): - liquid = Liquid.WATER # default to WATER - # [-1][0]: get last liquid in well, [0] is indexing into the tuple - if len(op.liquids) > 0 and op.liquids[-1][0] is not None: - liquid = op.liquids[-1][0] - - hamilton_liquid_classes.append(get_star_liquid_class( - tip_volume=op.tip.maximal_volume, - is_core=False, - is_tip=True, - has_filter=op.tip.has_filter, - liquid=liquid, - jet=jet[i], - blow_out=blow_out[i] - )) - self._assert_valid_resources([op.resource for op in ops]) - # correct volumes using the liquid class - volumes = [hlc.compute_corrected_volume(op.volume) if hlc is not None else op.volume - for op, hlc in zip(ops, hamilton_liquid_classes)] - well_bottoms = [op.resource.get_absolute_location().z + op.offset.z + \ op.resource.material_z_thickness for op in ops] liquid_surfaces_no_lld = liquid_surfaces_no_lld or [wb + (op.liquid_height or 0) @@ -1591,9 +1558,7 @@ async def aspirate( ] else: lld_search_height = [(wb + sh) for wb, sh in zip(well_bottoms, lld_search_height)] - clot_detection_height = _fill_in_defaults(clot_detection_height, - default=[hlc.aspiration_clot_retract_height if hlc is not None else 0 - for hlc in hamilton_liquid_classes]) + clot_detection_height = _fill_in_defaults(clot_detection_height, default=[0]*n) pull_out_distance_transport_air = _fill_in_defaults(pull_out_distance_transport_air, [10]*n) second_section_height = _fill_in_defaults(second_section_height, [3.2]*n) second_section_ratio = _fill_in_defaults(second_section_ratio, [618.0]*n) @@ -1602,16 +1567,9 @@ async def aspirate( immersion_depth = _fill_in_defaults(immersion_depth, [0]*n) immersion_depth_direction = _fill_in_defaults(immersion_depth_direction, [0]*n) surface_following_distance = _fill_in_defaults(surface_following_distance, [0]*n) - flow_rates = [ - op.flow_rate or (hlc.aspiration_flow_rate if hlc is not None else 100) - for op, hlc in zip(ops, hamilton_liquid_classes)] - transport_air_volume = _fill_in_defaults(transport_air_volume, - default=[hlc.aspiration_air_transport_volume if hlc is not None else 0 - for hlc in hamilton_liquid_classes]) - blow_out_air_volumes = [(op.blow_out_air_volume or - (hlc.aspiration_blow_out_volume - if hlc is not None else 0)) - for op, hlc in zip(ops, hamilton_liquid_classes)] + flow_rates = [op.flow_rate or 100 for op in ops] + transport_air_volume = _fill_in_defaults(transport_air_volume, default=[0]*n) + blow_out_air_volumes = [op.blow_out_air_volume or 0 for op in ops] pre_wetting_volume = _fill_in_defaults(pre_wetting_volume, [0]*n) lld_mode = _fill_in_defaults(lld_mode, [self.__class__.LLDMode.OFF]*n) gamma_lld_sensitivity = _fill_in_defaults(gamma_lld_sensitivity, [1]*n) @@ -1620,19 +1578,13 @@ async def aspirate( _fill_in_defaults(aspirate_position_above_z_touch_off, [0]*n) detection_height_difference_for_dual_lld = \ _fill_in_defaults(detection_height_difference_for_dual_lld, [0]*n) - swap_speed = _fill_in_defaults(swap_speed, - default=[hlc.aspiration_swap_speed if hlc is not None else 100 - for hlc in hamilton_liquid_classes]) - settling_time = _fill_in_defaults(settling_time, - default=[hlc.aspiration_settling_time if hlc is not None else 0 - for hlc in hamilton_liquid_classes]) + swap_speed = _fill_in_defaults(swap_speed, default=[100]*n) + settling_time = _fill_in_defaults(settling_time, default=[0]*n) mix_volume = _fill_in_defaults(mix_volume, [0]*n) mix_cycles = _fill_in_defaults(mix_cycles, [0]*n) mix_position_from_liquid_surface = \ _fill_in_defaults(mix_position_from_liquid_surface, [0]*n) - mix_speed = _fill_in_defaults(mix_speed, - default=[hlc.aspiration_mix_flow_rate if hlc is not None else 50.0 - for hlc in hamilton_liquid_classes]) + mix_speed = _fill_in_defaults(mix_speed, [50]*n) mix_surface_following_distance = \ _fill_in_defaults(mix_surface_following_distance, [0]*n) limit_curve_index = _fill_in_defaults(limit_curve_index, [0]*n) @@ -1657,7 +1609,7 @@ async def aspirate( x_positions=x_positions, y_positions=y_positions, - aspiration_volumes=[round(vol * 10) for vol in volumes], + aspiration_volumes=[round(op.volume * 10) for op in ops], lld_search_height=[round(lsh * 10) for lsh in lld_search_height], clot_detection_height=[round(cd * 10) for cd in clot_detection_height], liquid_surface_no_lld=[round(ls * 10) for ls in liquid_surfaces_no_lld], @@ -1808,6 +1760,9 @@ async def dispense( documentation. Dispense mode 4. """ + if hamilton_liquid_classes is not None: + raise NotImplementedError("Hamilton liquid classes are deprecated.") + x_positions, y_positions, channels_involved = \ self._ops_to_fw_positions(ops, use_channels) @@ -1820,28 +1775,6 @@ async def dispense( if blow_out is None: blow_out = [False] * n - if hamilton_liquid_classes is None: - hamilton_liquid_classes = [] - for i, op in enumerate(ops): - liquid = Liquid.WATER # default to WATER - # [-1][0]: get last liquid in tip, [0] is indexing into the tuple - if len(op.liquids) > 0 and op.liquids[-1][0] is not None: - liquid = op.liquids[-1][0] - - hamilton_liquid_classes.append(get_star_liquid_class( - tip_volume=op.tip.maximal_volume, - is_core=False, - is_tip=True, - has_filter=op.tip.has_filter, - liquid=liquid, - jet=jet[i], - blow_out=blow_out[i] - )) - - # correct volumes using the liquid class - volumes = [hlc.compute_corrected_volume(op.volume) if hlc is not None else op.volume - for op, hlc in zip(ops, hamilton_liquid_classes)] - well_bottoms = [op.resource.get_absolute_location().z + op.offset.z + \ op.resource.material_z_thickness for op in ops] liquid_surfaces_no_lld = liquid_surface_no_lld or \ @@ -1865,37 +1798,22 @@ async def dispense( immersion_depth = _fill_in_defaults(immersion_depth, [0]*n) immersion_depth_direction = _fill_in_defaults(immersion_depth_direction, [0]*n) surface_following_distance = _fill_in_defaults(surface_following_distance, [0]*n) - flow_rates = [ - op.flow_rate or (hlc.dispense_flow_rate if hlc is not None else 120) - for op, hlc in zip(ops, hamilton_liquid_classes)] + flow_rates = [op.flow_rate or 120 for op in ops] cut_off_speed = _fill_in_defaults(cut_off_speed, [5.0]*n) - stop_back_volume = _fill_in_defaults(stop_back_volume, - default=[hlc.dispense_stop_back_volume if hlc is not None else 0 - for hlc in hamilton_liquid_classes]) - transport_air_volume = _fill_in_defaults(transport_air_volume, - default=[hlc.dispense_air_transport_volume if hlc is not None else 0 - for hlc in hamilton_liquid_classes]) - blow_out_air_volumes = [(op.blow_out_air_volume or - (hlc.dispense_blow_out_volume - if hlc is not None else 0)) - for op, hlc in zip(ops, hamilton_liquid_classes)] + stop_back_volume = _fill_in_defaults(stop_back_volume, default=[0]*n) + transport_air_volume = _fill_in_defaults(transport_air_volume, [0]*n) + blow_out_air_volumes = [op.blow_out_air_volume or 0 for op in ops] lld_mode = _fill_in_defaults(lld_mode, [self.__class__.LLDMode.OFF]*n) dispense_position_above_z_touch_off = _fill_in_defaults(dispense_position_above_z_touch_off, default=[0]*n) gamma_lld_sensitivity = _fill_in_defaults(gamma_lld_sensitivity, [1]*n) dp_lld_sensitivity = _fill_in_defaults(dp_lld_sensitivity, [1]*n) - swap_speed = _fill_in_defaults(swap_speed, - default=[hlc.dispense_swap_speed if hlc is not None else 10.0 - for hlc in hamilton_liquid_classes]) - settling_time = _fill_in_defaults(settling_time, - default=[hlc.dispense_settling_time if hlc is not None else 0 - for hlc in hamilton_liquid_classes]) + swap_speed = _fill_in_defaults(swap_speed, [10.0]*n) + settling_time = _fill_in_defaults(settling_time, [0]*n) mix_volume = _fill_in_defaults(mix_volume, [0]*n) mix_cycles = _fill_in_defaults(mix_cycles, [0]*n) mix_position_from_liquid_surface = _fill_in_defaults(mix_position_from_liquid_surface, [0]*n) - mix_speed = _fill_in_defaults(mix_speed, - default=[hlc.dispense_mix_flow_rate if hlc is not None else 50.0 - for hlc in hamilton_liquid_classes]) + mix_speed = _fill_in_defaults(mix_speed, [50.0]*n) mix_surface_following_distance = _fill_in_defaults(mix_surface_following_distance, [0]*n) limit_curve_index = _fill_in_defaults(limit_curve_index, [0]*n) @@ -1906,7 +1824,7 @@ async def dispense( y_positions=y_positions, dispensing_mode=dispensing_modes, - dispense_volumes=[round(vol*10) for vol in volumes], + dispense_volumes=[round(op.volume*10) for op in ops], lld_search_height=[round(lsh*10) for lsh in lld_search_height], liquid_surface_no_lld=[round(ls*10) for ls in liquid_surfaces_no_lld], pull_out_distance_transport_air=[round(po*10) for po in pull_out_distance_transport_air], @@ -2006,8 +1924,6 @@ async def drop_tips96( async def aspirate96( self, aspiration: Union[AspirationPlate, AspirationContainer], - jet: bool = False, - blow_out: bool = False, use_lld: bool = False, liquid_height: float = 0, @@ -2033,7 +1949,7 @@ async def aspirate96( mix_cycles: int = 0, mix_position_from_liquid_surface: float = 0, surface_following_distance_during_mix: float = 0, - speed_of_mix: float = 120.0, + mix_speed: float = 120.0, limit_curve_index: int = 0, ): """ Aspirate using the Core96 head. @@ -2041,13 +1957,6 @@ async def aspirate96( Args: aspiration: The aspiration to perform. - jet: Whether to search for a jet liquid class. Only used on dispense. - blow_out: Whether to use "blow out" dispense mode. Only used on dispense. Note that this is - labelled as "empty" in the VENUS liquid editor, but "blow out" in the firmware - documentation. - hlc: The Hamiltonian liquid class to use. If `None`, the liquid class will be determined - automatically. - use_lld: If True, use gamma liquid level detection. If False, use liquid height. liquid_height: The height of the liquid above the bottom of the well, in millimeters. air_transport_retract_dist: The distance to retract after aspirating, in millimeters. @@ -2076,10 +1985,13 @@ async def aspirate96( liquid surface. surface_following_distance_during_mix: The distance to follow the liquid surface during mix. - speed_of_mix: The speed of mix. + mix_speed: The speed of mix. limit_curve_index: The index of the limit curve to use. """ + if hlc is not None: + raise NotImplementedError("Hamilton liquid classes are deprecated.") + assert self.core96_head_installed, "96 head must be installed" # get the first well and tip as representatives @@ -2090,57 +2002,27 @@ async def aspirate96( else: position = aspiration.container.get_absolute_location(y="b") + aspiration.offset - tip = aspiration.tips[0] - liquid_height = position.z + liquid_height - liquid_to_be_aspirated = Liquid.WATER - if len(aspiration.liquids[0]) > 0 and aspiration.liquids[0][0][0] is not None: - # [channel][liquid][PyLabRobot.resources.liquid.Liquid] - liquid_to_be_aspirated = aspiration.liquids[0][0][0] - hlc = hlc or get_star_liquid_class( - tip_volume=tip.maximal_volume, - is_core=True, - is_tip=True, - has_filter=tip.has_filter, - # get last liquid in pipette, first to be dispensed - liquid=liquid_to_be_aspirated, - jet=jet, - blow_out=blow_out, # see comment in method docstring - ) - - if hlc is not None: - volume = hlc.compute_corrected_volume(aspiration.volume) + if transport_air_volume is None: + transport_air_volume = 0 + if aspiration.blow_out_air_volume is None: + blow_out_air_volume = 0.0 + else: + blow_out_air_volume = aspiration.blow_out_air_volume + if aspiration.flow_rate is None: + flow_rate = 250.0 else: - volume = aspiration.volume - - # Get better default values from the HLC if available - transport_air_volume = transport_air_volume or \ - (hlc.aspiration_air_transport_volume if hlc is not None else 0) - blow_out_air_volume = (aspiration.blow_out_air_volume or \ - (hlc.aspiration_blow_out_volume if hlc is not None else 0)) - flow_rate = aspiration.flow_rate or (hlc.aspiration_flow_rate if hlc is not None else 250) - swap_speed = swap_speed or (hlc.aspiration_swap_speed if hlc is not None else 100) - settling_time = settling_time or \ - (hlc.aspiration_settling_time if hlc is not None else 0.5) - speed_of_mix = speed_of_mix or \ - (hlc.aspiration_mix_flow_rate if hlc is not None else 10.0) + flow_rate = aspiration.flow_rate + if swap_speed is None: + swap_speed = 100 + if settling_time is None: + settling_time = 0.5 + if mix_speed is None: + mix_speed = 10.0 channel_pattern = [True]*12*8 - # Was this ever true? Just copied it over from pyhamilton. Could have something to do with - # the liquid classes and whether blow_out mode is enabled. - # # Unfortunately, `blow_out_air_volume` does not work correctly, so instead we aspirate air - # # manually. - # if blow_out_air_volume is not None and blow_out_air_volume > 0: - # await self.aspirate_core_96( - # x_position=int(position.x * 10), - # y_positions=int(position.y * 10), - # lld_mode=0, - # liquid_surface_at_function_without_lld=int((liquid_height + 30) * 10), - # aspiration_volumes=int(blow_out_air_volume * 10) - # ) - return await self.aspirate_core_96( x_position=round(position.x * 10), x_direction=0, @@ -2161,7 +2043,7 @@ async def aspirate96( immersion_depth_direction=immersion_depth_direction, liquid_surface_sink_distance_at_the_end_of_aspiration= round(liquid_surface_sink_distance_at_the_end_of_aspiration*10), - aspiration_volumes=round(volume * 10), + aspiration_volumes=round(aspiration.volume * 10), aspiration_speed=round(flow_rate * 10), transport_air_volume=round(transport_air_volume * 10), blow_out_air_volume=round(blow_out_air_volume * 10), @@ -2176,7 +2058,7 @@ async def aspirate96( round(mix_position_from_liquid_surface * 10), surface_following_distance_during_mix= round(surface_following_distance_during_mix * 10), - speed_of_mix=round(speed_of_mix * 10), + mix_speed=round(mix_speed * 10), channel_pattern=channel_pattern, limit_curve_index=limit_curve_index, tadm_algorithm=False, @@ -2213,7 +2095,7 @@ async def dispense96( mixing_cycles: int = 0, mixing_position_from_liquid_surface: float = 0, surface_following_distance_during_mixing: float = 0, - speed_of_mixing: float = 120.0, + mix_speed: float = 120.0, limit_curve_index: int = 0, cut_off_speed: float = 5.0, stop_back_volume: float = 0, @@ -2249,12 +2131,15 @@ async def dispense96( mixing_cycles: Mixing cycles. mixing_position_from_liquid_surface: Mixing position from liquid surface, in mm. surface_following_distance_during_mixing: Surface following distance during mixing, in mm. - speed_of_mixing: Speed of mixing, in ul/s. + mix_speed: Speed of mixing, in ul/s. limit_curve_index: Limit curve index. cut_off_speed: Unknown. stop_back_volume: Unknown. """ + if hlc is not None: + raise NotImplementedError("Hamilton liquid classes are deprecated.") + assert self.core96_head_installed, "96 head must be installed" # get the first well and tip as representatives @@ -2264,40 +2149,27 @@ async def dispense96( Coordinate(z=top_left_well.material_z_thickness) + dispense.offset else: position = dispense.container.get_absolute_location(y="b") + dispense.offset - tip = dispense.tips[0] liquid_height = position.z + liquid_height dispense_mode = _dispensing_mode_for_op(empty=empty, jet=jet, blow_out=blow_out) - liquid_to_be_dispensed = Liquid.WATER # default to water. - if len(dispense.liquids[0]) > 0 and dispense.liquids[0][-1][0] is not None: - # [channel][liquid][PyLabRobot.resources.liquid.Liquid] - liquid_to_be_dispensed = dispense.liquids[0][-1][0] - hlc = hlc or get_star_liquid_class( - tip_volume=tip.maximal_volume, - is_core=True, - is_tip=True, - has_filter=tip.has_filter, - # get last liquid in pipette, first to be dispensed - liquid=liquid_to_be_dispensed, - jet=jet, - blow_out=blow_out, # see comment in method docstring - ) - - if hlc is not None: - volume = hlc.compute_corrected_volume(dispense.volume) + if transport_air_volume is None: + transport_air_volume = 0 + if dispense.blow_out_air_volume is None: + blow_out_air_volume = 0.0 else: - volume = dispense.volume - - transport_air_volume = transport_air_volume or \ - (hlc.dispense_air_transport_volume if hlc is not None else 0) - blow_out_air_volume = (dispense.blow_out_air_volume or \ - (hlc.dispense_blow_out_volume if hlc is not None else 0)) - flow_rate = dispense.flow_rate or (hlc.dispense_flow_rate if hlc is not None else 120) - swap_speed = swap_speed or (hlc.dispense_swap_speed if hlc is not None else 100) - settling_time = settling_time or (hlc.dispense_settling_time if hlc is not None else 5) - speed_of_mixing = speed_of_mixing or (hlc.dispense_mix_flow_rate if hlc is not None else 100) + blow_out_air_volume = dispense.blow_out_air_volume + if dispense.flow_rate is None: + flow_rate = 120.0 + else: + flow_rate = dispense.flow_rate + if swap_speed is None: + swap_speed = 100 + if settling_time is None: + settling_time = 5 + if mix_speed is None: + mix_speed = 100 channel_pattern = [True]*12*8 @@ -2321,7 +2193,7 @@ async def dispense96( immersion_depth_direction=immersion_depth_direction, liquid_surface_sink_distance_at_the_end_of_dispense= round(liquid_surface_sink_distance_at_the_end_of_dispense*10), - dispense_volume=round(volume*10), + dispense_volume=round(dispense.volume*10), dispense_speed=round(flow_rate*10), transport_air_volume=round(transport_air_volume*10), blow_out_air_volume=round(blow_out_air_volume*10), @@ -2333,7 +2205,7 @@ async def dispense96( mixing_cycles=mixing_cycles, mixing_position_from_liquid_surface=round(mixing_position_from_liquid_surface*10), surface_following_distance_during_mixing=round(surface_following_distance_during_mixing*10), - speed_of_mixing=round(speed_of_mixing*10), + mix_speed=round(mix_speed*10), channel_pattern=channel_pattern, limit_curve_index=limit_curve_index, tadm_algorithm=False, @@ -5050,7 +4922,7 @@ async def aspirate_core_96( mix_cycles: int = 0, mix_position_from_liquid_surface: int = 250, surface_following_distance_during_mix: int = 0, - speed_of_mix: int = 1000, + mix_speed: int = 1000, channel_pattern: List[bool] = [True] * 96, limit_curve_index: int = 0, tadm_algorithm: bool = False, @@ -5104,7 +4976,7 @@ async def aspirate_core_96( liquid surface (LLD or absolute terms) [0.1mm]. Must be between 0 and 990. Default 250. surface_following_distance_during_mix: surface following distance during mix [0.1mm]. Must be between 0 and 990. Default 0. - speed_of_mix: Speed of mix [0.1ul/s]. Must be between 3 and 5000. + mix_speed: Speed of mix [0.1ul/s]. Must be between 3 and 5000. Default 1000. todo: TODO: 24 hex chars. Must be between 4 and 5000. limit_curve_index: limit curve index. Must be between 0 and 999. Default 0. @@ -5150,8 +5022,8 @@ async def aspirate_core_96( "mix_position_from_liquid_surface must be between 0 and 990" assert 0 <= surface_following_distance_during_mix <= 990, \ "surface_following_distance_during_mix must be between 0 and 990" - assert 3 <= speed_of_mix <= 5000, \ - "speed_of_mix must be between 3 and 5000" + assert 3 <= mix_speed <= 5000, \ + "mix_speed must be between 3 and 5000" assert 0 <= limit_curve_index <= 999, "limit_curve_index must be between 0 and 999" assert 0 <= recording_mode <= 2, "recording_mode must be between 0 and 2" @@ -5192,7 +5064,7 @@ async def aspirate_core_96( hc=f"{mix_cycles:02}", hp=f"{mix_position_from_liquid_surface:03}", mj=f"{surface_following_distance_during_mix:03}", - hs=f"{speed_of_mix:04}", + hs=f"{mix_speed:04}", cw=channel_pattern_hex, cr=f"{limit_curve_index:03}", cj=tadm_algorithm, @@ -5232,7 +5104,7 @@ async def dispense_core_96( mixing_cycles: int = 0, mixing_position_from_liquid_surface: int = 250, surface_following_distance_during_mixing: int = 0, - speed_of_mixing: int = 1000, + mix_speed: int = 1000, channel_pattern: List[bool] = [True]*12*8, limit_curve_index: int = 0, tadm_algorithm: bool = False, @@ -5289,7 +5161,7 @@ async def dispense_core_96( surface (LLD or absolute terms) [0.1mm]. Must be between 0 and 990. Default 250. surface_following_distance_during_mixing: surface following distance during mixing [0.1mm]. Must be between 0 and 990. Default 0. - speed_of_mixing: Speed of mixing [0.1ul/s]. Must be between 3 and 5000. Default 1000. + mix_speed: Speed of mixing [0.1ul/s]. Must be between 3 and 5000. Default 1000. channel_pattern: list of 96 boolean values limit_curve_index: limit curve index. Must be between 0 and 999. Default 0. tadm_algorithm: TADM algorithm. Default False. @@ -5336,7 +5208,7 @@ async def dispense_core_96( "mixing_position_from_liquid_surface must be between 0 and 990" assert 0 <= surface_following_distance_during_mixing <= 990, \ "surface_following_distance_during_mixing must be between 0 and 990" - assert 3 <= speed_of_mixing <= 5000, "speed_of_mixing must be between 3 and 5000" + assert 3 <= mix_speed <= 5000, "mix_speed must be between 3 and 5000" assert 0 <= limit_curve_index <= 999, "limit_curve_index must be between 0 and 999" assert 0 <= recording_mode <= 2, "recording_mode must be between 0 and 2" @@ -5378,7 +5250,7 @@ async def dispense_core_96( hc=f"{mixing_cycles:02}", hp=f"{mixing_position_from_liquid_surface:03}", mj=f"{surface_following_distance_during_mixing:03}", - hs=f"{speed_of_mixing:04}", + hs=f"{mix_speed:04}", cw=channel_pattern_hex, cr=f"{limit_curve_index:03}", cj=tadm_algorithm, diff --git a/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py b/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py index b171a5554c..8a35469d36 100644 --- a/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py +++ b/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py @@ -4,6 +4,10 @@ from pylabrobot.liquid_handling import LiquidHandler from pylabrobot.liquid_handling.standard import Pickup, GripDirection +from pylabrobot.liquid_handling.liquid_classes.hamilton.star import ( + StandardVolumeFilter_Water_DispenseSurface, + StandardVolumeFilter_Water_DispenseJet_Empty, + HighVolumeFilter_96COREHead1000ul_Water_DispenseSurface_Empty) from pylabrobot.plate_reading import PlateReader from pylabrobot.plate_reading.plate_reader_tests import MockPlateReaderBackend from pylabrobot.resources import ( @@ -224,6 +228,10 @@ def __init__(self, name: str): await self.lh.setup() + self.hlc = StandardVolumeFilter_Water_DispenseSurface.copy() + self.hlc.aspiration_air_transport_volume = 0 + self.hlc.dispense_air_transport_volume = 0 + async def asyncTearDown(self): await self.lh.stop() @@ -370,7 +378,9 @@ async def test_aspirate56(self): self.plate.lid.unassign() for well in self.plate.get_items(["A1", "B1"]): well.tracker.set_liquids([(None, 100 * 1.072)]) # liquid class correction - await self.lh.aspirate(self.plate["A1", "B1"], vols=[100, 100], use_channels=[4, 5]) + corrected_vol = self.hlc.compute_corrected_volume(100) + await self.lh.aspirate(self.plate["A1", "B1"], vols=[corrected_vol, corrected_vol], + use_channels=[4, 5], **self.hlc.make_asp_kwargs(2)) self._assert_command_sent_once("C0ASid0004at0 0 0 0 0 0 0&tm0 0 0 0 1 1 0&xp00000 00000 00000 " "00000 02983 02983 00000&yp0000 0000 0000 0000 1457 1367 0000&th2450te2450lp2000 2000 2000 " "2000 2000 2000 2000&ch000 000 000 000 000 000 000&zl1866 1866 1866 1866 1866 1866 1866&" @@ -394,7 +404,8 @@ async def test_single_channel_aspiration(self): self.plate.lid.unassign() well = self.plate.get_item("A1") well.tracker.set_liquids([(None, 100 * 1.072)]) # liquid class correction - await self.lh.aspirate([well], vols=[100]) + await self.lh.aspirate([well], vols=[self.hlc.compute_corrected_volume(100)], + **self.hlc.make_asp_kwargs(1)) # This passes the test, but is not the real command. self._assert_command_sent_once( @@ -413,7 +424,8 @@ async def test_single_channel_aspiration_liquid_height(self): self.plate.lid.unassign() well = self.plate.get_item("A1") well.tracker.set_liquids([(None, 100 * 1.072)]) # liquid class correction - await self.lh.aspirate([well], vols=[100], liquid_height=[10]) + await self.lh.aspirate([well], vols=[self.hlc.compute_corrected_volume(100)], + liquid_height=[10], **self.hlc.make_asp_kwargs(1)) # This passes the test, but is not the real command. self._assert_command_sent_once( @@ -433,7 +445,9 @@ async def test_multi_channel_aspiration(self): wells = self.plate.get_items("A1:B1") for well in wells: well.tracker.set_liquids([(None, 100 * 1.072)]) # liquid class correction - await self.lh.aspirate(self.plate["A1:B1"], vols=[100]*2) + corrected_vol = self.hlc.compute_corrected_volume(100) + await self.lh.aspirate(self.plate["A1:B1"], vols=[corrected_vol]*2, + **self.hlc.make_asp_kwargs(2)) # This passes the test, but is not the real command. self._assert_command_sent_once( @@ -449,8 +463,10 @@ async def test_multi_channel_aspiration(self): async def test_aspirate_single_resource(self): self.lh.update_head_state({i: self.tip_rack.get_tip(i) for i in range(5)}) + corrected_vol = self.hlc.compute_corrected_volume(10) with no_volume_tracking(): - await self.lh.aspirate([self.bb]*5,vols=[10]*5, use_channels=[0,1,2,3,4], liquid_height=[1]*5) + await self.lh.aspirate([self.bb]*5,vols=[corrected_vol]*5, use_channels=[0,1,2,3,4], + liquid_height=[1]*5, **self.hlc.make_asp_kwargs(5)) self._assert_command_sent_once( "C0ASid0002at0 0 0 0 0 0&tm1 1 1 1 1 0&xp04865 04865 04865 04865 04865 00000&yp2098 1962 " "1825 1688 1552 0000&th2450te2450lp2000 2000 2000 2000 2000 2000&ch000 000 000 000 000 000&" @@ -469,9 +485,11 @@ async def test_aspirate_single_resource(self): async def test_dispense_single_resource(self): self.lh.update_head_state({i: self.tip_rack.get_tip(i) for i in range(5)}) + hlc = StandardVolumeFilter_Water_DispenseJet_Empty + corrected_vol = hlc.compute_corrected_volume(10) with no_volume_tracking(): - await self.lh.dispense([self.bb]*5, vols=[10]*5, use_channels=[0,1,2,3,4], - liquid_height=[1]*5, blow_out=[True]*5, jet=[True]*5) + await self.lh.dispense([self.bb]*5, vols=[corrected_vol]*5, liquid_height=[1]*5, + jet=[True]*5, blow_out=[True]*5, **hlc.make_disp_kwargs(5)) self._assert_command_sent_once( "C0DSid0002dm1 1 1 1 1 1&tm1 1 1 1 1 0&xp04865 04865 04865 04865 04865 00000&yp2098 1962 " "1825 1688 1552 0000&zx1200 1200 1200 1200 1200 1200&lp2000 2000 2000 2000 2000 2000&zl1210 " @@ -489,8 +507,11 @@ async def test_single_channel_dispense(self): self.lh.update_head_state({0: self.tip_rack.get_tip("A1")}) assert self.plate.lid is not None self.plate.lid.unassign() + hlc = StandardVolumeFilter_Water_DispenseJet_Empty + corrected_vol = hlc.compute_corrected_volume(100) with no_volume_tracking(): - await self.lh.dispense(self.plate["A1"], vols=[100], jet=[True], blow_out=[True]) + await self.lh.dispense(self.plate["A1"], vols=[corrected_vol], jet=[True], blow_out=[True], + **hlc.make_disp_kwargs(1)) self._assert_command_sent_once( "C0DSid0002dm1 1&tm1 0&xp02983 00000&yp1457 0000&zx1866 1866&lp2000 2000&zl1866 1866&" "po0100 0100&ip0000 0000&it0 0&fp0000 0000&zu0032 0032&zr06180 06180&th2450te2450" @@ -504,8 +525,11 @@ async def test_multi_channel_dispense(self): # TODO: Hamilton liquid classes assert self.plate.lid is not None self.plate.lid.unassign() + hlc = StandardVolumeFilter_Water_DispenseJet_Empty + corrected_vol = hlc.compute_corrected_volume(100) with no_volume_tracking(): - await self.lh.dispense(self.plate["A1:B1"], vols=[100]*2, jet=[True]*2, blow_out=[True]*2) + await self.lh.dispense(self.plate["A1:B1"], vols=[corrected_vol]*2, jet=[True]*2, + blow_out=[True]*2, **hlc.make_disp_kwargs(2)) self._assert_command_sent_once( "C0DSid0002dm1 1 1&tm1 1 0&xp02983 02983 00000&yp1457 1367 0000&zx1866 1866 1866&lp2000 2000 " @@ -553,7 +577,9 @@ async def test_core_96_aspirate(self): # TODO: Hamilton liquid classes assert self.plate.lid is not None self.plate.lid.unassign() - await self.lh.aspirate96(self.plate, volume=100, blow_out=True) + hlc = HighVolumeFilter_96COREHead1000ul_Water_DispenseSurface_Empty + corrected_volume = hlc.compute_corrected_volume(100) + await self.lh.aspirate96(self.plate, volume=corrected_volume, **hlc.make_asp96_kwargs()) # volume used to be 01072, but that was generated using a non-core liquid class. self._assert_command_sent_once( @@ -568,12 +594,14 @@ async def test_core_96_dispense(self): await self.lh.pick_up_tips96(self.tip_rack2) # pick up high volume tips if self.plate.lid is not None: self.plate.lid.unassign() - await self.lh.aspirate96(self.plate, 100, blow_out=True) # aspirate first + hlc = HighVolumeFilter_96COREHead1000ul_Water_DispenseSurface_Empty + corrected_volume = hlc.compute_corrected_volume(100) + await self.lh.aspirate96(self.plate, corrected_volume, **hlc.make_asp96_kwargs()) with no_volume_tracking(): - await self.lh.dispense96(self.plate, 100, blow_out=True) + await self.lh.dispense96(self.plate, corrected_volume, blow_out=True, + **hlc.make_disp96_kwargs()) - # volume used to be 01072, but that was generated using a non-core liquid class. self._assert_command_sent_once( "C0EDid0001da3xs02983xd0yh1457zh2450ze2450lz1999zt1866zm1866iw000ix0fh000df01083dg1200vt050" "bv00000cm0cs1bs0020wh00hv00000hc00hp000hs1200es0050ev000zv0032ej00zq06180mj000cj0cx0cr000" diff --git a/pylabrobot/liquid_handling/backends/hamilton/vantage.py b/pylabrobot/liquid_handling/backends/hamilton/vantage.py index c16a29bb71..2f5cfad047 100644 --- a/pylabrobot/liquid_handling/backends/hamilton/vantage.py +++ b/pylabrobot/liquid_handling/backends/hamilton/vantage.py @@ -7,8 +7,7 @@ from typing import Dict, List, Optional, Sequence, Union, cast from pylabrobot.liquid_handling.backends.hamilton.base import HamiltonLiquidHandler -from pylabrobot.liquid_handling.liquid_classes.hamilton import ( - HamiltonLiquidClass, get_vantage_liquid_class) +from pylabrobot.liquid_handling.liquid_classes.hamilton import HamiltonLiquidClass from pylabrobot.liquid_handling.standard import ( Pickup, PickupTipRack, @@ -22,7 +21,7 @@ DispenseContainer, Move ) -from pylabrobot.resources import Coordinate, Liquid, Resource, TipRack, Well +from pylabrobot.resources import Coordinate, Resource, TipRack, Well from pylabrobot.resources.ml_star import HamiltonTip, TipPickupMethod, TipSize @@ -539,8 +538,6 @@ async def aspirate( self, ops: List[Aspiration], use_channels: List[int], - jet: Optional[List[bool]] = None, - blow_out: Optional[List[bool]] = None, hlcs: Optional[List[Optional[HamiltonLiquidClass]]] = None, type_of_aspiration: Optional[List[int]] = None, @@ -588,41 +585,16 @@ async def aspirate( blow_out: Whether to search for a "blow out" liquid class. This is only used on dispense. Note that in the VENUS liquid editor, the term "empty" is used for this, but in the firmware documentation, "empty" is used for a different mode (dm4). - hlcs: The Hamiltonian liquid classes to use. If `None`, the liquid classes will be - determined automatically based on the tip and liquid used. """ + if hlcs is not None: + raise NotImplementedError("hlcs is deprecated") + x_positions, y_positions, channels_involved = \ self._ops_to_fw_positions(ops, use_channels) - if jet is None: - jet = [False]*len(ops) - if blow_out is None: - blow_out = [False]*len(ops) - - if hlcs is None: - hlcs = [] - for j, bo, op in zip(jet, blow_out, ops): - liquid = Liquid.WATER # default to WATER - # [-1][0]: get last liquid in well, [0] is indexing into the tuple - if len(op.liquids) > 0 and op.liquids[-1][0] is not None: - liquid = op.liquids[-1][0] - hlcs.append(get_vantage_liquid_class( - tip_volume=op.tip.maximal_volume, - is_core=False, - is_tip=True, - has_filter=op.tip.has_filter, - liquid=liquid, - jet=j, - blow_out=bo - )) - self._assert_valid_resources([op.resource for op in ops]) - # correct volumes using the liquid class - volumes = [hlc.compute_corrected_volume(op.volume) if hlc is not None else op.volume - for op, hlc in zip(ops, hlcs)] - well_bottoms = [op.resource.get_absolute_location().z + op.offset.z + \ op.resource.material_z_thickness for op in ops] liquid_surfaces_no_lld = liquid_surface_at_function_without_lld or [wb + (op.liquid_height or 0) @@ -632,12 +604,10 @@ async def aspirate( (2.7-1 if isinstance(op.resource, Well) else 5) #? for wb, op in zip(well_bottoms, ops)] - flow_rates = [ - op.flow_rate or (hlc.aspiration_flow_rate if hlc is not None else 100) - for op, hlc in zip(ops, hlcs)] - blow_out_air_volumes = [(op.blow_out_air_volume or - (hlc.dispense_blow_out_volume if hlc is not None else 0)) - for op, hlc in zip(ops, hlcs)] + flow_rates = [op.flow_rate or 100 for op in ops] + blow_out_air_volumes = [op.blow_out_air_volume or 0 for op in ops] + + print(mix_speed) return await self.pip_aspirate( x_position=x_positions, @@ -663,11 +633,9 @@ async def aspirate( immersion_depth=[round(id_*10) for id_ in immersion_depth or [0]*len(ops)], surface_following_distance=[round(sfd*10) for sfd in surface_following_distance or [0]*len(ops)], - aspiration_volume=[round(vol*100) for vol in volumes], + aspiration_volume=[round(op.volume*100) for op in ops], aspiration_speed=[round(fr * 10) for fr in flow_rates], - transport_air_volume=[round(tav*10) for tav in - transport_air_volume or [hlc.aspiration_air_transport_volume if hlc is not None else 0 - for hlc in hlcs]], + transport_air_volume=[round(tav*10) for tav in transport_air_volume or [0]*len(ops)], blow_out_air_volume=[round(bav*100) for bav in blow_out_air_volumes], pre_wetting_volume=[round(pwv*100) for pwv in pre_wetting_volume or [0]*len(ops)], lld_mode=lld_mode or [0]*len(ops), @@ -742,8 +710,6 @@ async def dispense( Args: ops: The aspiration operations. use_channels: The channels to use. - hlcs: The Hamiltonian liquid classes to use. If `None`, the liquid classes will be - determined automatically based on the tip and liquid used. jet: Whether to use jetting for each dispense. Defaults to `False` for all. Used for determining the dispense mode. True for dispense mode 0 or 1. @@ -755,6 +721,9 @@ async def dispense( documentation. Dispense mode 4. """ + if hlcs is not None: + raise NotImplementedError("hlcs is deprecated") + x_positions, y_positions, channels_involved = \ self._ops_to_fw_positions(ops, use_channels) @@ -765,29 +734,8 @@ async def dispense( if blow_out is None: blow_out = [False]*len(ops) - if hlcs is None: - hlcs = [] - for j, bo, op in zip(jet, blow_out, ops): - liquid = Liquid.WATER # default to WATER - # [-1][0]: get last liquid in tip, [0] is indexing into the tuple - if len(op.liquids) > 0 and op.liquids[-1][0] is not None: - liquid = op.liquids[-1][0] - hlcs.append(get_vantage_liquid_class( - tip_volume=op.tip.maximal_volume, - is_core=False, - is_tip=True, - has_filter=op.tip.has_filter, - liquid=liquid, - jet=j, - blow_out=bo, - )) - self._assert_valid_resources([op.resource for op in ops]) - # correct volumes using the liquid class - volumes = [hlc.compute_corrected_volume(op.volume) if hlc is not None else op.volume - for op, hlc in zip(ops, hlcs)] - well_bottoms = [op.resource.get_absolute_location().z + op.offset.z + \ op.resource.material_z_thickness for op in ops] liquid_surfaces_no_lld = [wb + (op.liquid_height or 0) @@ -797,13 +745,9 @@ async def dispense( (2.7-1 if isinstance(op.resource, Well) else 5) #? for wb, op in zip(well_bottoms, ops)] - flow_rates = [ - op.flow_rate or (hlc.dispense_flow_rate if hlc is not None else 100) - for op, hlc in zip(ops, hlcs)] + flow_rates = [op.flow_rate or 100 for op in ops] - blow_out_air_volumes = [(op.blow_out_air_volume or - (hlc.dispense_blow_out_volume if hlc is not None else 0)) - for op, hlc in zip(ops, hlcs)] + blow_out_air_volumes = [op.blow_out_air_volume or 0 for op in ops] type_of_dispensing_mode = type_of_dispensing_mode or \ [_get_dispense_mode(jet=jet[i], empty=empty[i], blow_out=blow_out[i]) @@ -831,13 +775,11 @@ async def dispense( minimal_traverse_height_at_begin_of_command or [self._traversal_height]*len(ops)], minimal_height_at_command_end= [round(mh*10) for mh in minimal_height_at_command_end or [self._traversal_height]*len(ops)], - dispense_volume=[round(vol*100) for vol in volumes], + dispense_volume=[round(op.volume*100) for op in ops], dispense_speed=[round(fr*10) for fr in flow_rates], cut_off_speed=[round(cs*10) for cs in cut_off_speed or [250]*len(ops)], stop_back_volume=[round(sbv*100) for sbv in stop_back_volume or [0]*len(ops)], - transport_air_volume=[round(tav*10) for tav in transport_air_volume or - [hlc.dispense_air_transport_volume if hlc is not None else 0 - for hlc in hlcs]], + transport_air_volume=[round(tav*10) for tav in transport_air_volume or [0]*len(ops)], blow_out_air_volume=[round(boav*100) for boav in blow_out_air_volumes], lld_mode=lld_mode or [0]*len(ops), side_touch_off_distance=round(side_touch_off_distance*10), @@ -917,8 +859,6 @@ async def drop_tips96( async def aspirate96( self, aspiration: Union[AspirationPlate, AspirationContainer], - jet: bool = False, - blow_out: bool = False, hlc: Optional[HamiltonLiquidClass] = None, type_of_aspiration: int = 0, @@ -930,7 +870,6 @@ async def aspirate96( immersion_depth: float = 0, surface_following_distance: float = 0, transport_air_volume: Optional[float] = None, - blow_out_air_volume: Optional[float] = None, pre_wetting_volume: float = 0, lld_mode: int = 0, lld_sensitivity: int = 4, @@ -946,18 +885,11 @@ async def aspirate96( tadm_algorithm_on_off: int = 0, recording_mode: int = 0, ): - """ Aspirate from a plate. - - Args: - jet: Whether to find a liquid class with "jet" mode. Only used on dispense. - blow_out: Whether to find a liquid class with "blow out" mode. Only used on dispense. Note - that this is called "empty" in the VENUS liquid editor, but "blow out" in the firmware - documentation. - hlc: The Hamiltonian liquid classes to use. If `None`, the liquid classes will be - determined automatically based on the tip and liquid used in the first well. - """ # assert self.core96_head_installed, "96 head must be installed" + if hlc is not None: + raise NotImplementedError("hlc is deprecated") + if isinstance(aspiration, AspirationPlate): top_left_well = aspiration.wells[0] position = top_left_well.get_absolute_location() + top_left_well.center() + \ @@ -973,33 +905,20 @@ async def aspirate96( liquid_height = position.z + (aspiration.liquid_height or 0) - tip = aspiration.tips[0] - liquid_to_be_aspirated = Liquid.WATER # default to water - if len(aspiration.liquids[0]) > 0 and aspiration.liquids[0][-1][0] is not None: - # first part of tuple in last liquid of first well - liquid_to_be_aspirated = aspiration.liquids[0][-1][0] - if hlc is None: - hlc = get_vantage_liquid_class( - tip_volume=tip.maximal_volume, - is_core=True, - is_tip=True, - has_filter=tip.has_filter, - liquid=liquid_to_be_aspirated, - jet=jet, - blow_out=blow_out - ) - - volume = hlc.compute_corrected_volume(aspiration.volume) if hlc is not None \ - else aspiration.volume - - transport_air_volume = transport_air_volume or \ - (hlc.aspiration_air_transport_volume if hlc is not None else 0) - blow_out_air_volume = blow_out_air_volume or \ - (hlc.aspiration_blow_out_volume if hlc is not None else 0) - flow_rate = aspiration.flow_rate or (hlc.aspiration_flow_rate if hlc is not None else 250) - swap_speed = swap_speed or (hlc.aspiration_swap_speed if hlc is not None else 100) - settling_time = settling_time or \ - (hlc.aspiration_settling_time if hlc is not None else 5) + if transport_air_volume is None: + transport_air_volume = 0 + if aspiration.blow_out_air_volume is None: + blow_out_air_volume = 0.0 + else: + blow_out_air_volume = aspiration.blow_out_air_volume + if aspiration.flow_rate is None: + flow_rate = 250.0 + else: + flow_rate = aspiration.flow_rate + if swap_speed is None: + swap_speed = 100 + if settling_time is None: + settling_time = 5 return await self.core96_aspiration_of_liquid( x_position=round(position.x * 10), @@ -1018,7 +937,7 @@ async def aspirate96( tube_2nd_section_ratio=round(tube_2nd_section_ratio*10), immersion_depth=round(immersion_depth*10), surface_following_distance=round(surface_following_distance*10), - aspiration_volume=round(volume * 100), + aspiration_volume=round(aspiration.volume * 100), aspiration_speed=round(flow_rate * 10), transport_air_volume=round(transport_air_volume*10), blow_out_air_volume=round(blow_out_air_volume*100), @@ -1058,7 +977,6 @@ async def dispense96( cut_off_speed: float = 250.0, stop_back_volume: float = 0, transport_air_volume: Optional[float] = None, - blow_out_air_volume: Optional[float] = None, lld_mode: int = 0, lld_sensitivity: int = 4, side_touch_off_distance: float = 0, @@ -1089,6 +1007,9 @@ async def dispense96( determined based on the jet, blow_out, and empty parameters. """ + if hlc is not None: + raise NotImplementedError("hlc is deprecated") + if isinstance(dispense, DispensePlate): top_left_well = dispense.wells[0] position = top_left_well.get_absolute_location() + top_left_well.center() + \ @@ -1104,35 +1025,24 @@ async def dispense96( liquid_height = position.z + (dispense.liquid_height or 0) + 10 - tip = dispense.tips[0] - liquid_to_be_dispensed = Liquid.WATER # default to WATER - if len(dispense.liquids[0]) > 0 and dispense.liquids[0][-1][0] is not None: - # first part of tuple in last liquid of first well - liquid_to_be_dispensed = dispense.liquids[0][-1][0] - if hlc is None: - hlc = get_vantage_liquid_class( - tip_volume=tip.maximal_volume, - is_core=True, - is_tip=True, - has_filter=tip.has_filter, - liquid=liquid_to_be_dispensed, - jet=jet, - blow_out=blow_out # see method docstring - ) - volume = hlc.compute_corrected_volume(dispense.volume) if hlc is not None \ - else dispense.volume - - transport_air_volume = transport_air_volume or \ - (hlc.dispense_air_transport_volume if hlc is not None else 0) - blow_out_air_volume = blow_out_air_volume or \ - (hlc.dispense_blow_out_volume if hlc is not None else 0) - flow_rate = dispense.flow_rate or (hlc.dispense_flow_rate if hlc is not None else 250) - swap_speed = swap_speed or (hlc.dispense_swap_speed if hlc is not None else 100) - settling_time = settling_time or \ - (hlc.dispense_settling_time if hlc is not None else 5) - mix_speed = mix_speed or (hlc.dispense_mix_flow_rate if hlc is not None else 100) - type_of_dispensing_mode = type_of_dispensing_mode or \ - _get_dispense_mode(jet=jet, empty=empty, blow_out=blow_out) + if transport_air_volume is None: + transport_air_volume = 0 + if dispense.blow_out_air_volume is None: + blow_out_air_volume = 0.0 + else: + blow_out_air_volume = dispense.blow_out_air_volume + if dispense.flow_rate is None: + flow_rate = 250.0 + else: + flow_rate = dispense.flow_rate + if swap_speed is None: + swap_speed = 100 + if settling_time is None: + settling_time = 5 + if mix_speed is None: + mix_speed = 100 + if type_of_dispensing_mode is None: + type_of_dispensing_mode = _get_dispense_mode(jet=jet, empty=empty, blow_out=blow_out) return await self.core96_dispensing_of_liquid( x_position=round(position.x * 10), @@ -1151,7 +1061,7 @@ async def dispense96( round((minimal_traverse_height_at_begin_of_command or self._traversal_height) * 10), minimal_height_at_command_end= round((minimal_height_at_command_end or self._traversal_height)*10), - dispense_volume=round(volume * 100), + dispense_volume=round(dispense.volume * 100), dispense_speed=round(flow_rate * 10), cut_off_speed=round(cut_off_speed * 10), stop_back_volume=round(stop_back_volume * 100), diff --git a/pylabrobot/liquid_handling/backends/hamilton/vantage_tests.py b/pylabrobot/liquid_handling/backends/hamilton/vantage_tests.py index 56a920775c..9552d6c1b2 100644 --- a/pylabrobot/liquid_handling/backends/hamilton/vantage_tests.py +++ b/pylabrobot/liquid_handling/backends/hamilton/vantage_tests.py @@ -2,11 +2,15 @@ import unittest from pylabrobot.liquid_handling import LiquidHandler +from pylabrobot.liquid_handling.liquid_classes.hamilton.vantage import ( + HighVolumeFilter_Water_DispenseSurface_Part, + HighVolumeFilter_Water_DispenseSurface_Empty, + HighVolumeFilter_96COREHead1000ul_Water_DispenseJet_Empty) +from pylabrobot.liquid_handling.standard import Pickup from pylabrobot.resources import ( Coordinate, TIP_CAR_480_A00, PLT_CAR_L5AC_A00, HT_L, LT_L, Cor_96_wellplate_360ul_Fb, ) from pylabrobot.resources.hamilton import VantageDeck -from pylabrobot.liquid_handling.standard import Pickup from .vantage import ( Vantage, @@ -276,25 +280,29 @@ async def test_small_tip_drop(self): async def test_aspirate(self): await self.lh.pick_up_tips(self.tip_rack["A1"]) # pick up tips first - await self.lh.aspirate(self.plate["A1"], vols=[100]) + hlc = HighVolumeFilter_Water_DispenseSurface_Part + corrected_volume = hlc.compute_corrected_volume(100) + await self.lh.aspirate(self.plate["A1"], vols=[corrected_volume], **hlc.make_asp_kwargs(1)) self._assert_command_sent_once( "A1PMDAid0248at0&tm1 0&xp05683 0&yp1457 0 &th2450&te2450&lp1990&" "ch000&zl1866&zx1866&ip0000&fp0000&av010830&as2500&ta000&ba00000&oa000&lm0&ll4&lv4&de0020&" - "wt10&mv00000&mc00&mp000&ms2500&gi000&gj0gk0zu0000&zr00000&mh0000&zo005&po0109&dj0la0&lb0&" + "wt10&mv00000&mc00&mp000&ms1200&gi000&gj0gk0zu0000&zr00000&mh0000&zo005&po0109&dj0la0&lb0&" "lc0&", ASPIRATE_FORMAT) async def test_dispense(self): await self.lh.pick_up_tips(self.tip_rack["A1"]) # pick up tips first - await self.lh.aspirate(self.plate["A1"], vols=[100]) - await self.lh.dispense(self.plate["A2"], vols=[100], liquid_height=[5], jet=[False], - blow_out=[True]) + hlc = HighVolumeFilter_Water_DispenseSurface_Empty + corrected_volume = hlc.compute_corrected_volume(100) + await self.lh.aspirate(self.plate["A1"], vols=[corrected_volume], **hlc.make_asp_kwargs(1)) + await self.lh.dispense(self.plate["A2"], vols=[corrected_volume], liquid_height=[5], + jet=[False], blow_out=[True], **hlc.make_disp_kwargs(1)) self._assert_command_sent_once( "A1PMDDid0253dm3&tm1 0&xp05773 0&yp1457 0&zx1866&lp1990&zl1916&" "ip0000&fp0021&th2450&te2450&dv010830&ds1200&ss2500&rv000&ta050&ba00000&lm0&zo005&ll1&lv1&" - "de0010&mv00000&mc00&mp000&ms0010&wt00&gi000&gj0gk0zu0000&dj00zr00000&mh0000&po0050&la0&", + "de0020&mv00000&mc00&mp000&ms1200&wt00&gi000&gj0gk0zu0000&dj00zr00000&mh0000&po0050&la0&", DISPENSE_FORMAT) async def test_zero_volume_liquid_handling(self): @@ -318,9 +326,11 @@ async def test_tip_drop96(self): async def test_aspirate96(self): await self.lh.pick_up_tips96(self.tip_rack) - await self.lh.aspirate96(self.plate, volume=100, jet=True, blow_out=True) + hlc = HighVolumeFilter_96COREHead1000ul_Water_DispenseJet_Empty + await self.lh.aspirate96(self.plate, volume=hlc.compute_corrected_volume(100), + **hlc.make_asp96_kwargs()) self._assert_command_sent_once( - "A1HMDAid0236at0xp05683yp1457th2450te2450lp1990zl1866zx1866ip000fp000av010720as2500ta050" + "A1HMDAid0236at0xp05683yp1457th2450te2450lp1990zl1866zx1866ip000fp000av010920as2500ta050" "ba004000oa00000lm0ll4de0020wt10mv00000mc00mp000ms2500zu0000zr00000mh000gj0gk0gi000" "cwFFFFFFFFFFFFFFFFFFFFFFFFpo0050", {"xp": "int", "yp": "int", "th": "int", "te": "int", "lp": "int", "zl": "int", "zx": "int", @@ -331,10 +341,13 @@ async def test_aspirate96(self): async def test_dispense96(self): await self.lh.pick_up_tips96(self.tip_rack) - await self.lh.aspirate96(self.plate, volume=100, jet=True, blow_out=True) - await self.lh.dispense96(self.plate, volume=100, jet=True, blow_out=True) + hlc = HighVolumeFilter_96COREHead1000ul_Water_DispenseJet_Empty + await self.lh.aspirate96(self.plate, volume=hlc.compute_corrected_volume(100), + **hlc.make_asp96_kwargs()) + await self.lh.dispense96(self.plate, volume=hlc.compute_corrected_volume(100), + jet=True, blow_out=True, **hlc.make_disp96_kwargs()) self._assert_command_sent_once( - "A1HMDDid0238dm1xp05683yp1457th2450te2450lp1990zl1966zx1866ip000fp029dv010720ds4000ta050" + "A1HMDDid0238dm1xp05683yp1457th2450te2450lp1990zl1966zx1866ip000fp029dv10920ds4000ta050" "ba004000lm0ll4de0010wt00mv00000mc00mp000ms0010ss2500rv000zu0000dj00zr00000mh000gj0gk0gi000" "cwFFFFFFFFFFFFFFFFFFFFFFFFpo0050", {"xp": "int", "yp": "int", "th": "int", "te": "int", "lp": "int", "zl": "int", "zx": "int", diff --git a/pylabrobot/liquid_handling/liquid_classes/hamilton/base.py b/pylabrobot/liquid_handling/liquid_classes/hamilton/base.py index 2f411eedf2..2081d86b9f 100644 --- a/pylabrobot/liquid_handling/liquid_classes/hamilton/base.py +++ b/pylabrobot/liquid_handling/liquid_classes/hamilton/base.py @@ -1,4 +1,4 @@ -from typing import Any, Dict +from typing import Any, Dict, List class HamiltonLiquidClass: @@ -111,3 +111,49 @@ def serialize(self) -> Dict[str, Any]: "dispense_stop_flow_rate": self.dispense_stop_flow_rate, "dispense_stop_back_volume": self.dispense_stop_back_volume, } + + def copy(self) -> "HamiltonLiquidClass": + return HamiltonLiquidClass(**self.serialize()) + + def make_asp_kwargs(self, num_channels: int) -> Dict[str, List[Any]]: + return { + "flow_rates": [self.aspiration_flow_rate] * num_channels, + "mix_speed": [self.aspiration_mix_flow_rate] * num_channels, + "transport_air_volume": [self.aspiration_air_transport_volume] * num_channels, + "blow_out_air_volume": [self.aspiration_blow_out_volume] * num_channels, + "swap_speed": [self.aspiration_swap_speed] * num_channels, + "settling_time": [self.aspiration_settling_time] * num_channels, + "clot_detection_height": [self.aspiration_clot_retract_height] * num_channels, + } + + def make_disp_kwargs(self, num_channels: int) -> Dict[str, List[Any]]: + return { + "flow_rates": [self.dispense_flow_rate] * num_channels, + "mix_speed": [self.dispense_mix_flow_rate] * num_channels, + "transport_air_volume": [self.dispense_air_transport_volume] * num_channels, + "blow_out_air_volume": [self.dispense_blow_out_volume] * num_channels, + "swap_speed": [self.dispense_swap_speed] * num_channels, + "settling_time": [self.dispense_settling_time] * num_channels, + "stop_back_volume": [self.dispense_stop_back_volume] * num_channels, + } + + def make_asp96_kwargs(self) -> Dict[str, Any]: + return { + "flow_rate": self.aspiration_flow_rate, + "mix_speed": self.aspiration_mix_flow_rate, + "transport_air_volume": self.aspiration_air_transport_volume, + "blow_out_air_volume": self.aspiration_blow_out_volume, + "swap_speed": self.aspiration_swap_speed, + "settling_time": self.aspiration_settling_time, + } + + def make_disp96_kwargs(self) -> Dict[str, Any]: + return { + "flow_rate": self.dispense_flow_rate, + "mix_speed": self.dispense_mix_flow_rate, + "transport_air_volume": self.dispense_air_transport_volume, + "blow_out_air_volume": self.dispense_blow_out_volume, + "swap_speed": self.dispense_swap_speed, + "settling_time": self.dispense_settling_time, + "stop_back_volume": self.dispense_stop_back_volume, + } diff --git a/pylabrobot/liquid_handling/liquid_classes/hamilton/star.py b/pylabrobot/liquid_handling/liquid_classes/hamilton/star.py index 6c91f61ae2..acb955a926 100644 --- a/pylabrobot/liquid_handling/liquid_classes/hamilton/star.py +++ b/pylabrobot/liquid_handling/liquid_classes/hamilton/star.py @@ -1,6 +1,6 @@ # pylint: skip-file -from typing import Dict, Tuple, Optional +from typing import Dict, Tuple from pylabrobot.resources.liquid import Liquid from pylabrobot.liquid_handling.liquid_classes.hamilton.base import HamiltonLiquidClass @@ -8,42 +8,8 @@ star_mapping: Dict[Tuple[int, bool, bool, bool, Liquid, bool, bool], HamiltonLiquidClass] = {} -def get_star_liquid_class( - tip_volume: float, - is_core: bool, - is_tip: bool, - has_filter: bool, - liquid: Liquid, - jet: bool, - blow_out: bool -) -> Optional[HamiltonLiquidClass]: - """ Get the Hamilton STAR liquid class for the given parameters. - - Args: - tip_volume: The volume of the tip in microliters. - is_core: Whether the tip is a core tip. - is_tip: Whether the tip is a tip tip or a needle. - has_filter: Whether the tip has a filter. - liquid: The liquid to be dispensed. - jet: Whether the liquid is dispensed using a jet. - blow_out: This is called "empty" in the Hamilton Liquid editor and liquid class names, but - "blow out" in the firmware documentation. "Empty" in the firmware documentation means fully - emptying the tip, which is the terminology PyLabRobot adopts. Blow_out is the opposite of - partial dispense. - """ - - # Tip volumes from resources (mostly where they have filters) are slightly different from the ones - # in the liquid class mapping, so we need to map them here. If no mapping is found, we use the - # given maximal volume of the tip. - tip_volume = int({ - 360.0: 300.0, - 1065.0: 1000.0, - 1250.0: 1000.0, - 4367.0: 4000.0, - 5420.0: 5000.0, - }.get(tip_volume, tip_volume)) - - return star_mapping.get((tip_volume, is_core, is_tip, has_filter, liquid, jet, blow_out), None) +def get_star_liquid_class(**kwargs): + raise NotImplementedError("Deprecated, use HamiltonLiquidClass directly") star_mapping[(1000, False, False, False, Liquid.WATER, True, True)] = \ diff --git a/pylabrobot/liquid_handling/liquid_classes/hamilton/vantage.py b/pylabrobot/liquid_handling/liquid_classes/hamilton/vantage.py index b82b691556..6d0dd1926c 100644 --- a/pylabrobot/liquid_handling/liquid_classes/hamilton/vantage.py +++ b/pylabrobot/liquid_handling/liquid_classes/hamilton/vantage.py @@ -1,6 +1,6 @@ # pylint: skip-file -from typing import Dict, Tuple, Optional +from typing import Dict, Tuple from pylabrobot.resources.liquid import Liquid from pylabrobot.liquid_handling.liquid_classes.hamilton.base import HamiltonLiquidClass @@ -8,43 +8,8 @@ vantage_mapping: Dict[Tuple[int, bool, bool, bool, Liquid, bool, bool], HamiltonLiquidClass] = {} -def get_vantage_liquid_class( - tip_volume: float, - is_core: bool, - is_tip: bool, - has_filter: bool, - liquid: Liquid, - jet: bool, - blow_out: bool -) -> Optional[HamiltonLiquidClass]: - """ Get the Hamilton Vantage liquid class for the given parameters. - - Args: - tip_volume: The volume of the tip in microliters. - is_core: Whether the tip is a core tip. - is_tip: Whether the tip is a tip tip or a needle. - has_filter: Whether the tip has a filter. - liquid: The liquid to be dispensed. - jet: Whether the liquid is dispensed using a jet. - blow_out: This is called "empty" in the Hamilton Liquid editor and liquid class names, but - "blow out" in the firmware documentation. "Empty" in the firmware documentation means fully - emptying the tip, which is the terminology PyLabRobot adopts. Blow_out is the opposite of - partial dispense. - """ - - # Tip volumes from resources (mostly where they have filters) are slightly different form the ones - # in the liquid class mapping, so we need to map them here. If no mapping is found, we use the - # given maximal volume of the tip. - tip_volume = int({ - 360.0: 300.0, - 1065.0: 1000.0, - 1250.0: 1000.0, - 4367.0: 4000.0, - 5420.0: 5000.0, - }.get(tip_volume, tip_volume)) - - return vantage_mapping.get((tip_volume, is_core, is_tip, has_filter, liquid, jet, blow_out), None) - +def get_vantage_liquid_class(**kwargs): + raise NotImplementedError("deprecated") vantage_mapping[(1000, False, False, False, Liquid.WATER, True, True)] = \ _1000ulNeedleCRWater_DispenseJet_Empty = HamiltonLiquidClass(