Implement flow rate for Tecan EVO backend (#549)

Co-authored-by: nedru004 <david.nedrud@gmail.com>

Author: rickwierenga

pylabrobot/liquid_handling/backends/tecan/EVO_backend.py

@@ -464,7 +464,7 @@ async def dispense(self, ops: List[SingleChannelDispense], use_channels: List[in
     x, _ = self._first_valid(x_positions)
     y, yi = self._first_valid(y_positions)
     assert x is not None and y is not None
-    await self.liha.set_z_travel_height(z if z else self._z_range for z in z_positions["travel"])
+    await self.liha.set_z_travel_height([z if z else self._z_range for z in z_positions["travel"]])
     await self.liha.position_absolute_all_axis(
       x,
       y - yi * ys,
@@ -732,11 +732,13 @@ def _aspirate_airgap(
       assert tlc is not None
       pvl[channel] = 0
       if airgap == "lag":
-        sep[channel] = int(tlc.aspirate_lag_speed * 12)  # 6? TODO: verify step unit
-        ppr[channel] = int(tlc.aspirate_lag_volume * 6)  # 3?
+        sep[channel] = int(
+          tlc.aspirate_lag_speed * 6
+        )  # 6? TODO: verify step unit (half step per second)
+        ppr[channel] = int(tlc.aspirate_lag_volume * 3)  # 3? (Relative position in full steps)
       elif airgap == "tag":
-        sep[channel] = int(tlc.aspirate_tag_speed * 12)  # 6?
-        ppr[channel] = int(tlc.aspirate_tag_volume * 6)  # 3?
+        sep[channel] = int(tlc.aspirate_tag_speed * 6)  # 6?
+        ppr[channel] = int(tlc.aspirate_tag_volume * 3)  # 3?
 
     return pvl, sep, ppr
 
@@ -802,10 +804,11 @@ def _aspirate_action(
       tlc = tecan_liquid_classes[i]
       z = zadd[channel]
       assert tlc is not None and z is not None
-      sep[channel] = int(tlc.aspirate_speed * 12)  # 6?
-      ssz[channel] = round(z * tlc.aspirate_speed / ops[i].volume)
+      flow_rate = ops[i].flow_rate or tlc.aspirate_speed
+      sep[channel] = int(flow_rate * 6)  # 6?
+      ssz[channel] = round(z * flow_rate / ops[i].volume)
       volume = tlc.compute_corrected_volume(ops[i].volume)
-      mtr[channel] = round(volume * 6)  # 3?
+      mtr[channel] = round(volume * 3)  # 3?  # Relative position in full steps
       ssz_r[channel] = int(tlc.aspirate_retract_speed * 10)
 
     return ssz, sep, stz, mtr, ssz_r
@@ -838,15 +841,16 @@ def _dispense_action(
     for i, channel in enumerate(use_channels):
       tlc = tecan_liquid_classes[i]
       assert tlc is not None
-      sep[channel] = int(tlc.dispense_speed * 12)  # 6?
-      spp[channel] = int(tlc.dispense_breakoff * 12)  # 6?
+      flow_rate = ops[i].flow_rate or tlc.dispense_speed
+      sep[channel] = int(flow_rate * 6)  # 6?
+      spp[channel] = int(tlc.dispense_breakoff * 6)  # 6? half step per second
       stz[channel] = 0
       volume = (
         tlc.compute_corrected_volume(ops[i].volume)
         + tlc.aspirate_lag_volume
         + tlc.aspirate_tag_volume
       )
-      mtr[channel] = -round(volume * 6)  # 3?
+      mtr[channel] = -round(volume * 3)  # 3?
 
     return sep, spp, stz, mtr
 

pylabrobot/liquid_handling/backends/tecan/EVO_tests.py

@@ -178,12 +178,12 @@ async def test_aspirate(self):
         call(
           module="C5",
           command="SEP",
-          params=[840, None, None, None, None, None, None, None],
+          params=[420, None, None, None, None, None, None, None],
         ),
         call(
           module="C5",
           command="PPR",
-          params=[30, None, None, None, None, None, None, None],
+          params=[15, None, None, None, None, None, None, None],
         ),
         call(module="C5", command="SDM", params=[7, 1]),
         call(
@@ -234,7 +234,7 @@ async def test_aspirate(self):
         call(
           module="C5",
           command="SEP",
-          params=[1200, None, None, None, None, None, None, None],
+          params=[600, None, None, None, None, None, None, None],
         ),
         call(
           module="C5",
@@ -244,7 +244,7 @@ async def test_aspirate(self):
         call(
           module="C5",
           command="MTR",
-          params=[626, None, None, None, None, None, None, None],
+          params=[313, None, None, None, None, None, None, None],
         ),
         call(
           module="C5",
@@ -264,12 +264,12 @@ async def test_aspirate(self):
         call(
           module="C5",
           command="SEP",
-          params=[840, None, None, None, None, None, None, None],
+          params=[420, None, None, None, None, None, None, None],
         ),
         call(
           module="C5",
           command="PPR",
-          params=[60, None, None, None, None, None, None, None],
+          params=[30, None, None, None, None, None, None, None],
         ),
       ]
     )
@@ -309,12 +309,12 @@ async def test_dispense(self):
         call(
           module="C5",
           command="SEP",
-          params=[7200, None, None, None, None, None, None, None],
+          params=[600, None, None, None, None, None, None, None],
         ),
         call(
           module="C5",
           command="SPP",
-          params=[4800, None, None, None, None, None, None, None],
+          params=[2400, None, None, None, None, None, None, None],
         ),
         call(
           module="C5",
@@ -324,11 +324,52 @@ async def test_dispense(self):
         call(
           module="C5",
           command="MTR",
-          params=[-716, None, None, None, None, None, None, None],
+          params=[-358, None, None, None, None, None, None, None],
         ),
       ]
     )
 
+  # async def test_aspirate_custom_flow_rate(self):
+  #   op = SingleChannelAspiration(
+  #     resource=self.plate.get_item("A1"),
+  #     offset=Coordinate.zero(),
+  #     tip=self.tr.get_tip("A1"),
+  #     volume=100,
+  #     flow_rate=200,
+  #     liquid_height=10,
+  #     blow_out_air_volume=0,
+  #     liquids=[(None, 100)],
+  #   )
+  #   await self.evo.aspirate([op], use_channels=[0])
+  #   self.evo.send_command.assert_any_call(  # type: ignore[attr-defined]
+  #     module="C5",
+  #     command="SSZ",
+  #     params=[60, None, None, None, None, None, None, None],
+  #   )
+  #   self.evo.send_command.assert_any_call(  # type: ignore[attr-defined]
+  #     module="C5",
+  #     command="SEP",
+  #     params=[2400, None, None, None, None, None, None, None],
+  #   )
+
+  # async def test_dispense_custom_flow_rate(self):
+  #   op = SingleChannelDispense(
+  #     resource=self.plate.get_item("A1"),
+  #     offset=Coordinate.zero(),
+  #     tip=self.tr.get_tip("A1"),
+  #     volume=100,
+  #     flow_rate=200,
+  #     liquid_height=10,
+  #     blow_out_air_volume=0,
+  #     liquids=[(None, 100)],
+  #   )
+  #   await self.evo.dispense([op], use_channels=[0])
+  #   self.evo.send_command.assert_any_call(  # type: ignore[attr-defined]
+  #     module="C5",
+  #     command="SEP",
+  #     params=[2400, None, None, None, None, None, None, None],
+  #   )
+
   async def test_move_resource(self):
     pickup = ResourcePickup(
       resource=self.plate,