From 03630d7a9b4d7cce1065261e9e937b920b9d06cd Mon Sep 17 00:00:00 2001 From: Camillo Moschner Date: Sun, 19 Jul 2026 12:58:20 +0100 Subject: [PATCH 1/2] `STARBackend`: resolve installed X-arms into `XArmInformation` at setup Fuse the STAR's X-arm configuration information (arm widths, drive travel ranges, working envelopes) into one record resolved once at setup, mirroring iSWAPInformation and Head96Information. Adds XArmInformation / SingleXArmInformation, the x_arm_information property, _build_x_arm_information, and DriveConfiguration.is_present; the drive-range and working-envelope request methods now parse into typed dicts instead of returning the raw firmware string. Co-Authored-By: Claude Opus 4.8 --- .../backends/hamilton/STAR_backend.py | 139 +++++++++++++++++- .../backends/hamilton/STAR_chatterbox.py | 35 ++++- .../backends/hamilton/STAR_tests.py | 52 +++++++ 3 files changed, 219 insertions(+), 7 deletions(-) diff --git a/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py b/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py index f9d2ab5b192..6f70a1c9d41 100644 --- a/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py +++ b/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py @@ -1249,6 +1249,11 @@ class DriveConfiguration: imaging_channel_installed: bool = False robotic_channel_installed: bool = False + @property + def is_present(self) -> bool: + """Whether this X-drive carries any module, i.e. the X-arm exists.""" + return any(vars(self).values()) + @dataclass class MachineConfiguration: @@ -1372,6 +1377,56 @@ class ExtendedConfiguration: """Right arm minimal Y position [mm] (yx). Default: 6.0.""" +@dataclass(frozen=True, eq=False) +class XArmInformation: + """The machine's X-arm layout, resolved once at setup. + + The top-level per-machine X-arm record: one `SingleXArmInformation` per installed + arm, keyed by the rail it sits on. A STAR carries an arm on the `left` rail; an + optional second arm on the `right` rail makes `right` non-None (so `number_x_arms` + is 1 or 2). Built by `STARBackend._build_x_arm_information` from the machine + configuration and the X-drive queries, and immutable thereafter. This is the + reference frame the arm-mounted modules - pipetting channels, the 96-head, the + iSWAP - are positioned against. + """ + + left: Optional["SingleXArmInformation"] = None + right: Optional["SingleXArmInformation"] = None + + @property + def number_x_arms(self) -> int: + """Number of installed X-arms.""" + return sum(arm is not None for arm in (self.left, self.right)) + + +@dataclass(frozen=True, eq=False) +class SingleXArmInformation: + """One installed X-arm's configuration information, resolved at setup from firmware. + + Populated by `STARBackend._build_x_arm_information` from the machine configuration + (width) and the X-drive range and working-envelope queries. Immutable post-setup. + """ + + position: Literal["left", "right"] + """Which rail this arm is on.""" + + width: float + """Arm width (mm), from the machine configuration.""" + + model: str + """Arm variant, derived from `width` (wide arms span both rails, narrow arms one).""" + + reference_point: Literal["center", "right"] + """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.""" + + x_range: Tuple[float, float] + """Drive travel `(min, max)` in mm.""" + + workspace_range: Tuple[float, float] + """Reachable X workspace `(min, max)` in mm.""" + + @dataclass class PipChannelInformation: """Installed hardware information for a single pipetting channel (VW command).""" @@ -1664,6 +1719,8 @@ def __init__( self.left_side_panel_installed = left_side_panel_installed self._machine_conf: Optional[MachineConfiguration] = None + # X-arm configuration information (widths, models, ranges) resolved from firmware at setup. + self._x_arm_information: Optional[XArmInformation] = None self._iswap_parked: Optional[bool] = None self._num_channels: Optional[int] = None @@ -2199,6 +2256,8 @@ async def set_up_arm_modules(): # the core grippers. self._core_parked = True + await self._build_x_arm_information() + self._setup_done = True async def stop(self): @@ -6268,15 +6327,83 @@ async def request_right_x_arm_position(self) -> float: resp_dmm = await self.send_command(module="C0", command="QX", fmt="rx#####") return cast(float, resp_dmm["rx"]) / 10 - async def request_maximal_ranges_of_x_drives(self): - """Request maximal ranges of X drives""" + @property + def x_arm_information(self) -> XArmInformation: + """The machine's X-arm configuration information, resolved at setup (widths, models, ranges).""" + if self._x_arm_information is None: + raise RuntimeError("X-arm information not loaded; forgot to call `setup`?") + return self._x_arm_information + + @staticmethod + def _x_arm_model_and_reference(width: float) -> Tuple[str, Literal["center", "right"]]: + """Arm variant and reference point for a given arm width. + + Wide arms span both rails and reference their centre; narrow arms sit on a single + (right) rail and reference their right edge. + """ + if width > 300: + return "hamilton_legacy_star_dual_rail_arm", "center" + return "hamilton_legacy_star_single_right_rail_arm", "right" + + async def _build_x_arm_information(self) -> None: + """Resolve the machine's X-arm configuration information from firmware and cache it. + + Fuses the machine configuration (widths), the X-drive travel ranges, and the arm + working-envelope query into an `XArmInformation` - one `SingleXArmInformation` per + installed arm. Called by setup() once the configuration is loaded. + """ + ranges = await self.request_maximal_ranges_of_x_drives() + wraps = await self.request_working_envelopes_per_arm() + widths = { + "left": self.extended_conf.left_x_arm_width, + "right": self.extended_conf.right_x_arm_width, + } + drives = {"left": self.extended_conf.left_x_drive, "right": self.extended_conf.right_x_drive} + + def build(position: Literal["left", "right"]) -> Optional[SingleXArmInformation]: + if not drives[position].is_present: + return None + model, reference_point = self._x_arm_model_and_reference(widths[position]) + _wrap, workspace_range = wraps[position] + return SingleXArmInformation( + position=position, + width=widths[position], + model=model, + reference_point=reference_point, + x_range=ranges[position], + workspace_range=workspace_range, + ) + + self._x_arm_information = XArmInformation(left=build("left"), right=build("right")) + + async def request_maximal_ranges_of_x_drives(self) -> Dict[str, Tuple[float, float]]: + """Request the maximal travel range of each X drive. - return await self.send_command(module="C0", command="RU") + Returns: + The `(minimum, maximum)` X position in mm each drive can reach, keyed by side: + `{"left": (min, max), "right": (min, max)}`. + """ + resp = await self.send_command(module="C0", command="RU") + values = [int(v) / 10 for v in resp.split("ru")[-1].strip().split()] + left_min, left_max, right_min, right_max = values + return {"left": (left_min, left_max), "right": (right_min, right_max)} - async def request_present_wrap_size_of_installed_arms(self): - """Request present wrap size of installed arms""" + async def request_working_envelopes_per_arm( + self, + ) -> Dict[str, Tuple[float, Tuple[float, float]]]: + """Request the working envelope of each installed arm. - return await self.send_command(module="C0", command="UA") + Returns: + Per side, `(wrap_size, (workspace_min, workspace_max))` in mm, keyed by side. A + `wrap_size` of 0 means that arm is not installed. + """ + resp = await self.send_command(module="C0", command="UA") + values = [int(v) / 10 for v in resp.split("ua")[-1].strip().split()] + left_wrap, right_wrap, left_min, left_max, right_min, right_max = values + return { + "left": (left_wrap, (left_min, left_max)), + "right": (right_wrap, (right_min, right_max)), + } async def request_left_x_arm_last_collision_type(self): """Request left X-Arm last collision type (after error 27) diff --git a/pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py b/pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py index 844b4427191..0ea0cf0e8c3 100644 --- a/pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py +++ b/pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py @@ -3,7 +3,7 @@ import logging import warnings from contextlib import asynccontextmanager -from typing import Dict, List, Literal, Optional, Union +from typing import Dict, List, Literal, Optional, Tuple, Union from pylabrobot.io.validation_utils import LOG_LEVEL_IO from pylabrobot.liquid_handling.backends import LiquidHandlerBackend @@ -16,6 +16,7 @@ iSWAPInformation, ) from pylabrobot.resources.container import Container +from pylabrobot.resources.hamilton.hamilton_decks import HamiltonDeck from pylabrobot.resources.tip_tracker import does_tip_tracking from pylabrobot.resources.well import Well @@ -39,6 +40,11 @@ max_iswap_collision_free_position=600.0, ) +# Minimal left-drive X position of a dual-rail arm. Validated against real hardware; +# the single-rail minimum is not yet known (#822). Only the chatterbox needs this +# literal - a physical STAR reports its own value from the drive-range query. +_DUAL_RAIL_LEFT_X_MIN = 95.0 + # 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 @@ -190,6 +196,10 @@ async def setup( else: self._iswap_information = None + # Fuse the (mocked) configuration and X-drive replies into X-arm information, + # mirroring STARBackend.setup. + await self._build_x_arm_information() + async def stop(self): await LiquidHandlerBackend.stop(self) self._setup_done = False @@ -226,6 +236,29 @@ async def request_extended_configuration(self) -> ExtendedConfiguration: assert self._extended_conf is not None return self._extended_conf + def _simulated_x_reach_max(self) -> float: + """Rightmost reachable X (mm) in simulation, from the deck's reachable range.""" + deck = self._deck + if isinstance(deck, HamiltonDeck): + return deck.rails_to_location(deck.num_rails).x + if deck is not None: + return deck.get_size_x() + return 1338.0 # nominal STAR reach + + async def request_maximal_ranges_of_x_drives(self) -> Dict[str, Tuple[float, float]]: + x_range = (_DUAL_RAIL_LEFT_X_MIN, self._simulated_x_reach_max()) + return {"left": x_range, "right": x_range} + + async def request_working_envelopes_per_arm( + self, + ) -> Dict[str, Tuple[float, Tuple[float, float]]]: + workspace = (-323.2, self._simulated_x_reach_max()) + left = (595.2, workspace) # wrap, workspace + # A wrap of 0 signals "arm not installed" (per the base method's contract), so an + # absent right drive reports 0 rather than mirroring the left arm. + right = (595.2, workspace) if self.extended_conf.right_x_drive.is_present else (0.0, (0.0, 0.0)) + return {"left": left, "right": right} + # # # # # # # # 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 69505002ece..9ea0887083a 100644 --- a/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py +++ b/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py @@ -2583,3 +2583,55 @@ async def test_duplicate_channels_serialize_measurements(self): self.assertEqual(len(result), 2) self.assertAlmostEqual(result[0], 0 - well_a.get_absolute_location("c", "c", "cavity_bottom").z) self.assertAlmostEqual(result[1], 0 - well_b.get_absolute_location("c", "c", "cavity_bottom").z) + + +class TestXArmInformation(unittest.IsolatedAsyncioTestCase): + """setup() fuses QM/RU/UA firmware into XArmInformation.""" + + async def asyncSetUp(self): + self.lh = LiquidHandler(STARChatterboxBackend(), deck=STARLetDeck()) + await self.lh.setup() + + async def test_single_left_arm(self): + info = self.lh.backend.x_arm_information + self.assertEqual(info.number_x_arms, 1) + self.assertIsNone(info.right) + + async def test_left_arm_fields(self): + left = self.lh.backend.x_arm_information.left + assert left is not None + self.assertEqual(left.position, "left") + self.assertEqual(left.reference_point, "center") # QM width -> model_and_reference -> slot + # x_range comes from RU, workspace_range from UA: fusion routes the two firmware + # sources into distinct slots (the part not covered by the parse/branch unit tests). + self.assertEqual(left.x_range[0], 95.0) + self.assertEqual(left.workspace_range[0], -323.2) + + def test_model_and_reference_by_width(self): + self.assertEqual( + STARBackend._x_arm_model_and_reference(370.0), + ("hamilton_legacy_star_dual_rail_arm", "center"), + ) + # 246.0 is an illustrative <300 width; no single-rail dump exists yet to measure one. + self.assertEqual( + STARBackend._x_arm_model_and_reference(246.0), + ("hamilton_legacy_star_single_right_rail_arm", "right"), + ) + + +class TestXArmRangeQueries(unittest.IsolatedAsyncioTestCase): + """RU/UA parse the firmware replies observed on real machines.""" + + def setUp(self): + self.star = STARBackend() + self.star.send_command = unittest.mock.AsyncMock() + + async def test_maximal_ranges_of_x_drives(self): + self.star.send_command.return_value = "C0RUid0002er00/00ru00950 13402 30000 30000" + ranges = await self.star.request_maximal_ranges_of_x_drives() + self.assertEqual(ranges, {"left": (95.0, 1340.2), "right": (3000.0, 3000.0)}) + + async def test_working_envelopes_per_arm(self): + self.star.send_command.return_value = "C0UAid0001er00/00ua5952 0000 -03232 +15172 +30000 +30000" + wraps = await self.star.request_working_envelopes_per_arm() + self.assertEqual(wraps, {"left": (595.2, (-323.2, 1517.2)), "right": (0.0, (3000.0, 3000.0))}) From 3adf75a52fd7ec4a16af0b86789b8ebdcd0b807b Mon Sep 17 00:00:00 2001 From: Rick Wierenga Date: Mon, 20 Jul 2026 23:25:40 -0700 Subject: [PATCH 2/2] `STARBackend`: fold X-arm geometry into `DriveConfiguration` Remove `XArmInformation`/`SingleXArmInformation` and resolve each X-drive's geometry (width, travel range, workspace range) directly onto its `DriveConfiguration` inside `request_extended_configuration`, alongside the module bits. `model` and `reference_point` become properties derived from `width`. `right_x_drive` is now `Optional`, `None` when no second arm is installed, so arm presence is a plain `is None` check. The separate `_resolve_x_arm_geometry` setup step is gone. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../backends/hamilton/STAR_backend.py | 167 ++++++------------ .../backends/hamilton/STAR_chatterbox.py | 42 ++++- .../backends/hamilton/STAR_tests.py | 35 ++-- 3 files changed, 101 insertions(+), 143 deletions(-) diff --git a/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py b/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py index 6f70a1c9d41..d29f7f20ce6 100644 --- a/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py +++ b/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py @@ -1233,9 +1233,15 @@ def _dispensing_mode_for_op(empty: bool, jet: bool, blow_out: bool) -> int: @dataclass class DriveConfiguration: - """Configuration for an X drive (left or right). + """Configuration and geometry for an X drive (left or right). + + The installed-module bits combine byte 1 (xl/xr) and byte 2 (xn/xo). The arm + geometry - width, travel range, workspace range - comes from the X-drive range (RU) + and working-envelope (UA) queries, so it is None on a drive built from the module + bits alone (e.g. a simulated configuration) and populated when + `request_extended_configuration` builds the drive. `model` and `reference_point` + follow from `width`. - Combines byte 1 (xl/xr) and byte 2 (xn/xo) into a single object. Note: the installed modules on left and right drives must be different. """ @@ -1249,10 +1255,27 @@ class DriveConfiguration: imaging_channel_installed: bool = False robotic_channel_installed: bool = False + width: Optional[float] = None + """Arm width (mm), from the machine configuration.""" + x_range: Optional[Tuple[float, float]] = None + """Drive travel `(min, max)` in mm.""" + workspace_range: Optional[Tuple[float, float]] = None + """Reachable X workspace `(min, max)` in mm.""" + + @property + def model(self) -> str: + """Arm variant derived from `width`: wide arms span both rails, narrow arms one.""" + assert self.width is not None, "arm geometry not resolved" + if self.width > 300: + return "hamilton_legacy_star_dual_rail_arm" + return "hamilton_legacy_star_single_right_rail_arm" + @property - def is_present(self) -> bool: - """Whether this X-drive carries any module, i.e. the X-arm exists.""" - return any(vars(self).values()) + def reference_point(self) -> Literal["center", "right"]: + """Where along the arm's width the tracked X refers to: the arm center for a + dual-rail arm, the right edge for a single-rail arm.""" + assert self.width is not None, "arm geometry not resolved" + return "center" if self.width > 300 else "right" @dataclass @@ -1349,8 +1372,8 @@ class ExtendedConfiguration: """Tip waste X-position [mm] (xw). Default: 1340.0.""" left_x_drive: DriveConfiguration = field(default_factory=DriveConfiguration) """Left X drive configuration (xl + xn).""" - right_x_drive: DriveConfiguration = field(default_factory=DriveConfiguration) - """Right X drive configuration (xr + xo).""" + right_x_drive: Optional[DriveConfiguration] = None + """Right X drive configuration (xr + xo), or None when no right arm is installed.""" min_iswap_collision_free_position: float = 350.0 """Minimal iSWAP collision free position for direct X access [mm] (xm). Default: 350.0.""" max_iswap_collision_free_position: float = 1140.0 @@ -1377,56 +1400,6 @@ class ExtendedConfiguration: """Right arm minimal Y position [mm] (yx). Default: 6.0.""" -@dataclass(frozen=True, eq=False) -class XArmInformation: - """The machine's X-arm layout, resolved once at setup. - - The top-level per-machine X-arm record: one `SingleXArmInformation` per installed - arm, keyed by the rail it sits on. A STAR carries an arm on the `left` rail; an - optional second arm on the `right` rail makes `right` non-None (so `number_x_arms` - is 1 or 2). Built by `STARBackend._build_x_arm_information` from the machine - configuration and the X-drive queries, and immutable thereafter. This is the - reference frame the arm-mounted modules - pipetting channels, the 96-head, the - iSWAP - are positioned against. - """ - - left: Optional["SingleXArmInformation"] = None - right: Optional["SingleXArmInformation"] = None - - @property - def number_x_arms(self) -> int: - """Number of installed X-arms.""" - return sum(arm is not None for arm in (self.left, self.right)) - - -@dataclass(frozen=True, eq=False) -class SingleXArmInformation: - """One installed X-arm's configuration information, resolved at setup from firmware. - - Populated by `STARBackend._build_x_arm_information` from the machine configuration - (width) and the X-drive range and working-envelope queries. Immutable post-setup. - """ - - position: Literal["left", "right"] - """Which rail this arm is on.""" - - width: float - """Arm width (mm), from the machine configuration.""" - - model: str - """Arm variant, derived from `width` (wide arms span both rails, narrow arms one).""" - - reference_point: Literal["center", "right"] - """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.""" - - x_range: Tuple[float, float] - """Drive travel `(min, max)` in mm.""" - - workspace_range: Tuple[float, float] - """Reachable X workspace `(min, max)` in mm.""" - - @dataclass class PipChannelInformation: """Installed hardware information for a single pipetting channel (VW command).""" @@ -1719,8 +1692,6 @@ def __init__( self.left_side_panel_installed = left_side_panel_installed self._machine_conf: Optional[MachineConfiguration] = None - # X-arm configuration information (widths, models, ranges) resolved from firmware at setup. - self._x_arm_information: Optional[XArmInformation] = None self._iswap_parked: Optional[bool] = None self._num_channels: Optional[int] = None @@ -2256,8 +2227,6 @@ async def set_up_arm_modules(): # the core grippers. self._core_parked = True - await self._build_x_arm_information() - self._setup_done = True async def stop(self): @@ -6091,10 +6060,13 @@ async def request_machine_configuration(self) -> MachineConfiguration: ) async def request_extended_configuration(self) -> ExtendedConfiguration: - """Request extended configuration (QM command). + """Request extended configuration (QM command) with X-arm geometry resolved. Returns the full instrument configuration matching the AK - (Set Instrument Configuration) [SFCO.0026] parameter set. + (Set Instrument Configuration) [SFCO.0026] parameter set. Each installed X-drive's + geometry (width, travel range, workspace range) is resolved from the X-drive range + (RU) and working-envelope (UA) queries; `right_x_drive` is None when no second arm + is installed. """ resp = await self.send_command( @@ -6104,7 +6076,15 @@ async def request_extended_configuration(self) -> ExtendedConfiguration: + "ys###kl###km###ym####yu####yx####", ) - def _parse_drive(byte1: int, byte2: int) -> DriveConfiguration: + ranges = await self.request_maximal_ranges_of_x_drives() + wraps = await self.request_working_envelopes_per_arm() + + def _build_drive( + byte1: int, byte2: int, side: Literal["left", "right"], width: float + ) -> Optional[DriveConfiguration]: + wrap, workspace_range = wraps[side] + if wrap == 0: # arm not installed + return None return DriveConfiguration( pip_installed=bool(byte1 & (1 << 0)), iswap_installed=bool(byte1 & (1 << 1)), @@ -6115,8 +6095,14 @@ def _parse_drive(byte1: int, byte2: int) -> DriveConfiguration: tube_gripper_installed=bool(byte1 & (1 << 6)), imaging_channel_installed=bool(byte1 & (1 << 7)), robotic_channel_installed=bool(byte2 & (1 << 0)), + width=width, + x_range=ranges[side], + workspace_range=workspace_range, ) + left_x_drive = _build_drive(resp["xl"], resp["xn"], "left", resp["xu"] / 10) + assert left_x_drive is not None, "STAR must have a left X-arm" + ka = resp["ka"] return ExtendedConfiguration( left_x_drive_large=bool(ka & (1 << 0)), @@ -6146,8 +6132,8 @@ def _parse_drive(byte1: int, byte2: int) -> DriveConfiguration: instrument_size_slots=resp["xt"], auto_load_size_slots=resp["xa"], tip_waste_x_position=resp["xw"] / 10, - left_x_drive=_parse_drive(resp["xl"], resp["xn"]), - right_x_drive=_parse_drive(resp["xr"], resp["xo"]), + left_x_drive=left_x_drive, + right_x_drive=_build_drive(resp["xr"], resp["xo"], "right", resp["xv"] / 10), min_iswap_collision_free_position=resp["xm"] / 10, max_iswap_collision_free_position=resp["xx"] / 10, left_x_arm_width=resp["xu"] / 10, @@ -6327,55 +6313,6 @@ async def request_right_x_arm_position(self) -> float: resp_dmm = await self.send_command(module="C0", command="QX", fmt="rx#####") return cast(float, resp_dmm["rx"]) / 10 - @property - def x_arm_information(self) -> XArmInformation: - """The machine's X-arm configuration information, resolved at setup (widths, models, ranges).""" - if self._x_arm_information is None: - raise RuntimeError("X-arm information not loaded; forgot to call `setup`?") - return self._x_arm_information - - @staticmethod - def _x_arm_model_and_reference(width: float) -> Tuple[str, Literal["center", "right"]]: - """Arm variant and reference point for a given arm width. - - Wide arms span both rails and reference their centre; narrow arms sit on a single - (right) rail and reference their right edge. - """ - if width > 300: - return "hamilton_legacy_star_dual_rail_arm", "center" - return "hamilton_legacy_star_single_right_rail_arm", "right" - - async def _build_x_arm_information(self) -> None: - """Resolve the machine's X-arm configuration information from firmware and cache it. - - Fuses the machine configuration (widths), the X-drive travel ranges, and the arm - working-envelope query into an `XArmInformation` - one `SingleXArmInformation` per - installed arm. Called by setup() once the configuration is loaded. - """ - ranges = await self.request_maximal_ranges_of_x_drives() - wraps = await self.request_working_envelopes_per_arm() - widths = { - "left": self.extended_conf.left_x_arm_width, - "right": self.extended_conf.right_x_arm_width, - } - drives = {"left": self.extended_conf.left_x_drive, "right": self.extended_conf.right_x_drive} - - def build(position: Literal["left", "right"]) -> Optional[SingleXArmInformation]: - if not drives[position].is_present: - return None - model, reference_point = self._x_arm_model_and_reference(widths[position]) - _wrap, workspace_range = wraps[position] - return SingleXArmInformation( - position=position, - width=widths[position], - model=model, - reference_point=reference_point, - x_range=ranges[position], - workspace_range=workspace_range, - ) - - self._x_arm_information = XArmInformation(left=build("left"), right=build("right")) - 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 0ea0cf0e8c3..3868b46fc3f 100644 --- a/pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py +++ b/pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py @@ -3,6 +3,7 @@ import logging import warnings from contextlib import asynccontextmanager +from dataclasses import replace from typing import Dict, List, Literal, Optional, Tuple, Union from pylabrobot.io.validation_utils import LOG_LEVEL_IO @@ -196,10 +197,6 @@ async def setup( else: self._iswap_information = None - # Fuse the (mocked) configuration and X-drive replies into X-arm information, - # mirroring STARBackend.setup. - await self._build_x_arm_information() - async def stop(self): await LiquidHandlerBackend.stop(self) self._setup_done = False @@ -233,8 +230,35 @@ async def request_machine_configuration(self) -> MachineConfiguration: return self._machine_configuration async def request_extended_configuration(self) -> ExtendedConfiguration: - assert self._extended_conf is not None - return self._extended_conf + """Return the configured extended configuration with X-arm geometry resolved. + + Mirrors STARBackend.request_extended_configuration: fills each installed drive's + geometry from the mocked X-drive range/envelope replies. A right drive that was not + configured (None) stays None. + """ + conf = self._extended_conf + assert conf is not None + ranges = await self.request_maximal_ranges_of_x_drives() + wraps = await self.request_working_envelopes_per_arm() + + def _with_geometry( + drive: Optional[DriveConfiguration], side: str, width: float + ) -> Optional[DriveConfiguration]: + if drive is None: + return None + wrap, workspace_range = wraps[side] + if wrap == 0: # arm not installed + return None + return replace(drive, width=width, x_range=ranges[side], workspace_range=workspace_range) + + left_x_drive = _with_geometry(conf.left_x_drive, "left", conf.left_x_arm_width) + assert left_x_drive is not None, "STAR must have a left X-arm" + + return replace( + conf, + left_x_drive=left_x_drive, + right_x_drive=_with_geometry(conf.right_x_drive, "right", conf.right_x_arm_width), + ) def _simulated_x_reach_max(self) -> float: """Rightmost reachable X (mm) in simulation, from the deck's reachable range.""" @@ -254,9 +278,9 @@ async def request_working_envelopes_per_arm( ) -> Dict[str, Tuple[float, Tuple[float, float]]]: workspace = (-323.2, self._simulated_x_reach_max()) left = (595.2, workspace) # wrap, workspace - # A wrap of 0 signals "arm not installed" (per the base method's contract), so an - # absent right drive reports 0 rather than mirroring the left arm. - right = (595.2, workspace) if self.extended_conf.right_x_drive.is_present else (0.0, (0.0, 0.0)) + # A wrap of 0 signals "arm not installed" (per the base method's contract). + right_installed = self.extended_conf.right_x_drive is not None + right = (595.2, workspace) if right_installed else (0.0, (0.0, 0.0)) return {"left": left, "right": right} # # # # # # # # 1_000 uL Channel: Basic Commands # # # # # # # # diff --git a/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py b/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py index 9ea0887083a..de87a7b6d39 100644 --- a/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py +++ b/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py @@ -36,6 +36,7 @@ from .STAR_backend import ( CommandSyntaxError, + DriveConfiguration, HamiltonNoTipError, HardwareError, Head96Information, @@ -2585,38 +2586,34 @@ async def test_duplicate_channels_serialize_measurements(self): self.assertAlmostEqual(result[1], 0 - well_b.get_absolute_location("c", "c", "cavity_bottom").z) -class TestXArmInformation(unittest.IsolatedAsyncioTestCase): - """setup() fuses QM/RU/UA firmware into XArmInformation.""" +class TestXArmGeometry(unittest.IsolatedAsyncioTestCase): + """setup() resolves QM/RU/UA firmware onto the X-drive DriveConfigurations.""" async def asyncSetUp(self): self.lh = LiquidHandler(STARChatterboxBackend(), deck=STARLetDeck()) await self.lh.setup() async def test_single_left_arm(self): - info = self.lh.backend.x_arm_information - self.assertEqual(info.number_x_arms, 1) - self.assertIsNone(info.right) + self.assertIsNotNone(self.lh.backend.extended_conf.left_x_drive.workspace_range) + self.assertIsNone(self.lh.backend.extended_conf.right_x_drive) async def test_left_arm_fields(self): - left = self.lh.backend.x_arm_information.left - assert left is not None - self.assertEqual(left.position, "left") - self.assertEqual(left.reference_point, "center") # QM width -> model_and_reference -> slot - # x_range comes from RU, workspace_range from UA: fusion routes the two firmware - # sources into distinct slots (the part not covered by the parse/branch unit tests). + left = self.lh.backend.extended_conf.left_x_drive + self.assertEqual(left.reference_point, "center") # QM width -> center rail + # x_range comes from RU, workspace_range from UA: request_extended_configuration routes + # the two firmware sources into distinct slots (not covered by the parse/branch tests). + assert left.x_range is not None and left.workspace_range is not None self.assertEqual(left.x_range[0], 95.0) self.assertEqual(left.workspace_range[0], -323.2) def test_model_and_reference_by_width(self): - self.assertEqual( - STARBackend._x_arm_model_and_reference(370.0), - ("hamilton_legacy_star_dual_rail_arm", "center"), - ) + dual = DriveConfiguration(width=370.0) + self.assertEqual(dual.model, "hamilton_legacy_star_dual_rail_arm") + self.assertEqual(dual.reference_point, "center") # 246.0 is an illustrative <300 width; no single-rail dump exists yet to measure one. - self.assertEqual( - STARBackend._x_arm_model_and_reference(246.0), - ("hamilton_legacy_star_single_right_rail_arm", "right"), - ) + single = DriveConfiguration(width=246.0) + self.assertEqual(single.model, "hamilton_legacy_star_single_right_rail_arm") + self.assertEqual(single.reference_point, "right") class TestXArmRangeQueries(unittest.IsolatedAsyncioTestCase):