From 0b05a7cbd5d364a32287d063df690cc6e9c5291f Mon Sep 17 00:00:00 2001 From: Camillo Moschner Date: Sun, 19 Jul 2026 18:04:33 +0100 Subject: [PATCH 1/2] `STARBackend`: bound X-arm and channel moves to the resolved x_range Enforce the per-arm drive range from XArmInformation on the two X-move entry points. experimental_x_arm_move and move_channel_x now call _check_x_arm_reachable, which reads the x_arm_information property (raising if setup has not resolved the ranges yet) and rejects a target outside the arm's x_range with a clear ValueError before any wire command is sent. Replaces the hardcoded 90-1350mm literal in experimental_x_arm_move, and drops the STARChatterboxBackend move_channel_x stub so the bounds check runs in simulation too. Co-Authored-By: Claude Opus 4.8 --- .../backends/hamilton/STAR_backend.py | 39 ++++++++- .../backends/hamilton/STAR_chatterbox.py | 3 - .../backends/hamilton/STAR_tests.py | 80 ++++++++++++++++++- 3 files changed, 114 insertions(+), 8 deletions(-) diff --git a/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py b/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py index d29f7f20ce6..2f11245fb01 100644 --- a/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py +++ b/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 diff --git a/pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py b/pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py index 3868b46fc3f..da6c147a7e7 100644 --- a/pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py +++ b/pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py @@ -331,9 +331,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") diff --git a/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py b/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py index de87a7b6d39..592ff2fb744 100644 --- a/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py +++ b/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") From e5a635aa8d965589e367b2ae57b027d1b74ca265 Mon Sep 17 00:00:00 2001 From: Camillo Moschner Date: Wed, 22 Jul 2026 12:39:31 +0100 Subject: [PATCH 2/2] `STARChatterboxBackend`: derive simulated X reach from the deck width The simulated X reach used the last rail's position, which stops short of the deck's right edge, so once moves are bounded to the resolved x_range a simulated move to the right of the deck was spuriously rejected. Derive it from the deck width instead. The hardware backend is unaffected: it reads the true drive travel from the RU query. --- .../backends/hamilton/STAR_chatterbox.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py b/pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py index da6c147a7e7..e976569dbb6 100644 --- a/pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py +++ b/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