STAR pipettes: where each channel's piston stands, in uL

Nothing could read a channel's dispensing drive, so where its piston stood was unknown.
dispensing_drive_request_uL_position reads one channel's, `Px RD`, in uL;
dispensing_drives_request_uL_positions reads every channel's, by channel. The simulator answers
from where it has each piston, at rest until something moves it.

The reads record what they read on piston_positions, one entry per channel; the all-channel read
takes the channels to ask, reads them together and answers None for the rest. Initialization homes
the pistons, so it ends with the first read. The setup-sequence test no longer pins where the
autoload, its own unit, comes up among the arm's features.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: 2 people

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

@@ -347,6 +347,8 @@ def __init__(self, driver: "STARDriver", configuration: Optional[PipettesConfigu
     # One resource per channel, in channel order, when the driver was given a deck. Setup puts them
     # on the arm; the reads keep them in step. Without a deck the list stays empty.
     self.resources: List[Resource] = []
+    # Where each piston stands, in uL, by channel, as last read; empty until a piston is read.
+    self.piston_positions: List[float] = []
     self.configuration = configuration or PipettesConfiguration()
     # The height the channels travel at when a command names none, in mm. Legacy STARBackend's
     # channel traversal height.
@@ -809,6 +811,8 @@ async def initialize(
     # `z_position_at_end_of_a_command`. Read both back, or the model has them where they were.
     await self._record_where_they_stopped("y")
     await self._record_where_they_stopped("z")
+    # Initialization homes the pistons as well: the first read of where they stand comes here.
+    await self.dispensing_drives_request_uL_positions()
     return resp
 
   def _min_pair_spacing(self, i: int, j: int) -> float:
@@ -1759,6 +1763,46 @@ async def move_to_safe_z(
     async with self._temporary_z_drive_profile(speed=speed, acceleration=acceleration):
       await self.probe_z_max()
 
+  # -- dispensing drive position -------------------------------------------------------------------
+
+  async def dispensing_drive_request_uL_position(self, channel: int) -> float:
+    """Read where one channel's dispensing drive stands, in uL, and record it. `Px RD`.
+
+    Args:
+      channel: which channel, 0-indexed from the back.
+
+    Returns:
+      The piston's position in uL, 0.0 at rest; air and liquid alike. Kept on `piston_positions`.
+    """
+    self._require_channel(channel)
+    resp = await self._driver.send_command(
+      module=self.channel_id(channel), command="RD", fmt="rd#####"
+    )
+    uL = self.configuration.dispensing_drive_increments_to_uL(cast(int, resp["rd"]))
+    # The channels not read yet stand at rest, where initialization leaves every piston.
+    self.piston_positions += [0.0] * (self.num_channels - len(self.piston_positions))
+    self.piston_positions[channel] = uL
+    return uL
+
+  async def dispensing_drives_request_uL_positions(
+    self, channels: Optional[List[int]] = None
+  ) -> List[Optional[float]]:
+    """Read where the dispensing drives stand, in uL, the channels together, and record them.
+
+    Args:
+      channels: which channels, 0-indexed from the back. Every channel when None.
+
+    Returns:
+      One entry per channel of the device: the piston's position in uL, 0.0 at rest, air and
+      liquid alike, for a channel asked; None for one not asked, whose record is left alone.
+    """
+    asked = list(range(self.num_channels)) if channels is None else channels
+    read = await asyncio.gather(*(self.dispensing_drive_request_uL_position(ch) for ch in asked))
+    positions: List[Optional[float]] = [None] * self.num_channels
+    for channel, uL in zip(asked, read):
+      positions[channel] = uL
+    return positions
+
   # -- x and y together ----------------------------------------------------------------------------
 
   async def _traverse_raise_targets(self, height: float) -> Dict[int, float]:

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

@@ -566,6 +566,12 @@ def _stand_at_y(self, *ys: List[float]):
     self.pipettes.request_y_positions = reads  # type: ignore[method-assign]
     self.pipettes.move_to_y_position = self.moves  # type: ignore[method-assign]
 
+  async def test_dispensing_drive_read(self):
+    reads = unittest.mock.AsyncMock(return_value={"rd": 1067})
+    self.pipettes._driver.send_command = reads  # type: ignore[method-assign]
+    self.assertEqual(await self.pipettes.dispensing_drive_request_uL_position(2), 50.0)
+    reads.assert_awaited_once_with(module="P3", command="RD", fmt="rd#####")
+
   async def test_x_firmware(self):
     await self.pipettes._unchecked_fw_probe_x_using_clld(134.0)
     self.assertEqual(self.sent, ["C0XLxs01340"])
@@ -1070,6 +1076,22 @@ async def asyncSetUp(self):
   async def asyncTearDown(self):
     await self.driver.stop()
 
+  async def test_the_pistons_stand_where_the_simulator_has_them(self):
+    # Initialization read them once already.
+    self.assertEqual(self.pipettes.piston_positions, [0.0] * 8)
+    self.assertEqual(await self.pipettes.dispensing_drives_request_uL_positions(), [0.0] * 8)
+    self.driver.dispensing_drive_uL[0] = 50.0
+    self.driver.dispensing_drive_uL[1] = 20.0
+    self.assertEqual(await self.pipettes.dispensing_drive_request_uL_position(0), 50.0)
+    self.assertEqual(self.pipettes.piston_positions[0], 50.0)
+    # Asking for some channels answers None for the others and leaves their record alone.
+    self.pipettes.piston_positions[0] = 99.0
+    self.assertEqual(
+      await self.pipettes.dispensing_drives_request_uL_positions([1, 3]),
+      [None, 20.0, None, 0.0, None, None, None, None],
+    )
+    self.assertEqual(self.pipettes.piston_positions[:2], [99.0, 20.0])
+
   def _surface(self, well) -> float:
     """Where the tracker's water stands in `well`, in mm on the deck, by the well's own model."""
     volume = well.tracker.get_used_volume()

pylabrobot/hamilton/star/driver/master_tests.py

@@ -319,7 +319,13 @@ async def run_setup(
     )
     with recorded_moves() as moves:
       await star.setup()
-    return list(moves)
+    # The autoload is its own unit and comes up alongside the arm's features, so where its two
+    # steps fall among theirs is the event loop's business; that it homes after the device is not.
+    autoload = [move for move in moves if "autoload" in move]
+    self.assertEqual(autoload, ["II autoload", "autoload park"])
+    if "VI device" in moves:
+      self.assertGreater(moves.index("II autoload"), moves.index("VI device"))
+    return [move for move in moves if "autoload" not in move]
 
   async def test_everything_already_up(self):
     self.assertEqual(
@@ -330,8 +336,6 @@ async def test_everything_already_up(self):
         "ZA channels to safe Z",
         "iSWAP park",
         "EV 96-head probe and retract",
-        "II autoload",
-        "autoload park",
       ],
     )
 
@@ -345,8 +349,6 @@ async def test_head_down_on_a_device_that_is_up(self):
         "iSWAP park",
         "EI 96-head",
         "EV 96-head probe and retract",
-        "II autoload",
-        "autoload park",
       ],
     )
 
@@ -362,8 +364,6 @@ async def test_head_down_with_nowhere_to_eject(self):
         "ZA channels to safe Z",
         "iSWAP park",
         "EV 96-head probe and retract",
-        "II autoload",
-        "autoload park",
       ],
     )
 
@@ -424,8 +424,6 @@ async def test_device_not_up(self):
         "iSWAP park",
         "EI 96-head",
         "EV 96-head probe and retract",
-        "II autoload",
-        "autoload park",
       ],
     )
 

pylabrobot/hamilton/star/driver/simulator.py

@@ -444,6 +444,12 @@ async def answer(self, module: str, command: str, **kwargs: Any) -> Optional[Tup
         {"rz": c.z_drive_mm_to_increments(self._modelled_z(channel))},
         f"where the model has channel {channel}'s stop disc",
       )
+    if command == "RD":
+      uL = self.device.dispensing_drive_uL.get(channel, 0.0)
+      return (
+        {"rd": c.dispensing_drive_uL_to_increments(uL)},
+        f"where channel {channel}'s piston stands",
+      )
     if command in ("ZL", "ZE"):
       return self._answer_liquid_search(channel, command, **kwargs)
     if command == "ZH":
@@ -1407,6 +1413,8 @@ def __init__(
     self.liquid_searches: Dict[int, Container] = {}
     # What each channel last detected liquid at, in mm on the deck; 0.0 until a search finds any.
     self.last_lld_heights: Dict[int, float] = {}
+    # Where each channel's piston stands, in uL.
+    self.dispensing_drive_uL: Dict[int, float] = {}
 
     # What each module says when asked whether it is initialized, and where things are.
     self.initialized = {module: initialized for module in ("C0", "I0", "R0", "H0")}