From db22606c619ca8d5afe864a576a4b27f380d0302 Mon Sep 17 00:00:00 2001 From: Camillo Moschner Date: Mon, 20 Jul 2026 15:39:38 +0100 Subject: [PATCH 1/5] `STARBackend`: track the X-arm carriage via a deck-owned `XArm` resource Add carriage-position tracking for the X-arm. The deck owns an `XArm` resource per present arm (created at setup from the resolved X-drive configuration), and the `XArm` owns an `XArmTracker` whose x is the resource's state - so the backend drives it from firmware and the Visualizer reads it through the standard state channel, like a TipSpot owns a TipTracker. The lowest-level X moves and position queries update the tracker: experimental_x_arm_move, position_left_x_arm_, position_right_x_arm_ commit the commanded target (invalidating on a failed move), and request_left_x_arm_position / request_right_x_arm_position record the read-back value. get_x_arm_position reports the tracked (left, right) reference points, None while unknown. The Visualizer draws the arm as a full-deck-height frame with a reference line at the tracked x, gliding on update. STARChatterboxBackend reports a simulated resting home in place of the position read. Co-Authored-By: Claude Opus 4.8 --- .../backends/hamilton/STAR_backend.py | 141 ++++++++++++-- .../backends/hamilton/STAR_chatterbox.py | 33 ++++ .../backends/hamilton/STAR_tests.py | 172 +++++++++++++++++- pylabrobot/resources/__init__.py | 2 + .../resources/hamilton/hamilton_decks.py | 37 ++++ pylabrobot/resources/x_arm.py | 51 ++++++ pylabrobot/resources/x_arm_tracker.py | 114 ++++++++++++ pylabrobot/visualizer/lib.js | 116 ++++++++++++ 8 files changed, 646 insertions(+), 20 deletions(-) create mode 100644 pylabrobot/resources/x_arm.py create mode 100644 pylabrobot/resources/x_arm_tracker.py diff --git a/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py b/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py index 2f11245fb01..678a6f7af44 100644 --- a/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py +++ b/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py @@ -99,12 +99,15 @@ ) from pylabrobot.resources.hamilton.hamilton_decks import ( HamiltonCoreGrippers, + HamiltonSTARDeck, rails_for_x_coordinate, ) from pylabrobot.resources.liquid import Liquid from pylabrobot.resources.rotation import Rotation from pylabrobot.resources.tip_tracker import does_tip_tracking from pylabrobot.resources.trash import Trash +from pylabrobot.resources.x_arm import XArm +from pylabrobot.resources.x_arm_tracker import XArmTracker T = TypeVar("T") @@ -2230,6 +2233,8 @@ async def set_up_arm_modules(): # the core grippers. self._core_parked = True + await self._setup_x_arms() + self._setup_done = True async def stop(self): @@ -2281,13 +2286,81 @@ async def experimental_x_arm_move( f"current_protection_limiter must be between 0 and 7, is {current_protection_limiter}" ) - return await self.send_command( - module="X0", - command="XP", - la=f"{round(x * 10):05}", - lr=str(acceleration_level), - lw=str(current_protection_limiter), - ) + tracker = self.left_x_arm_tracker + if tracker is not None and not tracker.is_disabled: + tracker.set_x(x, commit=False) + try: + resp = await self.send_command( + module="X0", + command="XP", + la=f"{round(x * 10):05}", + lr=str(acceleration_level), + lw=str(current_protection_limiter), + ) + except Exception: + if tracker is not None and not tracker.is_disabled: + tracker.invalidate() + raise + if tracker is not None and not tracker.is_disabled: + tracker.commit() + return resp + + @property + def left_x_arm_tracker(self) -> Optional[XArmTracker]: + """The left X-arm's tracker, owned by the deck's left X-arm (created at + setup). None before setup or without a deck.""" + return self._x_arm_tracker("left_x_arm") + + @property + def right_x_arm_tracker(self) -> Optional[XArmTracker]: + """The right X-arm's tracker, owned by the deck's right X-arm. None when no + right arm is installed, or before setup.""" + return self._x_arm_tracker("right_x_arm") + + def _x_arm_tracker(self, name: str) -> Optional[XArmTracker]: + if self._deck is not None and self._deck.has_resource(name): + x_arm = self._deck.get_resource(name) + if isinstance(x_arm, XArm): + return x_arm.tracker + return None + + def get_x_arm_position(self) -> Tuple[Optional[float], Optional[float]]: + """Tracked X-arm carriage reference points in mm, as (left, right). + + Either element is None while that X-arm's position is unknown: no tracked move or + position query has completed for it yet, the last tracked move failed, or that + X-arm is not present. + """ + left_tracker = self.left_x_arm_tracker + right_tracker = self.right_x_arm_tracker + left = left_tracker.get_x() if left_tracker is not None and left_tracker.is_known else None + right = right_tracker.get_x() if right_tracker is not None and right_tracker.is_known else None + return left, right + + async def _setup_x_arms(self) -> None: + """Create the deck-owned X-arm for each present arm and seed its tracker. + + The X-arm (an ``XArm``) owns its ``XArmTracker`` and reports it as state, so the + Visualizer positions it; this asks the deck to create it (sized and modelled from + the resolved X-drive configuration) and seeds the tracker from a position read (its + resting home in simulation, its true position on hardware). No-op without a deck. + """ + if self._deck is None: + return + deck = cast(HamiltonSTARDeck, self.deck) + drives = { + "left": self.extended_conf.left_x_drive, + "right": self.extended_conf.right_x_drive, + } + queries = {"left": self.request_left_x_arm_position, "right": self.request_right_x_arm_position} + for position in ("left", "right"): + arm = drives[position] + if arm is None: + continue + if arm.width is None: + raise RuntimeError(f"{position} X-arm geometry not resolved") + deck.get_or_create_x_arm(f"{position}_x_arm", arm.width, arm.model, arm.reference_point) + await queries[position]() 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``). @@ -6208,11 +6281,22 @@ async def position_left_x_arm_(self, x_position: int = 0): assert 0 <= x_position <= 30000, "x_position_ must be between 0 and 30000" - return await self.send_command( - module="C0", - command="JX", - xs=f"{x_position:05}", - ) + tracker = self.left_x_arm_tracker + if tracker is not None and not tracker.is_disabled: + tracker.set_x(x_position / 10, commit=False) + try: + resp = await self.send_command( + module="C0", + command="JX", + xs=f"{x_position:05}", + ) + except Exception: + if tracker is not None and not tracker.is_disabled: + tracker.invalidate() + raise + if tracker is not None and not tracker.is_disabled: + tracker.commit() + return resp async def position_right_x_arm_(self, x_position: int = 0): """Position right X-Arm @@ -6225,11 +6309,22 @@ async def position_right_x_arm_(self, x_position: int = 0): assert 0 <= x_position <= 30000, "x_position_ must be between 0 and 30000" - return await self.send_command( - module="C0", - command="JS", - xs=f"{x_position:05}", - ) + tracker = self.right_x_arm_tracker + if tracker is not None and not tracker.is_disabled: + tracker.set_x(x_position / 10, commit=False) + try: + resp = await self.send_command( + module="C0", + command="JS", + xs=f"{x_position:05}", + ) + except Exception: + if tracker is not None and not tracker.is_disabled: + tracker.invalidate() + raise + if tracker is not None and not tracker.is_disabled: + tracker.commit() + return resp async def move_left_x_arm_to_position_with_all_attached_components_in_z_safety_position( self, x_position: int = 0 @@ -6338,13 +6433,21 @@ async def release_all_occupied_areas(self): async def request_left_x_arm_position(self) -> float: """Request left X-Arm position""" resp_dmm = await self.send_command(module="C0", command="RX", fmt="rx#####") - return cast(float, resp_dmm["rx"]) / 10 + x = cast(float, resp_dmm["rx"]) / 10 + tracker = self.left_x_arm_tracker + if tracker is not None and not tracker.is_disabled: + tracker.set_x(x) + return x async def request_right_x_arm_position(self) -> float: """Request right X-Arm position""" resp_dmm = await self.send_command(module="C0", command="QX", fmt="rx#####") - return cast(float, resp_dmm["rx"]) / 10 + x = cast(float, resp_dmm["rx"]) / 10 + tracker = self.right_x_arm_tracker + if tracker is not None and not tracker.is_disabled: + tracker.set_x(x) + return x async def request_maximal_ranges_of_x_drives(self) -> Dict[str, Tuple[float, float]]: """Request the maximal travel range of each X drive. diff --git a/pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py b/pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py index e976569dbb6..3ed78d75c54 100644 --- a/pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py +++ b/pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py @@ -45,6 +45,13 @@ # literal - a physical STAR reports its own value from the drive-range query. _DUAL_RAIL_LEFT_X_MIN = 95.0 +# Simulated left X-arm resting x (mm), reported in place of a position read before any +# tracked move. Small enough to sit within reach of any STAR deck. The right arm's rest +# is derived from the deck instead (see _simulated_right_x_arm_home), since its reach +# differs sharply between a STARlet (~800 mm) and a STAR (~1340 mm) and the two arms +# must not overlap. +_SIM_LEFT_X_ARM_HOME = 362.9 + # 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 @@ -196,6 +203,8 @@ async def setup( else: self._iswap_information = None + await self._setup_x_arms() + async def stop(self): await LiquidHandlerBackend.stop(self) self._setup_done = False @@ -285,6 +294,30 @@ async def request_working_envelopes_per_arm( right = (595.2, workspace) if right_installed else (0.0, (0.0, 0.0)) return {"left": left, "right": right} + async def request_left_x_arm_position(self) -> float: + # No hardware to read: report the tracked carriage position, or a resting + # home before any tracked move has committed one. + tracker = self.left_x_arm_tracker + x = tracker.get_x() if tracker is not None and tracker.is_known else _SIM_LEFT_X_ARM_HOME + if tracker is not None and not tracker.is_disabled: + tracker.set_x(x) + return x + + async def request_right_x_arm_position(self) -> float: + tracker = self.right_x_arm_tracker + if tracker is None: + raise RuntimeError("Right X-arm is not installed.") + x = tracker.get_x() if tracker.is_known else self._simulated_right_x_arm_home() + if not tracker.is_disabled: + tracker.set_x(x) + return x + + def _simulated_right_x_arm_home(self) -> float: + """Rightmost on-deck resting x for the right X-arm in simulation, derived from the + deck's reachable range so it is valid for the deck in use and clear of the left + arm's rest. Positions the arm so its right edge sits at the far reachable x.""" + return self._simulated_x_reach_max() - self.extended_conf.right_x_arm_width / 2 + # # # # # # # # 1_000 uL Channel: Basic Commands # # # # # # # # async def request_tip_presence(self) -> List[Optional[bool]]: diff --git a/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py b/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py index 592ff2fb744..1efc88161b5 100644 --- a/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py +++ b/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py @@ -33,7 +33,13 @@ ) from pylabrobot.resources.barcode import Barcode from pylabrobot.resources.greiner import Greiner_384_wellplate_28ul_Fb -from pylabrobot.resources.hamilton import STARDeck, STARLetDeck, hamilton_96_tiprack_300uL_filter +from pylabrobot.resources.hamilton import ( + HamiltonSTARDeck, + STARDeck, + STARLetDeck, + hamilton_96_tiprack_300uL_filter, +) +from pylabrobot.resources.x_arm import XArm from .STAR_backend import ( CommandSyntaxError, @@ -2708,3 +2714,167 @@ 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") + + +class TestXArmPositionTracking(unittest.IsolatedAsyncioTestCase): + """The lowest-level X-arm move and position-query commands feed the deck X-arms' + X-arm trackers; no other commands update them.""" + + def setUp(self): + self.star = STARBackend() + self.deck = STARLetDeck() + self.star.set_deck(self.deck) + # X-arms (which own the trackers) are created by setup(); create the left one + # directly here so the low-level commands can be exercised with a mocked wire. + self.deck.get_or_create_x_arm( + "left_x_arm", 370.0, "hamilton_legacy_star_dual_rail_arm", "center" + ) + # setup() normally resolves the X-drive geometry from firmware; seed it so the + # reachability checks (move_channel_x / experimental_x_arm_move) have a range. + self.star._extended_conf = replace( + _DEFAULT_EXTENDED_CONFIGURATION, + left_x_drive=self._x_drive(_DEFAULT_EXTENDED_CONFIGURATION.left_x_drive), + ) + self.star.send_command = unittest.mock.AsyncMock(return_value={}) + + @staticmethod + def _x_drive(base): + return replace(base, width=370.0, x_range=(95.0, 1337.5), workspace_range=(-323.2, 1337.5)) + + def _add_right_x_arm(self): + self.deck.get_or_create_x_arm( + "right_x_arm", 370.0, "hamilton_legacy_star_dual_rail_arm", "center" + ) + conf = self.star._extended_conf + assert conf is not None + self.star._extended_conf = replace(conf, right_x_drive=self._x_drive(DriveConfiguration())) + + async def test_position_unknown_before_first_tracked_command(self): + self.assertEqual(self.star.get_x_arm_position(), (None, None)) + + async def test_experimental_x_arm_move_updates_tracker(self): + 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" + ) + self.assertEqual(self.star.get_x_arm_position(), (500.0, None)) + + async def test_position_left_x_arm_updates_tracker(self): + await self.star.position_left_x_arm_(12345) + self.star.send_command.assert_awaited_once_with(module="C0", command="JX", xs="12345") + self.assertEqual(self.star.get_x_arm_position(), (1234.5, None)) + + async def test_request_left_x_arm_position_updates_tracker(self): + self.star.send_command.return_value = {"rx": 6789} + x = await self.star.request_left_x_arm_position() + self.assertEqual(x, 678.9) + self.assertEqual(self.star.get_x_arm_position(), (678.9, None)) + + async def test_position_right_x_arm_updates_right_tracker(self): + self._add_right_x_arm() + await self.star.position_left_x_arm_(5000) + await self.star.position_right_x_arm_(11111) + self.star.send_command.assert_awaited_with(module="C0", command="JS", xs="11111") + self.assertEqual(self.star.get_x_arm_position(), (500.0, 1111.1)) + + async def test_request_right_x_arm_position_updates_right_tracker(self): + self._add_right_x_arm() + self.star.send_command.return_value = {"rx": 9876} + x = await self.star.request_right_x_arm_position() + self.assertEqual(x, 987.6) + assert self.star.right_x_arm_tracker is not None + self.assertEqual(self.star.right_x_arm_tracker.get_x(), 987.6) + + async def test_right_arm_command_without_right_x_arm_is_a_noop(self): + # No right X-arm: the command still sends, tracking is skipped. + await self.star.position_right_x_arm_(11111) + self.star.send_command.assert_awaited_with(module="C0", command="JS", xs="11111") + self.assertIsNone(self.star.right_x_arm_tracker) + + async def test_right_unknown_reported_as_none(self): + await self.star.position_left_x_arm_(5000) + self.assertEqual(self.star.get_x_arm_position(), (500.0, None)) + + async def test_failed_move_invalidates_position(self): + await self.star.experimental_x_arm_move(500.0) + self.star.send_command.side_effect = TimeoutError() + with self.assertRaises(TimeoutError): + await self.star.experimental_x_arm_move(600.0) + assert self.star.left_x_arm_tracker is not None + self.assertFalse(self.star.left_x_arm_tracker.is_known) + + async def test_disabled_tracker_is_not_updated(self): + assert self.star.left_x_arm_tracker is not None + self.star.left_x_arm_tracker.disable() + await self.star.experimental_x_arm_move(500.0) + self.star.left_x_arm_tracker.enable() + self.assertFalse(self.star.left_x_arm_tracker.is_known) + + +class TestXArmPresence(unittest.IsolatedAsyncioTestCase): + """setup() creates a deck X-arm (and thus a tracker) per present arm.""" + + async def _setup_deck(self, right_present: bool) -> HamiltonSTARDeck: + conf = copy.deepcopy(_DEFAULT_EXTENDED_CONFIGURATION) + if right_present: + conf.right_x_drive = DriveConfiguration(iswap_installed=True) + deck = STARLetDeck() + lh = LiquidHandler(STARChatterboxBackend(extended_configuration=conf), deck=deck) + await lh.setup() + return deck + + async def test_left_only_config_creates_only_left_x_arm(self): + deck = await self._setup_deck(right_present=False) + x_arms = [c.name for c in deck.children if c.category == "x_arm"] + self.assertEqual(x_arms, ["left_x_arm"]) + + async def test_right_present_creates_both_x_arms(self): + deck = await self._setup_deck(right_present=True) + x_arms = sorted(c.name for c in deck.children if c.category == "x_arm") + self.assertEqual(x_arms, ["left_x_arm", "right_x_arm"]) + + +class TestXArmVisualizerXArms(unittest.IsolatedAsyncioTestCase): + """The deck owns an X-arm per present arm; the X-arm owns the tracker and + reports it as state, so the backend drives it and the Visualizer reads it.""" + + async def asyncSetUp(self): + self.deck = STARLetDeck() + self.lh = LiquidHandler(STARChatterboxBackend(), deck=self.deck) + await self.lh.setup() + + async def test_left_only_creates_one_x_arm(self): + x_arms = [c.name for c in self.deck.children if c.category == "x_arm"] + self.assertEqual(x_arms, ["left_x_arm"]) + + async def test_x_arm_dimensions_and_geometry(self): + x_arm = self.deck.get_resource("left_x_arm") + self.assertEqual(x_arm.get_size_x(), self.lh.backend.extended_conf.left_x_arm_width) + self.assertEqual(x_arm.get_size_y(), self.deck.get_size_y()) + self.assertEqual(x_arm.get_size_z(), 140.0) + self.assertEqual(x_arm.category, "x_arm") + self.assertEqual(x_arm.model, "hamilton_legacy_star_dual_rail_arm") + assert x_arm.location is not None + self.assertEqual(x_arm.location.z, 248.0) + + async def test_x_arm_owns_a_tracker_reported_as_state(self): + x_arm = self.deck.get_resource("left_x_arm") + self.assertIsInstance(x_arm, XArm) + self.assertIn("tracker", x_arm.serialize_state()) + + async def test_seeded_at_home_not_zero(self): + # The arm must not start at x=0 (border-crash region); it seeds from a position read. + self.assertGreater(self.lh.backend.get_x_arm_position()[0], 0.0) + + async def test_tracked_move_updates_x_arm_tracker(self): + await self.lh.backend.experimental_x_arm_move(500.0) + x_arm = self.deck.get_resource("left_x_arm") + self.assertEqual(x_arm.tracker.get_x(), 500.0) + self.assertEqual(self.lh.backend.get_x_arm_position()[0], 500.0) + + async def test_resetup_reuses_x_arm(self): + # Regression: re-setup must not crash or duplicate the deck-owned x_arm. + await self.lh.stop() + await self.lh.setup() + x_arms = [c.name for c in self.deck.children if c.category == "x_arm"] + self.assertEqual(x_arms, ["left_x_arm"]) diff --git a/pylabrobot/resources/__init__.py b/pylabrobot/resources/__init__.py index ca92f0edc5b..00a5922ca95 100644 --- a/pylabrobot/resources/__init__.py +++ b/pylabrobot/resources/__init__.py @@ -70,3 +70,5 @@ ) from .vwr import * from .well import CrossSectionType, Well, WellBottomType +from .x_arm import XArm +from .x_arm_tracker import XArmTracker diff --git a/pylabrobot/resources/hamilton/hamilton_decks.py b/pylabrobot/resources/hamilton/hamilton_decks.py index 43ebc1f9cc4..5a4bc7e22b3 100644 --- a/pylabrobot/resources/hamilton/hamilton_decks.py +++ b/pylabrobot/resources/hamilton/hamilton_decks.py @@ -12,12 +12,17 @@ from pylabrobot.resources.resource import Resource from pylabrobot.resources.tip_rack import TipRack, TipSpot from pylabrobot.resources.trash import Trash +from pylabrobot.resources.x_arm import XArm logger = logging.getLogger("pylabrobot") _RAILS_WIDTH = 22.5 # space between rails (mm) +# The deck-owned X-arm resource spans the deck height; z places it above the deck. +_X_ARM_Z = 248.0 +_X_ARM_SIZE_Z = 140.0 + STARLET_NUM_RAILS = 32 STARLET_SIZE_X = 1005 STARLET_SIZE_Y = 653.5 @@ -543,6 +548,38 @@ def serialize(self) -> dict: "core_grippers": None, # data encoded as child. (not very pretty to have this key though...) } + def get_or_create_x_arm( + self, name: str, width: float, model: str, reference_point: Literal["center", "right"] + ) -> XArm: + """Get (or create, once) the deck-owned X-arm resource for `name`. + + The X-arm is a full-deck-height :class:`~pylabrobot.resources.x_arm.XArm` that owns + an ``XArmTracker``; the Visualizer draws it at the tracker's x. The deck owns it: + created as a child the first time and reused thereafter, so repeated ``setup()`` + calls don't duplicate it. It is assigned at a nominal x - the backend drives its + tracker and the Visualizer positions it from that. + + Args: + name: unique resource name, e.g. ``"left_x_arm"``. + width: arm width in mm (the firmware-reported value). + model: the arm variant, e.g. ``"hamilton_legacy_star_dual_rail_arm"``. + reference_point: where the tracked x refers to along the width. + """ + if self.has_resource(name): + return cast(XArm, self.get_resource(name)) + x_arm = XArm( + name=name, + size_x=width, + size_y=self.get_size_y(), + size_z=_X_ARM_SIZE_Z, + reference_point=reference_point, + model=model, + ) + self.assign_child_resource( + x_arm, location=Coordinate(0.0, 0.0, _X_ARM_Z), ignore_collision=True + ) + return x_arm + def rails_to_location(self, rails: int) -> Coordinate: x = 100.0 + (rails - 1) * _RAILS_WIDTH return Coordinate(x=x, y=63, z=100) diff --git a/pylabrobot/resources/x_arm.py b/pylabrobot/resources/x_arm.py new file mode 100644 index 00000000000..a5ba1b6a46a --- /dev/null +++ b/pylabrobot/resources/x_arm.py @@ -0,0 +1,51 @@ +from typing import Any, Dict, Literal + +from pylabrobot.resources.resource import Resource +from pylabrobot.resources.x_arm_tracker import XArmTracker + + +class XArm(Resource): + """A model of an X-arm carriage, owned by the deck. + + Like a :class:`~pylabrobot.resources.tip_rack.TipSpot` owns a ``TipTracker``, an + ``XArm`` owns an :class:`XArmTracker`: the tracker's x is the resource's state, so + it reaches the Visualizer through the standard state channel. The backend drives + the tracker (from firmware); the Visualizer reads its x and draws it there. + + ``reference_point`` says where along the arm's width the tracked x refers to (the + arm centre for a dual-rail arm, the right edge for a single-rail arm), so the + Visualizer positions the X-arm without re-deriving the convention. + """ + + def __init__( + self, + name: str, + size_x: float, + size_y: float, + size_z: float, + reference_point: Literal["center", "right"] = "center", + category: str = "x_arm", + model: str = "hamilton_legacy_star_dual_rail_arm", + ): + super().__init__( + name=name, + size_x=size_x, + size_y=size_y, + size_z=size_z, + category=category, + model=model, + ) + self.reference_point = reference_point + self.tracker = XArmTracker(thing=name) + self.tracker.register_callback(self._state_updated) + + def serialize(self) -> Dict[str, Any]: + return {**super().serialize(), "reference_point": self.reference_point} + + def serialize_state(self) -> Dict[str, Any]: + return {**super().serialize_state(), "tracker": self.tracker.serialize()} + + def load_state(self, state: Dict[str, Any]) -> None: + super().load_state(state) + if "tracker" in state: + self.tracker.load_state(state["tracker"]) diff --git a/pylabrobot/resources/x_arm_tracker.py b/pylabrobot/resources/x_arm_tracker.py new file mode 100644 index 00000000000..d55f0b3a0c0 --- /dev/null +++ b/pylabrobot/resources/x_arm_tracker.py @@ -0,0 +1,114 @@ +from typing import Callable, Optional + +from pylabrobot.serializer import SerializableMixin + +XArmTrackerCallback = Callable[[], None] + + +class XArmTracker(SerializableMixin): + """Tracks the X-arm carriage reference point in x (mm, deck coordinates). + + The tracked value mirrors the firmware's carriage position report, rounded to the + 0.1 mm firmware granularity. Positions of components attached to the arm (channels, + 96-head, iSWAP rotation drive) are derived from this reference point via their x + offsets and are not stored here. + + The position starts unknown and becomes known when the first tracked move commits. + A failed move invalidates the position: after e.g. a collision the physical + position cannot be inferred from the command, so the tracker returns to unknown. + """ + + def __init__(self, thing: str): + self.thing = thing + self._is_disabled = False + self._x: Optional[float] = None + self._pending_x: Optional[float] = None + + self._callback: Optional[XArmTrackerCallback] = None + + @property + def is_disabled(self) -> bool: + return self._is_disabled + + @property + def is_known(self) -> bool: + """Whether the tracker has a committed position. Pending operations don't count.""" + return self._x is not None + + def disable(self) -> None: + """Disable the tracker.""" + self._is_disabled = True + + def enable(self) -> None: + """Enable the tracker.""" + self._is_disabled = False + + def get_x(self) -> float: + """Get the committed X-arm carriage position in mm. + + Raises: + RuntimeError: If no tracked move has committed yet, or the last tracked + move failed. + """ + if self._x is None: + raise RuntimeError(f"{self.thing} position is unknown.") + return self._x + + def set_x(self, x: float, commit: bool = True) -> None: + """Record a commanded move target in mm, rounded to 0.1 mm. + + Args: + x: The target position in mm. + commit: Whether to commit the operation immediately. If `False`, the operation + will be committed later with `commit()` or rolled back with `rollback()`. + """ + if self.is_disabled: + raise RuntimeError("X-arm tracker is disabled. Call `enable()`.") + self._pending_x = round(x, 1) + + if commit: + self.commit() + + def invalidate(self) -> None: + """Mark the position as unknown, e.g. after a failed move.""" + if self.is_disabled: + raise RuntimeError("X-arm tracker is disabled. Call `enable()`.") + self._x = None + self._pending_x = None + + if self._callback is not None: + self._callback() + + def commit(self) -> None: + """Commit the pending operations.""" + assert not self.is_disabled, "X-arm tracker is disabled. Call `enable()`." + self._x = self._pending_x + + if self._callback is not None: + self._callback() + + def rollback(self) -> None: + """Rollback the pending operations.""" + assert not self.is_disabled, "X-arm tracker is disabled. Call `enable()`." + self._pending_x = self._x + + def serialize(self) -> dict: + """Serialize the state of the tracker.""" + return { + "x": self._x, + "pending_x": self._pending_x, + } + + def load_state(self, state: dict) -> None: + """Load a saved tracker state.""" + self._x = state["x"] + self._pending_x = state["pending_x"] + + def register_callback(self, callback: XArmTrackerCallback) -> None: + self._callback = callback + + def __repr__(self) -> str: + return ( + f"XArmTracker({self.thing}, is_disabled={self.is_disabled}, x={self._x}," + f" pending_x={self._pending_x})" + ) diff --git a/pylabrobot/visualizer/lib.js b/pylabrobot/visualizer/lib.js index 01ef98e04f1..8e613fa8436 100644 --- a/pylabrobot/visualizer/lib.js +++ b/pylabrobot/visualizer/lib.js @@ -3017,6 +3017,120 @@ function fillArmPanel(panel, armState) { } } +class XArm extends Resource { + // Visual model of an X-arm carriage: a translucent dark-grey full-deck-height frame + // (a rectangle with a rectangular hole) with a cyan reference line at the tracked x. + // The X-arm owns an XArmTracker (like TipSpot owns a TipTracker); its serialized + // tracker x drives the position - the group's left edge is placed at tracked_x minus + // the reference offset (the arm centre, or the right edge for a single-rail arm), so + // the frame is drawn as a normal resource (local [0, size_x]) and its bounding box + // and selection highlight stay aligned. The x glides on update so moves read as motion. + draggable = false; + canDelete = false; + + constructor(resourceData, parent = undefined) { + super(resourceData, parent); + this.reference_point = resourceData.reference_point || "center"; + } + + // Local x of the firmware reference within the frame (also the reference line's x). + get referenceOffset() { + return this.reference_point === "center" ? this.size_x / 2 : this.size_x; + } + + drawMainShape() { + const w = this.size_x; + const h = this.size_y; + const insetX = 95; // mm from each x border to the hole + const insetY = 20; // mm from each y border to the hole + const innerW = w - 2 * insetX; + const innerH = h - 2 * insetY; + + const g = new Konva.Group(); + // Reference line at the exact firmware x (local = referenceOffset), so it always + // marks the tracked x. Added first so the frame draws on top of it, visible only + // through the hole. + g.add( + new Konva.Line({ + points: [this.referenceOffset, 0, this.referenceOffset, h], + stroke: "rgba(0, 229, 255, 0.7)", + strokeWidth: 2, + listening: false, + }) + ); + g.add( + new Konva.Shape({ + sceneFunc: (ctx, shape) => { + ctx.beginPath(); + ctx.rect(0, 0, w, h); // outer, clockwise + if (innerW > 0 && innerH > 0) { + // inner rectangle wound counter-clockwise to punch a see-through hole + const l = insetX; + const r = w - insetX; + const t = insetY; + const b = h - insetY; + ctx.moveTo(l, t); + ctx.lineTo(l, b); + ctx.lineTo(r, b); + ctx.lineTo(r, t); + ctx.closePath(); + } + ctx.fillStrokeShape(shape); + }, + fill: "rgba(64, 64, 64, 0.5)", + stroke: "rgba(64, 64, 64, 0.3)", + strokeWidth: 3, + listening: false, + }) + ); + return g; + } + + setState(state) { + super.setState(state); // rotation + if (this.group === undefined) { + return; + } + // Position from the owned tracker's x. Hidden until a position is known; the + // group's left edge sits at tracked_x - referenceOffset. The x glides so moves + // read as motion rather than teleports (the tracker only has commanded targets, + // so this is purely cosmetic interpolation). + const trackerState = state && state.tracker; + const x = trackerState ? trackerState.x : null; + if (x === null || x === undefined) { + this.group.visible(false); + return; + } + this.group.visible(true); + const leftEdge = x - this.referenceOffset; + const reduce = + window.matchMedia && window.matchMedia("(prefers-reduced-motion: reduce)").matches; + if (reduce) { + this.group.x(leftEdge); + } else { + if (this._xTween) { + this._xTween.destroy(); + } + this._xTween = new Konva.Tween({ + node: this.group, + x: leftEdge, + duration: 0.2, + easing: Konva.Easings.EaseInOut, + }); + this._xTween.play(); + } + } + + destroy() { + // Stop any in-flight glide so it can't drive updates against the destroyed node. + if (this._xTween) { + this._xTween.destroy(); + this._xTween = undefined; + } + super.destroy(); + } +} + class LiquidHandler extends Resource { constructor(resource) { super(resource); @@ -3174,6 +3288,8 @@ function classForResourceType(type, category) { case "container": case "trough": return Container; + case "x_arm": + return XArm; default: return Resource; } From 1c0c44d1133cd31d32420012815ef0b3467769ce Mon Sep 17 00:00:00 2001 From: Camillo Moschner Date: Wed, 22 Jul 2026 19:36:32 +0100 Subject: [PATCH 2/5] `HamiltonDeck`: check placement overlap in 3D so above-deck resources do not block carriers The placement collision check compared only the x/y footprint, so a resource above the deck plane - the deck-owned X-arm - was treated as occupying every rail beneath it and rejected carrier assignment. Include the z axis: two resources collide only when their bounding boxes intersect on all three axes. Also seat the X-arm at the channel stop-disk safe Z (334.7 mm) rather than an eyeballed height, matching the design where the arm rides level with the raised stop disks to clear them during X travel. --- .../backends/hamilton/STAR_tests.py | 2 +- .../resources/hamilton/hamilton_deck_tests.py | 28 ++++++++++++++++++- .../resources/hamilton/hamilton_decks.py | 23 +++++++++++---- 3 files changed, 45 insertions(+), 8 deletions(-) diff --git a/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py b/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py index 1efc88161b5..a2527df98e0 100644 --- a/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py +++ b/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py @@ -2855,7 +2855,7 @@ async def test_x_arm_dimensions_and_geometry(self): self.assertEqual(x_arm.category, "x_arm") self.assertEqual(x_arm.model, "hamilton_legacy_star_dual_rail_arm") assert x_arm.location is not None - self.assertEqual(x_arm.location.z, 248.0) + self.assertEqual(x_arm.location.z, 334.7) # channel stop-disk safe Z async def test_x_arm_owns_a_tracker_reported_as_state(self): x_arm = self.deck.get_resource("left_x_arm") diff --git a/pylabrobot/resources/hamilton/hamilton_deck_tests.py b/pylabrobot/resources/hamilton/hamilton_deck_tests.py index 71d81f5fd71..d9b29664717 100644 --- a/pylabrobot/resources/hamilton/hamilton_deck_tests.py +++ b/pylabrobot/resources/hamilton/hamilton_deck_tests.py @@ -1,7 +1,7 @@ import textwrap import unittest -from pylabrobot.resources import TipRack +from pylabrobot.resources import Coordinate, Resource, TipRack from pylabrobot.resources.corning import ( cor_96_wellplate_360uL_Fb, ) @@ -108,3 +108,29 @@ def test_assign_gigantic_resource(self): "careful when grabbing this resource.", ], ) + + def test_resource_above_deck_plane_does_not_block(self): + """A resource whose z-box sits entirely above another (e.g. the X-arm gantry over a + carrier) does not occupy its footprint, so placement beneath it is allowed.""" + deck = STARLetDeck() + low = Resource(name="low", size_x=100, size_y=100, size_z=50) + high = Resource(name="high", size_x=100, size_y=100, size_z=50) + deck.assign_child_resource(low, location=Coordinate(300, 100, 100)) + deck.assign_child_resource(high, location=Coordinate(300, 100, 200)) # z-box [200,250] > low + self.assertIn("high", {c.name for c in deck.children}) + + def test_same_footprint_overlapping_z_raises(self): + deck = STARLetDeck() + a = Resource(name="a", size_x=100, size_y=100, size_z=100) + b = Resource(name="b", size_x=100, size_y=100, size_z=100) + deck.assign_child_resource(a, location=Coordinate(300, 100, 100)) # z-box [100,200] + with self.assertRaises(ValueError): + deck.assign_child_resource(b, location=Coordinate(300, 100, 150)) # z-box [150,250] overlaps + + def test_flush_z_faces_do_not_collide(self): + deck = STARLetDeck() + a = Resource(name="a", size_x=100, size_y=100, size_z=50) + b = Resource(name="b", size_x=100, size_y=100, size_z=50) + deck.assign_child_resource(a, location=Coordinate(300, 100, 100)) # top at 150 + deck.assign_child_resource(b, location=Coordinate(300, 100, 150)) # rests flush on a + self.assertIn("b", {c.name for c in deck.children}) diff --git a/pylabrobot/resources/hamilton/hamilton_decks.py b/pylabrobot/resources/hamilton/hamilton_decks.py index 5a4bc7e22b3..d79b37e0d9a 100644 --- a/pylabrobot/resources/hamilton/hamilton_decks.py +++ b/pylabrobot/resources/hamilton/hamilton_decks.py @@ -20,7 +20,10 @@ _RAILS_WIDTH = 22.5 # space between rails (mm) # The deck-owned X-arm resource spans the deck height; z places it above the deck. -_X_ARM_Z = 248.0 +# The X-arm rides at the channel stop-disk safe Z (matches +# STARBackend.MAXIMUM_CHANNEL_Z_POSITION): the arm is level with the raised stop disks so it +# clears them during X travel. Kept as a literal to avoid a resources -> backend import. +_X_ARM_Z = 334.7 _X_ARM_SIZE_Z = 140.0 STARLET_NUM_RAILS = 32 @@ -193,24 +196,32 @@ def should_check_collision(res: Resource) -> bool: for og_resource in self.children: og_x = cast(Coordinate, og_resource.location).x og_y = cast(Coordinate, og_resource.location).y + og_z = cast(Coordinate, og_resource.location).z - # A resource is not allowed to overlap with another resource. Resources overlap when a - # corner of one resource is inside the boundaries of another resource. - if any( + # A resource is not allowed to overlap with another resource. Resources overlap when + # their bounding boxes intersect on all three axes. The z axis is included so a resource + # above the deck plane (e.g. the X-arm gantry) does not block placement beneath it. + x_overlap = any( [ og_x <= resource_location.x < og_x + og_resource.get_absolute_size_x(), og_x < resource_location.x + resource.get_absolute_size_x() < og_x + og_resource.get_absolute_size_x(), ] - ) and any( + ) + y_overlap = any( [ og_y <= resource_location.y < og_y + og_resource.get_absolute_size_y(), og_y < resource_location.y + resource.get_absolute_size_y() < og_y + og_resource.get_absolute_size_y(), ] - ): + ) + z_overlap = ( + og_z < resource_location.z + resource.get_absolute_size_z() + and resource_location.z < og_z + og_resource.get_absolute_size_z() + ) + if x_overlap and y_overlap and z_overlap: raise ValueError( f"Location {resource_location} is already occupied by resource '{og_resource.name}'." ) From 55192409d883f89395d49586a407c671d9fd80e0 Mon Sep 17 00:00:00 2001 From: Camillo Moschner Date: Wed, 22 Jul 2026 19:43:00 +0100 Subject: [PATCH 3/5] `Visualizer`: draw the X-arm above the deck it overlays The X-arm is the highest resource on the deck, but the 2D canvas paints in scene-graph order, not z, so a resource added after the arm (e.g. a carrier assigned post-setup) painted over it. Re-float the arm to the top of the deck children after the events that add resources (set_root_resource, resource_assigned) so its translucent frame overlays the labware it spans; the reference line stays visible through the hole. Verified in a headless browser - the arm paints last, over an overlapping plate carrier. --- pylabrobot/visualizer/vis.js | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/pylabrobot/visualizer/vis.js b/pylabrobot/visualizer/vis.js index 43fa004f916..479c1b064e3 100644 --- a/pylabrobot/visualizer/vis.js +++ b/pylabrobot/visualizer/vis.js @@ -80,10 +80,23 @@ function setState(allStates) { } } +// The X-arm is physically the highest resource on the deck, so its translucent frame +// should overlay the labware it spans. 2D draw order is scene-graph order, not z, so a +// resource added after the arm lands on top of it; re-float the arm after any event that +// adds resources to keep it drawn above them. +function raiseXArmsToTop() { + for (const r of Object.values(resources)) { + if (r && r.category === "x_arm" && r.group) { + r.group.moveToTop(); + } + } +} + async function processCentralEvent(event, data) { switch (event) { case "set_root_resource": setRootResource(data); + raiseXArmsToTop(); break; case "resource_assigned": @@ -94,6 +107,7 @@ async function processCentralEvent(event, data) { addResourceToTree(resource); // Cache the newly-assigned plate/tip rack (or the owner it landed in). cacheResourceGroup(getCacheOwner(resource)); + raiseXArmsToTop(); break; case "resource_unassigned": From 0cfec908a0405c6c8f8bf62e0d65c083446bab93 Mon Sep 17 00:00:00 2001 From: Camillo Moschner Date: Wed, 22 Jul 2026 22:59:43 +0100 Subject: [PATCH 4/5] `HamiltonDeck`: seat each X-arm at its home x, dropping the collision override The X-arm resource was assigned at a nominal (0, 0), which put both arms in the same box and required ignore_collision to place the second one. Seat each arm so its reference point sits at the arm current x - read from the machine at setup (the left parks to a known home; the right reads its own) - so the two arms occupy distinct real positions and their placement is validated by the 3D overlap check rather than skipped. request_right_x_arm_position now gates on the configured right drive instead of the tracker existing, so setup can read the home before the tracker is created. --- .../backends/hamilton/STAR_backend.py | 7 +++++-- .../backends/hamilton/STAR_chatterbox.py | 14 +++++++++---- .../backends/hamilton/STAR_tests.py | 4 ++-- .../resources/hamilton/hamilton_decks.py | 21 +++++++++++++------ 4 files changed, 32 insertions(+), 14 deletions(-) diff --git a/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py b/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py index 678a6f7af44..89ae6d4dbf5 100644 --- a/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py +++ b/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py @@ -2359,8 +2359,11 @@ async def _setup_x_arms(self) -> None: continue if arm.width is None: raise RuntimeError(f"{position} X-arm geometry not resolved") - deck.get_or_create_x_arm(f"{position}_x_arm", arm.width, arm.model, arm.reference_point) - await queries[position]() + x = await queries[position]() # read the arm's current x before seating the resource + x_arm = deck.get_or_create_x_arm( + f"{position}_x_arm", x, arm.width, arm.model, arm.reference_point + ) + x_arm.tracker.set_x(x) # seed the tracker now that the arm (and its tracker) exists 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``). diff --git a/pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py b/pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py index 3ed78d75c54..16f9091654e 100644 --- a/pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py +++ b/pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py @@ -304,11 +304,17 @@ async def request_left_x_arm_position(self) -> float: return x async def request_right_x_arm_position(self) -> float: - tracker = self.right_x_arm_tracker - if tracker is None: + # Installed-ness is a config fact, not whether the tracker exists yet: setup reads the + # home to seat the arm before its tracker is created (mirrors request_left). + if self.extended_conf.right_x_drive is None: raise RuntimeError("Right X-arm is not installed.") - x = tracker.get_x() if tracker.is_known else self._simulated_right_x_arm_home() - if not tracker.is_disabled: + tracker = self.right_x_arm_tracker + x = ( + tracker.get_x() + if tracker is not None and tracker.is_known + else self._simulated_right_x_arm_home() + ) + if tracker is not None and not tracker.is_disabled: tracker.set_x(x) return x diff --git a/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py b/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py index a2527df98e0..68afebef662 100644 --- a/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py +++ b/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py @@ -2727,7 +2727,7 @@ def setUp(self): # X-arms (which own the trackers) are created by setup(); create the left one # directly here so the low-level commands can be exercised with a mocked wire. self.deck.get_or_create_x_arm( - "left_x_arm", 370.0, "hamilton_legacy_star_dual_rail_arm", "center" + "left_x_arm", 362.9, 370.0, "hamilton_legacy_star_dual_rail_arm", "center" ) # setup() normally resolves the X-drive geometry from firmware; seed it so the # reachability checks (move_channel_x / experimental_x_arm_move) have a range. @@ -2743,7 +2743,7 @@ def _x_drive(base): def _add_right_x_arm(self): self.deck.get_or_create_x_arm( - "right_x_arm", 370.0, "hamilton_legacy_star_dual_rail_arm", "center" + "right_x_arm", 800.0, 370.0, "hamilton_legacy_star_dual_rail_arm", "center" ) conf = self.star._extended_conf assert conf is not None diff --git a/pylabrobot/resources/hamilton/hamilton_decks.py b/pylabrobot/resources/hamilton/hamilton_decks.py index d79b37e0d9a..bb21755652c 100644 --- a/pylabrobot/resources/hamilton/hamilton_decks.py +++ b/pylabrobot/resources/hamilton/hamilton_decks.py @@ -560,18 +560,24 @@ def serialize(self) -> dict: } def get_or_create_x_arm( - self, name: str, width: float, model: str, reference_point: Literal["center", "right"] + self, + name: str, + x: float, + width: float, + model: str, + reference_point: Literal["center", "right"], ) -> XArm: """Get (or create, once) the deck-owned X-arm resource for `name`. The X-arm is a full-deck-height :class:`~pylabrobot.resources.x_arm.XArm` that owns an ``XArmTracker``; the Visualizer draws it at the tracker's x. The deck owns it: created as a child the first time and reused thereafter, so repeated ``setup()`` - calls don't duplicate it. It is assigned at a nominal x - the backend drives its - tracker and the Visualizer positions it from that. + calls don't duplicate it. It is seated so its reference point sits at the arm's + current x; the tracker then drives the live position and the Visualizer follows it. Args: name: unique resource name, e.g. ``"left_x_arm"``. + x: the arm's current x (mm); the reference point is seated here. width: arm width in mm (the firmware-reported value). model: the arm variant, e.g. ``"hamilton_legacy_star_dual_rail_arm"``. reference_point: where the tracked x refers to along the width. @@ -586,9 +592,12 @@ def get_or_create_x_arm( reference_point=reference_point, model=model, ) - self.assign_child_resource( - x_arm, location=Coordinate(0.0, 0.0, _X_ARM_Z), ignore_collision=True - ) + # Seat the frame so its reference point (the arm centre for a dual-rail arm, the right + # edge for a single-rail arm) lands at the arm's current x. Distinct per-arm homes keep + # the two arms from overlapping, so no collision override is needed; the z-separated, + # above-deck box also clears the carriers beneath it. + reference_offset = width / 2 if reference_point == "center" else width + self.assign_child_resource(x_arm, location=Coordinate(x - reference_offset, 0.0, _X_ARM_Z)) return x_arm def rails_to_location(self, rails: int) -> Coordinate: From 3dabf28237e0792b8db47d1020030059f2a5e24f Mon Sep 17 00:00:00 2001 From: Camillo Moschner Date: Thu, 23 Jul 2026 14:23:34 +0100 Subject: [PATCH 5/5] `Visualizer`: pass clicks through the X-arm gap, block under its solid frame The arm is drawn on top and translucent, so its frame was intercepting pointer events over everything it spans. Make the frame shape listening: its solid border occludes and captures clicks (you interact with the arm there) while the punched hole passes events through to the resources visible beneath. The hit region follows the same sceneFunc that draws the frame, so the hole is excluded automatically - no separate hit geometry to keep in sync. --- pylabrobot/visualizer/lib.js | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/pylabrobot/visualizer/lib.js b/pylabrobot/visualizer/lib.js index 8e613fa8436..3cb8be70770 100644 --- a/pylabrobot/visualizer/lib.js +++ b/pylabrobot/visualizer/lib.js @@ -3080,7 +3080,10 @@ class XArm extends Resource { fill: "rgba(64, 64, 64, 0.5)", stroke: "rgba(64, 64, 64, 0.3)", strokeWidth: 3, - listening: false, + // The solid frame occludes what is under it, so it captures clicks; its hit region + // follows the fill above, which excludes the punched hole, so resources seen through + // the hole stay clickable. (The reference line stays non-listening.) + listening: true, }) ); return g;