diff --git a/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py b/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py index 2f11245fb01..89ae6d4dbf5 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,84 @@ 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") + 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``). @@ -6208,11 +6284,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 +6312,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 +6436,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..16f9091654e 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,36 @@ 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: + # 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.") + 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 + + 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..68afebef662 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", 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. + 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", 800.0, 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, 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") + 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_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 43ebc1f9cc4..bb21755652c 100644 --- a/pylabrobot/resources/hamilton/hamilton_decks.py +++ b/pylabrobot/resources/hamilton/hamilton_decks.py @@ -12,12 +12,20 @@ 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. +# 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 STARLET_SIZE_X = 1005 STARLET_SIZE_Y = 653.5 @@ -188,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}'." ) @@ -543,6 +559,47 @@ 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, + 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 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. + """ + 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, + ) + # 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: 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..3cb8be70770 100644 --- a/pylabrobot/visualizer/lib.js +++ b/pylabrobot/visualizer/lib.js @@ -3017,6 +3017,123 @@ 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, + // 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; + } + + 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 +3291,8 @@ function classForResourceType(type, category) { case "container": case "trough": return Container; + case "x_arm": + return XArm; default: return Resource; } 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":