STARBackend: bound X-arm and channel moves to firmware-resolved x_range

pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py

@@ -1743,6 +1743,9 @@ def _ops_to_fw_positions(
     self, ops: Sequence[PipettingOp], use_channels: List[int]
   ) -> Tuple[List[int], List[int], List[bool]]:
     x_positions, y_positions, channels_involved = super()._ops_to_fw_positions(ops, use_channels)
+    # TODO: also bound each involved channel's x against the arm's resolved x_range here,
+    # raising once with every non-compliant (channel, x) pair. Extends the primitive guard
+    # in `_check_x_arm_reachable` to the high-level pipetting ops that route through here.
     if self.left_side_panel_installed:
       min_x = round(self.PIP_X_MIN_WITH_LEFT_SIDE_PANEL * 10)
       for x, involved in zip(x_positions, channels_involved):
@@ -2264,13 +2267,13 @@ async def experimental_x_arm_move(
     """Move the X-arm to an absolute X position with specified acceleration.
 
     Args:
-      x: Target X coordinate in mm. Must be between 90.0 and 1350.0.
+      x: Target X coordinate in mm. Must be within the arm's reachable X range
+        (see the X-drive's `x_range`).
       acceleration_level: Acceleration index (hardware units), 1-5. Default 3.
       current_protection_limiter: Motor current limit (hardware units), 0-7. Default 7.
     """
 
-    if not (90.0 <= x <= 1350.0):
-      raise ValueError(f"x must be between 90.0 and 1350.0 mm, is {x}")
+    self._check_x_arm_reachable(x)
     if not (1 <= acceleration_level <= 5):
       raise ValueError(f"acceleration_level must be between 1 and 5, is {acceleration_level}")
     if not (0 <= current_protection_limiter <= 7):
@@ -2286,6 +2289,35 @@ async def experimental_x_arm_move(
       lw=str(current_protection_limiter),
     )
 
+  def _check_x_arm_reachable(self, x: float, position: Literal["left", "right"] = "left") -> None:
+    """Raise if x (mm) is outside the X-drive's travel range (``x_range``).
+
+    ``x`` is the arm's position at its reference point (its center for a dual-rail arm, the
+    right edge for a single-rail arm), so the bound is that point's travel range ``x_range``.
+
+    Args:
+      x: Target X coordinate in mm.
+      position: Which X-arm the move drives. Defaults to the left (main) arm.
+
+    Reading ``extended_conf`` already raises if setup() has not run.
+
+    Raises:
+      RuntimeError: If the installed arm's geometry was not resolved.
+      ValueError: If no such arm is installed, or x is outside its ``x_range``.
+    """
+    arm = (
+      self.extended_conf.left_x_drive if position == "left" else self.extended_conf.right_x_drive
+    )
+    if arm is None:
+      raise ValueError(f"No {position} X-arm is installed.")
+    if arm.x_range is None:
+      raise RuntimeError(f"{position} X-arm geometry not resolved")
+    x_min, x_max = arm.x_range
+    if not x_min <= x <= x_max:
+      drives = (self.extended_conf.left_x_drive, self.extended_conf.right_x_drive)
+      label = f"{position} X-arm" if sum(d is not None for d in drives) > 1 else "X-arm"
+      raise ValueError(f"{label} x={x}mm is outside its drive travel range [{x_min}, {x_max}].")
+
   # # # # Single-Channel Pipette Commands # # # #
 
   # # # Machine Query (MEM-READ) Commands: Single-Channel # # #
@@ -5131,6 +5163,7 @@ async def move_channel_x(self, channel: int, x: float):
         f"PIP channel x={x}mm is below the minimum {self.PIP_X_MIN_WITH_LEFT_SIDE_PANEL}mm "
         f"(left side panel is installed)"
       )
+    self._check_x_arm_reachable(x)
     await self.position_left_x_arm_(round(x * 10))
 
   @need_iswap_parked

pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py

@@ -17,7 +17,6 @@
   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
 
@@ -261,10 +260,13 @@ def _with_geometry(
     )
 
   def _simulated_x_reach_max(self) -> float:
-    """Rightmost reachable X (mm) in simulation, from the deck's reachable range."""
+    """Rightmost reachable X (mm) in simulation, from the deck's width.
+
+    The hardware backend reads the true drive travel from the RU query; here there is no
+    machine, so the deck's own width is the closest bound that covers every on-deck
+    position without under-reporting reach (a per-rail estimate clips the deck's right edge).
+    """
     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
@@ -331,9 +333,6 @@ async def channels_request_y_minimum_spacing(self) -> List[float]:
   async def move_channel_y(self, channel: int, y: float):
     logger.info("moving channel %s to y: %s", channel, y)
 
-  async def move_channel_x(self, channel: int, x: float):
-    logger.info("moving channel %s to x: %s", channel, x)
-
   async def move_all_channels_in_z_safety(self):
     logger.info("moving all channels to z safety")
 

pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py

@@ -5,6 +5,7 @@
 import datetime
 import unittest
 import unittest.mock
