v1 STAR pipettes: old master is sent limited PIP parameters

pylabrobot/hamilton/star/driver/configuration.py

@@ -12,6 +12,8 @@
 from pylabrobot.hamilton.star.driver.features.pipettes import PipettesConfiguration
 from pylabrobot.hamilton.star.driver.features.x_arm import XArmConfiguration
 
+EXTENDED_PIP_PARAMETERS_SINCE = datetime.date(2011, 5, 12)
+
 
 @dataclass
 class DeviceConfiguration:
@@ -138,6 +140,11 @@ class DeviceConfiguration:
   right_arm_min_y_position: float = 6.0
   """Right arm minimal Y position [mm] (yx). Default: 6.0."""
 
+  @property
+  def takes_extended_pip_parameters(self) -> bool:
+    """Whether the master takes `ti`, `td`, TADM, second-section aspiration and `po`."""
+    return self.firmware_date is None or self.firmware_date >= EXTENDED_PIP_PARAMETERS_SINCE
+
 
 def _restore(hint: Any, value: Any) -> Any:
   """One value, back in the type its field is declared to hold.

pylabrobot/hamilton/star/driver/features/pipettes.py

@@ -2184,6 +2184,10 @@ def default_initialize_y_positions(self) -> List[float]:
     spacing = round((back - front) * 10) // (self.num_channels - 1)
     return [(round(back * 10) - channel * spacing) / 10 for channel in range(self.num_channels)]
 
+  def _extended(self, **parameters: Any) -> Dict[str, Any]:
+    device = self._driver.configuration
+    return parameters if device is None or device.takes_extended_pip_parameters else {}
+
   async def initialize(
     self,
     x_position: Optional[float] = None,
@@ -2253,7 +2257,7 @@ async def initialize(
       te=f"{round(minimum_traverse_height_end * 10):04}",
       tm=[f"{tm:01}" for tm in tip_pattern],
       tt=f"{tip_type:02}",
-      ti=discarding_method,
+      **self._extended(ti=discarding_method),
     )
     # Everything the channels carried is in the waste now, and belongs nowhere.
     for channel, involved in enumerate(tip_pattern):
@@ -4058,7 +4062,7 @@ async def _unchecked_fw_pick_up_tips(
       tp=f"{begin_tip_pick_up_process:04}",
       tz=f"{end_tip_pick_up_process:04}",
       th=f"{minimum_traverse_height_start:04}",
-      td=pickup_method.value,
+      **self._extended(td=pickup_method.value),
     )
 
   def _tip_traverse_height(
@@ -4353,7 +4357,7 @@ async def _unchecked_fw_drop_tips(
       tz=end_tip_deposit_process,
       th=minimum_traverse_height_start,
       te=minimum_traverse_height_end,
-      ti=discarding_method.value,
+      **self._extended(ti=discarding_method.value),
     )
 
   async def _drop_tips_in_one_move(
@@ -5489,7 +5493,7 @@ async def _unchecked_fw_aspirate(
       lp=[f"{lp:04}" for lp in lld_search_height],
       ch=[f"{ch:03}" for ch in clot_detection_height],
       zl=[f"{zl:04}" for zl in liquid_surface_no_lld],
-      po=[f"{po:04}" for po in pull_out_distance_transport_air],
+      **self._extended(po=[f"{po:04}" for po in pull_out_distance_transport_air]),
       zu=[f"{zu:04}" for zu in second_section_height],
       zr=[f"{zr:05}" for zr in second_section_ratio],
       zx=[f"{zx:04}" for zx in minimum_height],
@@ -5513,14 +5517,18 @@ async def _unchecked_fw_aspirate(
       mp=[f"{mp:03}" for mp in mix_position_from_liquid_surface],
       ms=[f"{ms:04}" for ms in mix_speed],
       mh=[f"{mh:04}" for mh in mix_surface_following_distance],
-      gi=[f"{gi:03}" for gi in limit_curve_index],
-      gj=tadm_algorithm,
-      gk=recording_mode,
-      lk=[1 if lk else 0 for lk in use_2nd_section_aspiration],
-      ik=[f"{ik:04}" for ik in retract_height_over_2nd_section_to_empty_tip],
+      **self._extended(
+        gi=[f"{gi:03}" for gi in limit_curve_index],
+        gj=tadm_algorithm,
+        gk=recording_mode,
+        lk=[1 if lk else 0 for lk in use_2nd_section_aspiration],
+        ik=[f"{ik:04}" for ik in retract_height_over_2nd_section_to_empty_tip],
+      ),
       sd=[f"{sd:04}" for sd in dispensation_speed_during_emptying_tip],
-      se=[f"{se:04}" for se in dosing_drive_speed_during_2nd_section_search],
-      sz=[f"{sz:04}" for sz in z_drive_speed_during_2nd_section_search],
+      **self._extended(
+        se=[f"{se:04}" for se in dosing_drive_speed_during_2nd_section_search],
+        sz=[f"{sz:04}" for sz in z_drive_speed_during_2nd_section_search],
+      ),
       io=[f"{io:04}" for io in cup_upper_edge],
     )
 
@@ -6462,7 +6470,7 @@ async def _unchecked_fw_dispense(
       zx=[f"{zx:04}" for zx in minimum_height],
       lp=[f"{lp:04}" for lp in lld_search_height],
       zl=[f"{zl:04}" for zl in liquid_surface_no_lld],
-      po=[f"{po:04}" for po in pull_out_distance_transport_air],
+      **self._extended(po=[f"{po:04}" for po in pull_out_distance_transport_air]),
       ip=[f"{ip:04}" for ip in immersion_depth],
       it=[f"{it}" for it in immersion_depth_direction],
       fp=[f"{fp:04}" for fp in surface_following_distance],
@@ -6488,9 +6496,11 @@ async def _unchecked_fw_dispense(
       mp=[f"{mp:03}" for mp in mix_position_from_liquid_surface],
       ms=[f"{ms:04}" for ms in mix_speed],
       mh=[f"{mh:04}" for mh in mix_surface_following_distance],
-      gi=[f"{gi:03}" for gi in limit_curve_index],
-      gj=tadm_algorithm,
-      gk=recording_mode,
+      **self._extended(
+        gi=[f"{gi:03}" for gi in limit_curve_index],
+        gj=tadm_algorithm,
+        gk=recording_mode,
+      ),
     )
 
   async def _dispense_in_one_move(

pylabrobot/hamilton/star/driver/features/pipettes_tests.py

@@ -1,4 +1,5 @@
 import asyncio
+import datetime
 import math
 import re
 import unittest
@@ -7,6 +8,7 @@
 
 from pylabrobot.hamilton.protocol.text.framing import assemble_command
 from pylabrobot.hamilton.star.device import RECORDING_STAR
+from pylabrobot.hamilton.star.driver.configuration import EXTENDED_PIP_PARAMETERS_SINCE
 from pylabrobot.hamilton.star.driver.errors import STARFirmwareError, check_fw_string_error
 from pylabrobot.hamilton.star.driver.features.pipettes import (
   Pipettes,
@@ -4155,6 +4157,41 @@ async def test_the_raw_dispense_is_legacy_dispense_pip_to_the_character(self):
     )
 
 
+class TestAMasterOlderThanTheExtendedPipParameters(unittest.IsolatedAsyncioTestCase):
+  """A master built before them is sent none of them, on the commands that carry them."""
+
+  async def asyncSetUp(self):
+    self.pipettes, self.rack, _ = await channels_over_a_rack()
+    device = cast(Any, self.pipettes._driver.configuration)
+    device.firmware_date = EXTENDED_PIP_PARAMETERS_SINCE - datetime.timedelta(days=1)
+    self.send = unittest.mock.AsyncMock(wraps=self.pipettes._driver.send_command)
+    self.pipettes._driver.send_command = self.send  # type: ignore[method-assign]
+
+  def sent(self, command: str) -> Mapping[str, Any]:
+    return next(c.kwargs for c in self.send.call_args_list if c.kwargs.get("command") == command)
+
+  async def test_the_tip_commands(self):
+    await self.pipettes.initialize()
+    await self.pipettes.pick_up_tips([self.rack.get_item("A1")])
+    await self.pipettes.drop_tips([self.rack.get_item("A1")])
+    self.assertNotIn("ti", self.sent("DI"))
+    self.assertNotIn("td", self.sent("TP"))
+    self.assertNotIn("ti", self.sent("TR"))
+
+  async def test_aspirate_and_dispense(self):
+    self.pipettes.get_mounted_tip = unittest.mock.Mock(  # type: ignore[method-assign]
+      return_value=self.rack.get_item("A1").tip
+    )
+    self.pipettes._record_after_command = unittest.mock.AsyncMock()  # type: ignore[method-assign]
+    places = [Coordinate(300.0, 300.0, 150.0)]
+    await self.pipettes._aspirate_in_one_move([0], places, [180.0], [120.0], [100.0])
+    await self.pipettes._dispense_in_one_move([0], places, [180.0], [120.0], [100.0])
+    for parameter in ("gi", "gj", "gk", "lk", "ik", "se", "sz", "po"):
+      self.assertNotIn(parameter, self.sent("AS"))
+    for parameter in ("gi", "gj", "gk", "po"):
+      self.assertNotIn(parameter, self.sent("DS"))
+
+
 class TestDispenseInOneMove(unittest.IsolatedAsyncioTestCase):
   """One `C0 DS` from mm and uL: the fields it fills, the defaults it takes, what it refuses."""