From a5590811b9bf32e36c233bdfe213b0b50ed051d2 Mon Sep 17 00:00:00 2001 From: Camillo Moschner Date: Fri, 2 Oct 2026 13:47:32 +0100 Subject: [PATCH] STAR pipettes: a master older than them is sent no extended PIP parameters A master built before the 2011-05-12 revision of its command set answers "Unknown parameter" to the channel commands as this driver sends them. DeviceConfiguration says from the master's build date whether it takes the extended parameters, and the commands leave them out when it does not: `ti` in C0 DI and C0 TR, `td` in C0 TP, `po`, `gi gj gk`, `lk ik` and `se sz` in C0 AS, `po` and `gi gj gk` in C0 DS. A master built later, or one with no build date, is sent exactly what it was before. Co-Authored-By: Claude Opus 5.5 --- .../hamilton/star/driver/configuration.py | 7 ++++ .../hamilton/star/driver/features/pipettes.py | 40 ++++++++++++------- .../star/driver/features/pipettes_tests.py | 37 +++++++++++++++++ 3 files changed, 69 insertions(+), 15 deletions(-) diff --git a/pylabrobot/hamilton/star/driver/configuration.py b/pylabrobot/hamilton/star/driver/configuration.py index 3205a035e81..4dc1ff55340 100644 --- a/pylabrobot/hamilton/star/driver/configuration.py +++ b/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. diff --git a/pylabrobot/hamilton/star/driver/features/pipettes.py b/pylabrobot/hamilton/star/driver/features/pipettes.py index 931221ef832..46cb141d80d 100644 --- a/pylabrobot/hamilton/star/driver/features/pipettes.py +++ b/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( diff --git a/pylabrobot/hamilton/star/driver/features/pipettes_tests.py b/pylabrobot/hamilton/star/driver/features/pipettes_tests.py index 5f6b854bf1d..7aa2f4b14d0 100644 --- a/pylabrobot/hamilton/star/driver/features/pipettes_tests.py +++ b/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."""