+from dataclasses import replace
 from typing import Literal, cast
 
 from pylabrobot.arms.standard import CartesianCoords
@@ -32,7 +33,7 @@
 )
 from pylabrobot.resources.barcode import Barcode
 from pylabrobot.resources.greiner import Greiner_384_wellplate_28ul_Fb
-from pylabrobot.resources.hamilton import STARLetDeck, hamilton_96_tiprack_300uL_filter
+from pylabrobot.resources.hamilton import STARDeck, STARLetDeck, hamilton_96_tiprack_300uL_filter
 
 from .STAR_backend import (
   CommandSyntaxError,
@@ -1955,7 +1956,17 @@ async def asyncSetUp(self):
 
     self.star._num_channels = 8
     self.star._machine_conf = _DEFAULT_MACHINE_CONFIGURATION
-    self.star._extended_conf = _DEFAULT_EXTENDED_CONFIGURATION
+    # setup() is mocked out below, so seed the left X-drive geometry it would normally
+    # resolve; the foil ops move channels in X, which is now bounds-checked against x_range.
+    self.star._extended_conf = replace(
+      _DEFAULT_EXTENDED_CONFIGURATION,
+      left_x_drive=replace(
+        _DEFAULT_EXTENDED_CONFIGURATION.left_x_drive,
+        width=370.0,
+        x_range=(95.0, 1337.5),
+        workspace_range=(-323.2, 1337.5),
+      ),
+    )
     self.star.setup = unittest.mock.AsyncMock()
     self.star._core_parked = True
     self.star._iswap_parked = True
@@ -2616,6 +2627,27 @@ def test_model_and_reference_by_width(self):
     self.assertEqual(single.reference_point, "right")
 
 
+class TestXArmReachByDeck(unittest.IsolatedAsyncioTestCase):
+  """A STAR reaches farther in X than a STARLet: an X past the STARLet's right edge
+  is reachable on a STAR but rejected on a STARLet."""
+
+  async def test_x_past_starlet_edge_reachable_on_star_only(self):
+    star = LiquidHandler(STARChatterboxBackend(), deck=STARDeck())
+    await star.setup()
+    starlet = LiquidHandler(STARChatterboxBackend(), deck=STARLetDeck())
+    await starlet.setup()
+
+    star_range = star.backend.extended_conf.left_x_drive.x_range
+    starlet_range = starlet.backend.extended_conf.left_x_drive.x_range
+    assert star_range is not None and starlet_range is not None
+    self.assertGreater(star_range[1], starlet_range[1])
+
+    x = (starlet_range[1] + star_range[1]) / 2  # past the STARLet edge, within STAR reach
+    star.backend._check_x_arm_reachable(x)  # reachable on STAR: no raise
+    with self.assertRaises(ValueError):
+      starlet.backend._check_x_arm_reachable(x)
+
+
 class TestXArmRangeQueries(unittest.IsolatedAsyncioTestCase):
   """RU/UA parse the firmware replies observed on real machines."""
 
@@ -2632,3 +2664,47 @@ 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))})
+
+
+class TestXArmRangeEnforcement(unittest.IsolatedAsyncioTestCase):
+  """move_channel_x / experimental_x_arm_move reject targets outside the arm's x_range."""
+
+  def setUp(self):
+    self.star = STARBackend()
+    # setup() normally resolves this from firmware; seed the left X-drive geometry so the
+    # reachability checks have a range to enforce against, without needing a deck-mounted
+    # arm or tracker.
+    self.star._extended_conf = replace(
+      _DEFAULT_EXTENDED_CONFIGURATION,
+      left_x_drive=replace(
+        _DEFAULT_EXTENDED_CONFIGURATION.left_x_drive,
+        width=370.0,
+        x_range=(95.0, 1337.5),
+        workspace_range=(-323.2, 1337.5),
+      ),
+    )
+    self.star.send_command = unittest.mock.AsyncMock(return_value={})
+
+  async def test_experimental_x_arm_move_rejects_out_of_range(self):
+    # x_range is (95.0, 1337.5): both ends reject, and no wire command is sent.
+    for x in (94.0, 1400.0):
+      with self.assertRaises(ValueError):
+        await self.star.experimental_x_arm_move(x)
+    self.star.send_command.assert_not_awaited()
+
+  async def test_experimental_x_arm_move_in_range_sends_command(self):
+    # A target inside x_range passes the guard and reaches the wire unchanged.
+    await self.star.experimental_x_arm_move(500.0)
+    self.star.send_command.assert_awaited_once_with(
+      module="X0", command="XP", la="05000", lr="3", lw="7"
+    )
+
+  async def test_move_channel_x_rejects_out_of_range(self):
+    with self.assertRaises(ValueError):
+      await self.star.move_channel_x(0, 1400.0)
+    self.star.send_command.assert_not_awaited()
+
+  async def test_rejects_target_on_absent_right_arm(self):
+    # The default single-arm STAR has no right X-arm, so a right-arm target is rejected.
+    with self.assertRaises(ValueError):
+      self.star._check_x_arm_reachable(400.0, "right")