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 x_arm_information), 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 <noreply@anthropic.com>

Author: BioCam

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.

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]]: