STARBackend: resolve installed X-arms into XArmInformation at setup

pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py

@@ -1233,9 +1233,15 @@ def _dispensing_mode_for_op(empty: bool, jet: bool, blow_out: bool) -> int:
 
 @dataclass
 class DriveConfiguration:
-  """Configuration for an X drive (left or right).
+  """Configuration and geometry for an X drive (left or right).
+
+  The installed-module bits combine byte 1 (xl/xr) and byte 2 (xn/xo). The arm
+  geometry - width, travel range, workspace range - comes from the X-drive range (RU)
+  and working-envelope (UA) queries, so it is None on a drive built from the module
+  bits alone (e.g. a simulated configuration) and populated when
+  `request_extended_configuration` builds the drive. `model` and `reference_point`
+  follow from `width`.
 
-  Combines byte 1 (xl/xr) and byte 2 (xn/xo) into a single object.
   Note: the installed modules on left and right drives must be different.
   """
 
@@ -1249,6 +1255,28 @@ class DriveConfiguration:
   imaging_channel_installed: bool = False
   robotic_channel_installed: bool = False
 
+  width: Optional[float] = None
+  """Arm width (mm), from the machine configuration."""
+  x_range: Optional[Tuple[float, float]] = None
+  """Drive travel `(min, max)` in mm."""
+  workspace_range: Optional[Tuple[float, float]] = None
+  """Reachable X workspace `(min, max)` in mm."""
+
+  @property
+  def model(self) -> str:
+    """Arm variant derived from `width`: wide arms span both rails, narrow arms one."""
+    assert self.width is not None, "arm geometry not resolved"
+    if self.width > 300:
+      return "hamilton_legacy_star_dual_rail_arm"
+    return "hamilton_legacy_star_single_right_rail_arm"
+
+  @property
+  def reference_point(self) -> Literal["center", "right"]:
+    """Where along the arm's width the tracked X refers to: the arm center for a
+    dual-rail arm, the right edge for a single-rail arm."""
+    assert self.width is not None, "arm geometry not resolved"
+    return "center" if self.width > 300 else "right"
+
 
 @dataclass
 class MachineConfiguration:
@@ -1344,8 +1372,8 @@ class ExtendedConfiguration:
   """Tip waste X-position [mm] (xw). Default: 1340.0."""
   left_x_drive: DriveConfiguration = field(default_factory=DriveConfiguration)
   """Left X drive configuration (xl + xn)."""
-  right_x_drive: DriveConfiguration = field(default_factory=DriveConfiguration)
-  """Right X drive configuration (xr + xo)."""
+  right_x_drive: Optional[DriveConfiguration] = None
+  """Right X drive configuration (xr + xo), or None when no right arm is installed."""
   min_iswap_collision_free_position: float = 350.0
   """Minimal iSWAP collision free position for direct X access [mm] (xm). Default: 350.0."""
   max_iswap_collision_free_position: float = 1140.0
@@ -6032,10 +6060,13 @@ async def request_machine_configuration(self) -> MachineConfiguration:
     )
 
   async def request_extended_configuration(self) -> ExtendedConfiguration:
-    """Request extended configuration (QM command).
+    """Request extended configuration (QM command) with X-arm geometry resolved.
 
     Returns the full instrument configuration matching the AK
-    (Set Instrument Configuration) [SFCO.0026] parameter set.
+    (Set Instrument Configuration) [SFCO.0026] parameter set. Each installed X-drive's
+    geometry (width, travel range, workspace range) is resolved from the X-drive range
+    (RU) and working-envelope (UA) queries; `right_x_drive` is None when no second arm
+    is installed.
     """
 
     resp = await self.send_command(
@@ -6045,7 +6076,15 @@ async def request_extended_configuration(self) -> ExtendedConfiguration:
       + "ys###kl###km###ym####yu####yx####",
     )
 
-    def _parse_drive(byte1: int, byte2: int) -> DriveConfiguration:
+    ranges = await self.request_maximal_ranges_of_x_drives()
+    wraps = await self.request_working_envelopes_per_arm()
+
+    def _build_drive(
+      byte1: int, byte2: int, side: Literal["left", "right"], width: float
+    ) -> Optional[DriveConfiguration]:
+      wrap, workspace_range = wraps[side]
+      if wrap == 0:  # arm not installed
+        return None
       return DriveConfiguration(
         pip_installed=bool(byte1 & (1 << 0)),
         iswap_installed=bool(byte1 & (1 << 1)),
@@ -6056,8 +6095,14 @@ def _parse_drive(byte1: int, byte2: int) -> DriveConfiguration:
         tube_gripper_installed=bool(byte1 & (1 << 6)),
         imaging_channel_installed=bool(byte1 & (1 << 7)),
         robotic_channel_installed=bool(byte2 & (1 << 0)),
+        width=width,
+        x_range=ranges[side],
+        workspace_range=workspace_range,
       )
 
+    left_x_drive = _build_drive(resp["xl"], resp["xn"], "left", resp["xu"] / 10)
+    assert left_x_drive is not None, "STAR must have a left X-arm"
+
     ka = resp["ka"]
     return ExtendedConfiguration(
       left_x_drive_large=bool(ka & (1 << 0)),
@@ -6087,8 +6132,8 @@ def _parse_drive(byte1: int, byte2: int) -> DriveConfiguration:
       instrument_size_slots=resp["xt"],
       auto_load_size_slots=resp["xa"],
       tip_waste_x_position=resp["xw"] / 10,
-      left_x_drive=_parse_drive(resp["xl"], resp["xn"]),
-      right_x_drive=_parse_drive(resp["xr"], resp["xo"]),
+      left_x_drive=left_x_drive,
+      right_x_drive=_build_drive(resp["xr"], resp["xo"], "right", resp["xv"] / 10),
       min_iswap_collision_free_position=resp["xm"] / 10,
       max_iswap_collision_free_position=resp["xx"] / 10,
       left_x_arm_width=resp["xu"] / 10,
@@ -6268,15 +6313,34 @@ async def request_right_x_arm_position(self) -> float:
     resp_dmm = await self.send_command(module="C0", command="QX", fmt="rx#####")
     return cast(float, resp_dmm["rx"]) / 10
 
-  async def request_maximal_ranges_of_x_drives(self):
-    """Request maximal ranges of X drives"""
+  async def request_maximal_ranges_of_x_drives(self) -> Dict[str, Tuple[float, float]]:
+    """Request the maximal travel range of each X drive.
 
-    return await self.send_command(module="C0", command="RU")
+    Returns:
+      The `(minimum, maximum)` X position in mm each drive can reach, keyed by side:
+      `{"left": (min, max), "right": (min, max)}`.
+    """
+    resp = await self.send_command(module="C0", command="RU")
+    values = [int(v) / 10 for v in resp.split("ru")[-1].strip().split()]
+    left_min, left_max, right_min, right_max = values
+    return {"left": (left_min, left_max), "right": (right_min, right_max)}
 
-  async def request_present_wrap_size_of_installed_arms(self):
-    """Request present wrap size of installed arms"""
+  async def request_working_envelopes_per_arm(
+    self,
+  ) -> Dict[str, Tuple[float, Tuple[float, float]]]:
+    """Request the working envelope of each installed arm.
 
-    return await self.send_command(module="C0", command="UA")
+    Returns:
+      Per side, `(wrap_size, (workspace_min, workspace_max))` in mm, keyed by side. A
+      `wrap_size` of 0 means that arm is not installed.
+    """
+    resp = await self.send_command(module="C0", command="UA")
+    values = [int(v) / 10 for v in resp.split("ua")[-1].strip().split()]
+    left_wrap, right_wrap, left_min, left_max, right_min, right_max = values
+    return {
+      "left": (left_wrap, (left_min, left_max)),
+      "right": (right_wrap, (right_min, right_max)),
+    }
 
   async def request_left_x_arm_last_collision_type(self):
     """Request left X-Arm last collision type (after error 27)

pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py

@@ -3,7 +3,8 @@
 import logging
 import warnings
 from contextlib import asynccontextmanager
-from typing import Dict, List, Literal, Optional, Union
+from dataclasses import replace
+from typing import Dict, List, Literal, Optional, Tuple, Union
 
 from pylabrobot.io.validation_utils import LOG_LEVEL_IO
 from pylabrobot.liquid_handling.backends import LiquidHandlerBackend
@@ -16,6 +17,7 @@
   iSWAPInformation,
 )
 from pylabrobot.resources.container import Container
+from pylabrobot.resources.hamilton.hamilton_decks import HamiltonDeck
 from pylabrobot.resources.tip_tracker import does_tip_tracking
 from pylabrobot.resources.well import Well
 
@@ -39,6 +41,11 @@
   max_iswap_collision_free_position=600.0,
 )
 
+# Minimal left-drive X position of a dual-rail arm. Validated against real hardware;
+# the single-rail minimum is not yet known (#822). Only the chatterbox needs this
+# literal - a physical STAR reports its own value from the drive-range query.
+_DUAL_RAIL_LEFT_X_MIN = 95.0
+
 # Hamilton factory defaults. Per-machine EEPROM calibration will differ
 # slightly (e.g., L1=137.8, L2=137.7, STRAIGHT=-45.01 on one tested machine);
 # these defaults are accurate enough for simulation but not for
@@ -223,8 +230,58 @@ async def request_machine_configuration(self) -> MachineConfiguration:
     return self._machine_configuration
 
   async def request_extended_configuration(self) -> ExtendedConfiguration:
-    assert self._extended_conf is not None
-    return self._extended_conf
+    """Return the configured extended configuration with X-arm geometry resolved.
+
+    Mirrors STARBackend.request_extended_configuration: fills each installed drive's
+    geometry from the mocked X-drive range/envelope replies. A right drive that was not
+    configured (None) stays None.
+    """
+    conf = self._extended_conf
+    assert conf is not None
+    ranges = await self.request_maximal_ranges_of_x_drives()
+    wraps = await self.request_working_envelopes_per_arm()
+
+    def _with_geometry(
+      drive: Optional[DriveConfiguration], side: str, width: float
+    ) -> Optional[DriveConfiguration]:
+      if drive is None:
+        return None
+      wrap, workspace_range = wraps[side]
+      if wrap == 0:  # arm not installed
+        return None
+      return replace(drive, width=width, x_range=ranges[side], workspace_range=workspace_range)
+
+    left_x_drive = _with_geometry(conf.left_x_drive, "left", conf.left_x_arm_width)
+    assert left_x_drive is not None, "STAR must have a left X-arm"
+
+    return replace(
+      conf,
+      left_x_drive=left_x_drive,
+      right_x_drive=_with_geometry(conf.right_x_drive, "right", conf.right_x_arm_width),
+    )
+
+  def _simulated_x_reach_max(self) -> float:
+    """Rightmost reachable X (mm) in simulation, from the deck's reachable range."""
+    deck = self._deck
+    if isinstance(deck, HamiltonDeck):
+      return deck.rails_to_location(deck.num_rails).x
+    if deck is not None:
+      return deck.get_size_x()
+    return 1338.0  # nominal STAR reach
+
+  async def request_maximal_ranges_of_x_drives(self) -> Dict[str, Tuple[float, float]]:
+    x_range = (_DUAL_RAIL_LEFT_X_MIN, self._simulated_x_reach_max())
+    return {"left": x_range, "right": x_range}
+
+  async def request_working_envelopes_per_arm(
+    self,
+  ) -> Dict[str, Tuple[float, Tuple[float, float]]]:
+    workspace = (-323.2, self._simulated_x_reach_max())
+    left = (595.2, workspace)  # wrap, workspace
+    # A wrap of 0 signals "arm not installed" (per the base method's contract).
+    right_installed = self.extended_conf.right_x_drive is not None
+    right = (595.2, workspace) if right_installed else (0.0, (0.0, 0.0))
+    return {"left": left, "right": right}
 
   # # # # # # # # 1_000 uL Channel: Basic Commands # # # # # # # #
 

pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py

@@ -36,6 +36,7 @@
 
 from .STAR_backend import (
   CommandSyntaxError,
+  DriveConfiguration,
   HamiltonNoTipError,
   HardwareError,
   Head96Information,
@@ -2583,3 +2584,51 @@ async def test_duplicate_channels_serialize_measurements(self):
     self.assertEqual(len(result), 2)
     self.assertAlmostEqual(result[0], 0 - well_a.get_absolute_location("c", "c", "cavity_bottom").z)
     self.assertAlmostEqual(result[1], 0 - well_b.get_absolute_location("c", "c", "cavity_bottom").z)
+
+
+class TestXArmGeometry(unittest.IsolatedAsyncioTestCase):
+  """setup() resolves QM/RU/UA firmware onto the X-drive DriveConfigurations."""
+
+  async def asyncSetUp(self):
+    self.lh = LiquidHandler(STARChatterboxBackend(), deck=STARLetDeck())
+    await self.lh.setup()
+
+  async def test_single_left_arm(self):
+    self.assertIsNotNone(self.lh.backend.extended_conf.left_x_drive.workspace_range)
+    self.assertIsNone(self.lh.backend.extended_conf.right_x_drive)
+
+  async def test_left_arm_fields(self):
+    left = self.lh.backend.extended_conf.left_x_drive
+    self.assertEqual(left.reference_point, "center")  # QM width -> center rail
+    # x_range comes from RU, workspace_range from UA: request_extended_configuration routes
+    # the two firmware sources into distinct slots (not covered by the parse/branch tests).
+    assert left.x_range is not None and left.workspace_range is not None
+    self.assertEqual(left.x_range[0], 95.0)
+    self.assertEqual(left.workspace_range[0], -323.2)
+
+  def test_model_and_reference_by_width(self):
+    dual = DriveConfiguration(width=370.0)
+    self.assertEqual(dual.model, "hamilton_legacy_star_dual_rail_arm")
+    self.assertEqual(dual.reference_point, "center")
+    # 246.0 is an illustrative <300 width; no single-rail dump exists yet to measure one.
+    single = DriveConfiguration(width=246.0)
+    self.assertEqual(single.model, "hamilton_legacy_star_single_right_rail_arm")
+    self.assertEqual(single.reference_point, "right")
+
+
+class TestXArmRangeQueries(unittest.IsolatedAsyncioTestCase):
+  """RU/UA parse the firmware replies observed on real machines."""
+
+  def setUp(self):
+    self.star = STARBackend()
+    self.star.send_command = unittest.mock.AsyncMock()
+
+  async def test_maximal_ranges_of_x_drives(self):
+    self.star.send_command.return_value = "C0RUid0002er00/00ru00950 13402 30000 30000"
+    ranges = await self.star.request_maximal_ranges_of_x_drives()
+    self.assertEqual(ranges, {"left": (95.0, 1340.2), "right": (3000.0, 3000.0)})
+
+  async def test_working_envelopes_per_arm(self):
+    self.star.send_command.return_value = "C0UAid0001er00/00ua5952 0000 -03232 +15172 +30000 +30000"
+    wraps = await self.star.request_working_envelopes_per_arm()
+    self.assertEqual(wraps, {"left": (595.2, (-323.2, 1517.2)), "right": (0.0, (3000.0, 3000.0))})