diff --git a/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py b/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py index f9d2ab5b192..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,6 +1255,28 @@ 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 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 class MachineConfiguration: @@ -1344,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 @@ -6032,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( @@ -6045,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)), @@ -6056,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)), @@ -6087,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, @@ -6268,15 +6313,34 @@ 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""" + 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..3868b46fc3f 100644 --- a/pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py +++ b/pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py @@ -3,7 +3,8 @@ import logging import warnings from contextlib import asynccontextmanager -from typing import Dict, List, Literal, Optional, Union +from dataclasses import replace +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 +17,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 +41,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 @@ -223,8 +230,58 @@ 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.""" + 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). + 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 69505002ece..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, @@ -2583,3 +2584,51 @@ 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 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): + 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.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): + 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. + 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): + """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))})