lh.{aspirate,dispense} mix parameter (OT2, STAR, Vantage) (#691)

Author: rickwierenga

pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py

@@ -1712,6 +1712,11 @@ async def aspirate(
           and 360. Defaults to well bottom + liquid height. Should use absolute z.
     """
 
+    if mix_volume is not None or mix_cycles is not None or mix_speed is not None:
+      raise NotImplementedError(
+        "Mixing through backend kwargs is deprecated. Use the `mix` parameter of LiquidHandler.aspirate instead."
+      )
+
     x_positions, y_positions, channels_involved = self._ops_to_fw_positions(ops, use_channels)
 
     n = len(ops)
@@ -1817,17 +1822,12 @@ async def aspirate(
         hlc.aspiration_settling_time if hlc is not None else 0.0 for hlc in hamilton_liquid_classes
       ],
     )
-    mix_volume = _fill_in_defaults(mix_volume, [0.0] * n)
-    mix_cycles = _fill_in_defaults(mix_cycles, [0] * n)
+    mix_volume = [op.mix.volume if op.mix is not None else 0.0 for op in ops]
+    mix_cycles = [op.mix.repetitions if op.mix is not None else 0 for op in ops]
     mix_position_from_liquid_surface = _fill_in_defaults(
       mix_position_from_liquid_surface, [0.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 = [op.mix.flow_rate if op.mix is not None else 100.0 for op in ops]
     mix_surface_following_distance = _fill_in_defaults(mix_surface_following_distance, [0.0] * n)
     limit_curve_index = _fill_in_defaults(limit_curve_index, [0] * n)
 
@@ -2007,6 +2007,11 @@ async def dispense(
         documentation. Dispense mode 4.
     """
 
+    if mix_volume is not None or mix_cycles is not None or mix_speed is not None:
+      raise NotImplementedError(
+        "Mixing through backend kwargs is deprecated. Use the `mix` parameter of LiquidHandler.dispense instead."
+      )
+
     x_positions, y_positions, channels_involved = self._ops_to_fw_positions(ops, use_channels)
 
     n = len(ops)
@@ -2113,17 +2118,12 @@ async def dispense(
         hlc.dispense_settling_time if hlc is not None else 0.0 for hlc in hamilton_liquid_classes
       ],
     )
-    mix_volume = _fill_in_defaults(mix_volume, [0.0] * n)
-    mix_cycles = _fill_in_defaults(mix_cycles, [0] * n)
+    mix_volume = [op.mix.volume if op.mix is not None else 0.0 for op in ops]
+    mix_cycles = [op.mix.repetitions if op.mix is not None else 0 for op in ops]
     mix_position_from_liquid_surface = _fill_in_defaults(
       mix_position_from_liquid_surface, [0.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 = [op.mix.flow_rate if op.mix is not None else 1.0 for op in ops]
     mix_surface_following_distance = _fill_in_defaults(mix_surface_following_distance, [0.0] * n)
     limit_curve_index = _fill_in_defaults(limit_curve_index, [0] * n)
 
@@ -2281,7 +2281,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,
+    speed_of_mix: float = 0.0,
     limit_curve_index: int = 0,
   ):
     """Aspirate using the Core96 head.
@@ -2327,6 +2327,11 @@ async def aspirate96(
       limit_curve_index: The index of the limit curve to use.
     """
 
+    if mix_volume != 0 or mix_cycles != 0 or speed_of_mix != 0:
+      raise NotImplementedError(
+        "Mixing through backend kwargs is deprecated. Use the `mix` parameter of LiquidHandler.aspirate96 instead."
+      )
+
     assert self.core96_head_installed, "96 head must be installed"
 
     # get the first well and tip as representatives
@@ -2395,7 +2400,6 @@ async def aspirate96(
     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)
 
     channel_pattern = [True] * 12 * 8
 
@@ -2444,11 +2448,11 @@ async def aspirate96(
       gamma_lld_sensitivity=gamma_lld_sensitivity,
       swap_speed=round(swap_speed * 10),
       settling_time=round(settling_time * 10),
-      mix_volume=round(mix_volume * 10),
-      mix_cycles=mix_cycles,
+      mix_volume=round(aspiration.mix.volume * 10) if aspiration.mix is not None else 0,
+      mix_cycles=aspiration.mix.repetitions if aspiration.mix is not None else 0,
       mix_position_from_liquid_surface=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),
+      speed_of_mix=round(aspiration.mix.flow_rate * 10) if aspiration.mix is not None else 1200,
       channel_pattern=channel_pattern,
       limit_curve_index=limit_curve_index,
       tadm_algorithm=False,
@@ -2482,7 +2486,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,
+    speed_of_mixing: float = 0.0,
     limit_curve_index: int = 0,
     cut_off_speed: float = 5.0,
     stop_back_volume: float = 0,
@@ -2523,6 +2527,11 @@ async def dispense96(
       stop_back_volume: Unknown.
     """
 
+    if mixing_volume != 0 or mixing_cycles != 0 or speed_of_mixing != 0:
+      raise NotImplementedError(
+        "Mixing through backend kwargs is deprecated. Use the `mix` parameter of LiquidHandler.dispense instead."
+      )
+
     assert self.core96_head_installed, "96 head must be installed"
 
     # get the first well and tip as representatives
@@ -2593,7 +2602,6 @@ async def dispense96(
     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)
 
     channel_pattern = [True] * 12 * 8
 
@@ -2628,11 +2636,11 @@ async def dispense96(
       gamma_lld_sensitivity=gamma_lld_sensitivity,
       swap_speed=round(swap_speed * 10),
       settling_time=round(settling_time * 10),
-      mixing_volume=round(mixing_volume * 10),
-      mixing_cycles=mixing_cycles,
+      mixing_volume=round(dispense.mix.volume * 10) if dispense.mix is not None else 0,
+      mixing_cycles=dispense.mix.repetitions if dispense.mix is not None else 0,
       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),
+      speed_of_mixing=round(dispense.mix.flow_rate * 10) if dispense.mix is not None else 1200,
       channel_pattern=channel_pattern,
       limit_curve_index=limit_curve_index,
       tadm_algorithm=False,

pylabrobot/liquid_handling/backends/hamilton/vantage_backend.py

@@ -622,6 +622,11 @@ async def aspirate(
         determined automatically based on the tip and liquid used.
     """
 
+    if mix_volume is not None or mix_cycles is not None or mix_speed is not None:
+      raise NotImplementedError(
+        "Mixing through backend kwargs is deprecated. Use the `mix` parameter of LiquidHandler.dispense instead."
+      )
+
     x_positions, y_positions, channels_involved = self._ops_to_fw_positions(ops, use_channels)
 
     if jet is None:
@@ -730,12 +735,12 @@ async def aspirate(
       ],
       swap_speed=[round(ss * 10) for ss in swap_speed or [2] * len(ops)],
       settling_time=[round(st * 10) for st in settling_time or [1] * len(ops)],
-      mix_volume=[round(mv * 100) for mv in mix_volume or [0] * len(ops)],
-      mix_cycles=mix_cycles or [0] * len(ops),
+      mix_volume=[round(op.mix.volume * 100) if op.mix is not None else 0 for op in ops],
+      mix_cycles=[op.mix.repetitions if op.mix is not None else 0 for op in ops],
       mix_position_in_z_direction_from_liquid_surface=[
         round(mp) for mp in mix_position_in_z_direction_from_liquid_surface or [0] * len(ops)
       ],
-      mix_speed=[round(ms * 10) for ms in mix_speed or [250] * len(ops)],
+      mix_speed=[round(op.mix.flow_rate * 10) if op.mix is not None else 2500 for op in ops],
       surface_following_distance_during_mixing=[
         round(sfdm * 10) for sfdm in surface_following_distance_during_mixing or [0] * len(ops)
       ],
@@ -808,6 +813,11 @@ async def dispense(
         documentation. Dispense mode 4.
     """
 
+    if mix_volume is not None or mix_cycles is not None or mix_speed is not None:
+      raise NotImplementedError(
+        "Mixing through backend kwargs is deprecated. Use the `mix` parameter of LiquidHandler.dispense instead."
+      )
+
     x_positions, y_positions, channels_involved = self._ops_to_fw_positions(ops, use_channels)
 
     if jet is None:
@@ -922,12 +932,12 @@ async def dispense(
       pressure_lld_sensitivity=pressure_lld_sensitivity or [1] * len(ops),
       swap_speed=[round(ss * 10) for ss in swap_speed or [1] * len(ops)],
       settling_time=[round(st * 10) for st in settling_time or [0] * len(ops)],
-      mix_volume=[round(mv * 100) for mv in mix_volume or [0] * len(ops)],
-      mix_cycles=mix_cycles or [0] * len(ops),
+      mix_volume=[round(op.mix.volume * 100) if op.mix is not None else 0 for op in ops],
+      mix_cycles=[op.mix.repetitions if op.mix is not None else 0 for op in ops],
       mix_position_in_z_direction_from_liquid_surface=[
         round(mp) for mp in mix_position_in_z_direction_from_liquid_surface or [0] * len(ops)
       ],
-      mix_speed=[round(ms * 10) for ms in mix_speed or [1] * len(ops)],
+      mix_speed=[round(op.mix.flow_rate * 100) if op.mix is not None else 10 for op in ops],
       surface_following_distance_during_mixing=[
         round(sfdm * 10) for sfdm in surface_following_distance_during_mixing or [0] * len(ops)
       ],
@@ -1028,7 +1038,7 @@ async def aspirate96(
     mix_cycles: int = 0,
     mix_position_in_z_direction_from_liquid_surface: float = 0,
     surface_following_distance_during_mixing: float = 0,
-    mix_speed: float = 2,
+    mix_speed: float = 0,
     limit_curve_index: int = 0,
     tadm_channel_pattern: Optional[List[bool]] = None,
     tadm_algorithm_on_off: int = 0,
@@ -1046,6 +1056,11 @@ async def aspirate96(
     """
     # assert self.core96_head_installed, "96 head must be installed"
 
+    if mix_volume != 0 or mix_cycles != 0 or mix_speed != 0:
+      raise NotImplementedError(
+        "Mixing through backend kwargs is deprecated. Use the `mix` parameter of LiquidHandler.dispense96 instead."
+      )
+
     if isinstance(aspiration, MultiHeadAspirationPlate):
       plate = aspiration.wells[0].parent
       assert isinstance(plate, Plate), "MultiHeadAspirationPlate well parent must be a Plate"
@@ -1141,15 +1156,15 @@ async def aspirate96(
       lld_sensitivity=lld_sensitivity,
       swap_speed=round(swap_speed * 10),
       settling_time=round(settling_time * 10),
-      mix_volume=round(mix_volume * 100),
-      mix_cycles=mix_cycles,
+      mix_volume=round(aspiration.mix.volume * 100) if aspiration.mix is not None else 0,
+      mix_cycles=aspiration.mix.repetitions if aspiration.mix is not None else 0,
       mix_position_in_z_direction_from_liquid_surface=round(
         mix_position_in_z_direction_from_liquid_surface * 100
       ),
       surface_following_distance_during_mixing=round(
         surface_following_distance_during_mixing * 100
       ),
-      mix_speed=round(mix_speed * 10),
+      mix_speed=round(aspiration.mix.flow_rate * 10) if aspiration.mix is not None else 20,
       limit_curve_index=limit_curve_index,
       tadm_channel_pattern=tadm_channel_pattern,
       tadm_algorithm_on_off=tadm_algorithm_on_off,
@@ -1205,6 +1220,11 @@ async def dispense96(
         determined based on the jet, blow_out, and empty parameters.
     """
 
+    if mix_volume != 0 or mix_cycles != 0 or mix_speed is not None:
+      raise NotImplementedError(
+        "Mixing through backend kwargs is deprecated. Use the `mix` parameter of LiquidHandler.dispense96 instead."
+      )
+
     if isinstance(dispense, MultiHeadDispensePlate):
       plate = dispense.wells[0].parent
       assert isinstance(plate, Plate), "MultiHeadDispensePlate well parent must be a Plate"
@@ -1267,7 +1287,6 @@ async def dispense96(
     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
     )
@@ -1303,13 +1322,13 @@ async def dispense96(
       side_touch_off_distance=round(side_touch_off_distance * 10),
       swap_speed=round(swap_speed * 10),
       settling_time=round(settling_time * 10),
-      mix_volume=round(mix_volume * 10),
-      mix_cycles=mix_cycles,
+      mix_volume=round(dispense.mix.volume * 100) if dispense.mix is not None else 0,
+      mix_cycles=dispense.mix.repetitions if dispense.mix is not None else 0,
       mix_position_in_z_direction_from_liquid_surface=round(
         mix_position_in_z_direction_from_liquid_surface * 10
       ),
       surface_following_distance_during_mixing=round(surface_following_distance_during_mixing * 10),
-      mix_speed=round(mix_speed * 10),
+      mix_speed=round(dispense.mix.flow_rate * 10) if dispense.mix is not None else 10,
       limit_curve_index=limit_curve_index,
       tadm_channel_pattern=tadm_channel_pattern,
       tadm_algorithm_on_off=tadm_algorithm_on_off,

pylabrobot/liquid_handling/backends/opentrons_backend.py

@@ -420,6 +420,19 @@ async def aspirate(self, ops: List[SingleChannelAspiration], use_channels: List[
       pipette_id=pipette_id,
     )
 
+    if op.mix is not None:
+      for _ in range(op.mix.repetitions):
+        ot_api.lh.aspirate_in_place(
+          volume=op.mix.volume,
+          flow_rate=op.mix.flow_rate,
+          pipette_id=pipette_id,
+        )
+        ot_api.lh.dispense_in_place(
+          volume=op.mix.volume,
+          flow_rate=op.mix.flow_rate,
+          pipette_id=pipette_id,
+        )
+
     ot_api.lh.aspirate_in_place(
       volume=volume,
       flow_rate=flow_rate,
@@ -490,6 +503,19 @@ async def dispense(self, ops: List[SingleChannelDispense], use_channels: List[in
       pipette_id=pipette_id,
     )
 
+    if op.mix is not None:
+      for _ in range(op.mix.repetitions):
+        ot_api.lh.aspirate_in_place(
+          volume=op.mix.volume,
+          flow_rate=op.mix.flow_rate,
+          pipette_id=pipette_id,
+        )
+        ot_api.lh.dispense_in_place(
+          volume=op.mix.volume,
+          flow_rate=op.mix.flow_rate,
+          pipette_id=pipette_id,
+        )
+
     traversal_location = op.resource.get_absolute_location("c", "c", "cavity_bottom") + op.offset
     traversal_location.z = self.traversal_height
     await self.move_pipette_head(

pylabrobot/liquid_handling/backends/serializing_backend.py

@@ -108,6 +108,7 @@ async def aspirate(self, ops: List[SingleChannelAspiration], use_channels: List[
         "liquid_height": serialize(op.liquid_height),
         "blow_out_air_volume": serialize(op.blow_out_air_volume),
         "liquids": serialize(op.liquids),
+        "mix": serialize(op.mix),
       }
       for op in ops
     ]
@@ -127,6 +128,7 @@ async def dispense(self, ops: List[SingleChannelDispense], use_channels: List[in
         "liquid_height": serialize(op.liquid_height),
         "blow_out_air_volume": serialize(op.blow_out_air_volume),
         "liquids": serialize(op.liquids),
+        "mix": serialize(op.mix),
       }
       for op in ops
     ]

pylabrobot/liquid_handling/backends/serializing_backend_tests.py

@@ -105,6 +105,7 @@ async def test_aspirate(self):
             "liquid_height": None,
             "blow_out_air_volume": None,
             "liquids": [[None, 10]],
+            "mix": None,
           }
         ],
         "use_channels": [0],
@@ -133,6 +134,7 @@ async def test_dispense(self):
             "liquid_height": None,
             "blow_out_air_volume": None,
             "liquids": [[None, 10]],
+            "mix": None,
           }
         ],
         "use_channels": [0],

pylabrobot/liquid_handling/backends/tecan/EVO_tests.py

@@ -143,6 +143,7 @@ async def test_aspirate(self):
       liquid_height=10,
       blow_out_air_volume=0,
       liquids=[(None, 100)],
+      mix=None,
     )
     await self.evo.aspirate([op], use_channels=[0])
     self.evo.send_command.assert_has_calls(  # type: ignore[attr-defined]
@@ -284,6 +285,7 @@ async def test_dispense(self):
       liquid_height=10,
       blow_out_air_volume=0,
       liquids=[(None, 100)],
+      mix=None,
     )
     await self.evo.dispense([op], use_channels=[0])
     self.evo.send_command.assert_has_calls(  # type: ignore[attr-defined]