diff --git a/docs/user_guide/hamilton/star/index.md b/docs/user_guide/hamilton/star/index.md index bbc2e225f80..86a959c4535 100644 --- a/docs/user_guide/hamilton/star/index.md +++ b/docs/user_guide/hamilton/star/index.md @@ -10,5 +10,6 @@ :maxdepth: 1 debug +simulation hardware/index ``` diff --git a/docs/user_guide/hamilton/star/simulation.md b/docs/user_guide/hamilton/star/simulation.md new file mode 100644 index 00000000000..8748d29bcbc --- /dev/null +++ b/docs/user_guide/hamilton/star/simulation.md @@ -0,0 +1,71 @@ +# STAR chatterbox simulation + +`STARChatterboxBackend` is a device-free STAR backend. Constructing it with no extra arguments uses +the unconfigured query defaults, so existing scripts and tests do not need to change. + +```python +from pylabrobot.legacy.liquid_handling import LiquidHandler +from pylabrobot.legacy.liquid_handling.backends.hamilton.STAR_chatterbox import ( + STARChatterboxBackend, + STARChatterboxState, +) +from pylabrobot.resources.hamilton import STARLetDeck + +lh = LiquidHandler(backend=STARChatterboxBackend(), deck=STARLetDeck()) +``` + +## Configurable query state + +Pass a `STARChatterboxState` to configure simulated device/environment readings for deterministic +protocol or recovery tests: + +```python +backend = STARChatterboxBackend( + chatterbox_state=STARChatterboxState( + iswap_initialization_status=False, + channel_z_positions=[100.0] * 8, + dispensing_drive_positions=[0.0] * 8, + ) +) +``` + +Configurable readings: + +- `iswap_initialization_status` — `request_iswap_initialization_status()` (default `True`) +- `channel_z_positions` — `request_z_pos_channel_n()` (default `285.0` mm per channel) +- `dispensing_drive_positions` — `channel_dispensing_drive_request_position()` (default `0.0`) + +These are simulated query replies, not physical verification. Motion commands do not update them. + +To change readings after construction, assign a new `STARChatterboxState`. Assignment replaces the +whole object: omitted vectors refill the unconfigured defaults. The setter copies the vectors and +checks their lengths against `num_channels`. Bound channel vectors are tuples; in-place mutation is +not supported. `setup()` does not reset this state. + +```python +backend.chatterbox_state = STARChatterboxState( + iswap_initialization_status=False, + channel_z_positions=[90.0] * backend.num_channels, + dispensing_drive_positions=[3.0] * backend.num_channels, +) +``` + +`channel_dispensing_drive_request_position(channel_idx, simulated_value=...)` still accepts a +per-call override. Omit the argument to use instance state. Pass `simulated_value=0.0` when the +override should be zero; that is distinct from omitting the argument. Overrides do not write back +to instance state. + +## Owned elsewhere (not in `STARChatterboxState`) + +- Tip presence, 96-head tip presence, and tip length: `TipTracker` +- Last LLD heights: latched by simulated LLD (`request_pip_height_last_lld`) +- Installed modules and factory geometry: `MachineConfiguration`, `ExtendedConfiguration`, + `iSWAPInformation`, and `channels_minimum_y_spacing` +- iSWAP parked / CoRe parked: existing latched backend flags + +## Not a fault scenario + +This is static configurable query state. It does not script response sequences, inject timeouts or +failures, model partial physical success, or replace the resource model. Carrier-presence and CoRe +resource-existence commands are not chatterbox overrides; they still follow the firmware command +path. diff --git a/pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_chatterbox.py b/pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_chatterbox.py index 59d37fccec7..997ce6194c8 100644 --- a/pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_chatterbox.py +++ b/pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_chatterbox.py @@ -4,7 +4,7 @@ import warnings from contextlib import asynccontextmanager from dataclasses import replace -from typing import Dict, List, Literal, Optional, Tuple, Union +from typing import Dict, List, Literal, Optional, Sequence, Tuple, Union from pylabrobot.io.validation_utils import LOG_LEVEL_IO from pylabrobot.legacy.liquid_handling.backends import LiquidHandlerBackend @@ -40,6 +40,136 @@ max_iswap_collision_free_position=600.0, ) +# Unconfigured chatterbox query values for channel Z (mm) and dispensing-drive position. +_DEFAULT_SIMULATED_CHANNEL_Z = 285.0 +_DEFAULT_SIMULATED_DISPENSING_DRIVE_POSITION = 0.0 + + +def _copy_channel_vector( + values: Optional[Sequence[float]], + num_channels: int, + name: str, + default: float, +) -> Tuple[float, ...]: + """Copy a per-channel vector, filling the unconfigured default when omitted.""" + if values is None or len(values) == 0: + return (default,) * num_channels + copied = tuple(float(v) for v in values) + if len(copied) != num_channels: + raise ValueError(f"{name} has {len(copied)} entries, expected {num_channels}.") + return copied + + +class STARChatterboxState: + """Simulated STAR query readings for one chatterbox instance. + + This is not physical verification, a motion model, or a fault-injection + framework. Values are the replies returned by the corresponding + ``STARChatterboxBackend`` query methods. + + Channel vectors are copied into tuples at construction. Change readings by + constructing a ``STARChatterboxState`` and assigning ``backend.chatterbox_state``; + that assignment replaces the whole object, re-validates lengths against + ``num_channels``, and refills omitted vectors with unconfigured defaults. + In-place mutation of bound vectors is not supported. + + Tip presence, 96-head tip presence, and tip length stay on ``TipTracker``. + Last-LLD heights stay latched by simulated LLD. Machine geometry stays on + ``MachineConfiguration`` / ``ExtendedConfiguration`` / ``iSWAPInformation``. + + ``STARChatterboxState`` is in-process test configuration. It is not written + by ``serialize()`` or restored by ``deserialize()``. + + Unconfigured defaults: iSWAP reported initialized, channel Z 285 mm, + dispensing-drive position 0. + + Example: + backend = STARChatterboxBackend( + chatterbox_state=STARChatterboxState( + iswap_initialization_status=False, + channel_z_positions=[100.0] * 8, + ) + ) + """ + + def __init__( + self, + iswap_initialization_status: bool = True, + channel_z_positions: Optional[Sequence[float]] = None, + dispensing_drive_positions: Optional[Sequence[float]] = None, + ): + """Store copies of the given simulated query readings. + + Args: + iswap_initialization_status: Value for ``request_iswap_initialization_status``. + channel_z_positions: Per-channel Z query values (mm). ``None`` or empty + means fill the unconfigured default when the state is bound to a backend. + dispensing_drive_positions: Per-channel dispensing-drive query values. + ``None`` or empty means fill the unconfigured default when bound. + """ + self._iswap_initialization_status = bool(iswap_initialization_status) + self._channel_z_positions = ( + tuple(float(v) for v in channel_z_positions) if channel_z_positions is not None else () + ) + self._dispensing_drive_positions = ( + tuple(float(v) for v in dispensing_drive_positions) + if dispensing_drive_positions is not None + else () + ) + + @property + def iswap_initialization_status(self) -> bool: + """Simulated iSWAP initialization query value.""" + return self._iswap_initialization_status + + @property + def channel_z_positions(self) -> Tuple[float, ...]: + """Simulated per-channel Z query values (mm).""" + return self._channel_z_positions + + @property + def dispensing_drive_positions(self) -> Tuple[float, ...]: + """Simulated per-channel dispensing-drive query values.""" + return self._dispensing_drive_positions + + def resolved(self, num_channels: int) -> "STARChatterboxState": + """Return a copy filled and checked against ``num_channels``.""" + if num_channels < 0: + raise ValueError(f"num_channels must be >= 0, got {num_channels}.") + return STARChatterboxState( + iswap_initialization_status=self.iswap_initialization_status, + channel_z_positions=_copy_channel_vector( + self.channel_z_positions, + num_channels, + "channel_z_positions", + _DEFAULT_SIMULATED_CHANNEL_Z, + ), + dispensing_drive_positions=_copy_channel_vector( + self.dispensing_drive_positions, + num_channels, + "dispensing_drive_positions", + _DEFAULT_SIMULATED_DISPENSING_DRIVE_POSITION, + ), + ) + + def __eq__(self, other: object) -> bool: + if not isinstance(other, STARChatterboxState): + return NotImplemented + return ( + self.iswap_initialization_status == other.iswap_initialization_status + and self.channel_z_positions == other.channel_z_positions + and self.dispensing_drive_positions == other.dispensing_drive_positions + ) + + def __repr__(self) -> str: + return ( + "STARChatterboxState(" + f"iswap_initialization_status={self.iswap_initialization_status!r}, " + f"channel_z_positions={self.channel_z_positions!r}, " + f"dispensing_drive_positions={self.dispensing_drive_positions!r})" + ) + + # 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. @@ -71,7 +201,26 @@ class STARChatterboxBackend(STARBackend): - """Chatterbox backend for 'STAR'""" + """Device-free STAR backend with optional per-instance query state. + + Constructing ``STARChatterboxBackend()`` with no extra arguments uses the + unconfigured query defaults. Pass ``chatterbox_state=STARChatterboxState(...)`` + to configure simulated device/environment readings for deterministic tests. + + This is not a physics simulator. Channel Z and dispensing-drive queries do + not follow motion commands; assign ``backend.chatterbox_state`` when a + protocol needs a specific reading. Tip sensors stay on the tip tracker. + + Example: + from pylabrobot.legacy.liquid_handling.backends.hamilton.STAR_chatterbox import ( + STARChatterboxBackend, + STARChatterboxState, + ) + + backend = STARChatterboxBackend( + chatterbox_state=STARChatterboxState(iswap_initialization_status=False) + ) + """ def __init__( self, @@ -80,6 +229,7 @@ def __init__( extended_configuration: ExtendedConfiguration = _DEFAULT_EXTENDED_CONFIGURATION, iswap_information: Optional[iSWAPInformation] = None, channels_minimum_y_spacing: Optional[List[float]] = None, + chatterbox_state: Optional[STARChatterboxState] = None, # deprecated parameters core96_head_installed: Optional[bool] = None, iswap_installed: Optional[bool] = None, @@ -96,6 +246,10 @@ def __init__( when the extended configuration reports iSWAP as installed. channels_minimum_y_spacing: Per-channel minimum Y spacing in mm. If None, defaults to `extended_configuration.min_raster_pitch_pip_channels` for all channels. + chatterbox_state: Optional simulated query readings. None uses the + unconfigured defaults (initialized iSWAP, channel Z 285 mm, + dispensing-drive position 0). Bound state is a copy; replace + ``backend.chatterbox_state`` to change readings after construction. core96_head_installed: Deprecated. Set `extended_configuration.left_x_drive .core_96_head_installed` instead. iswap_installed: Deprecated. Set `extended_configuration.left_x_drive @@ -108,6 +262,15 @@ def __init__( # Absolute heights latched by simulated LLD, one per channel. See # `request_pip_height_last_lld`; `_run_lld_on_channel_batch` writes them. self._last_lld_absolute_heights = [0.0] * num_channels + if chatterbox_state is None: + bound_state = STARChatterboxState() + elif isinstance(chatterbox_state, STARChatterboxState): + bound_state = chatterbox_state + else: + raise TypeError( + f"chatterbox_state must be STARChatterboxState, got {type(chatterbox_state).__name__}." + ) + self._chatterbox_state = bound_state.resolved(num_channels) if core96_head_installed is not None or iswap_installed is not None: extended_configuration = copy.deepcopy(extended_configuration) @@ -145,6 +308,34 @@ def __init__( extended_configuration.min_raster_pitch_pip_channels ] * num_channels + @property + def chatterbox_state(self) -> STARChatterboxState: + """Simulated query readings for this instance. + + Vectors are tuples. Assign a ``STARChatterboxState`` to replace the whole + object; omitted vectors refill unconfigured defaults. The setter copies and + validates lengths against ``num_channels``. ``setup()`` does not reset this + state. + """ + return self._chatterbox_state + + @chatterbox_state.setter + def chatterbox_state(self, state: STARChatterboxState) -> None: + if not isinstance(state, STARChatterboxState): + raise TypeError(f"chatterbox_state must be STARChatterboxState, got {type(state).__name__}.") + self._chatterbox_state = state.resolved(self.num_channels) + + def _require_channel_index(self, channel_idx: int, name: str = "channel") -> None: + """Raise if ``channel_idx`` is outside ``[0, num_channels)``.""" + if not 0 <= channel_idx < self.num_channels: + raise ValueError(f"{name} must be between 0 and {self.num_channels - 1}, got {channel_idx}.") + + def _channel_state_vector(self, values: Sequence[float], name: str) -> Sequence[float]: + """Return ``values`` after checking it has one entry per channel.""" + if len(values) != self.num_channels: + raise ValueError(f"{name} has {len(values)} entries, expected {self.num_channels}.") + return values + async def setup( self, skip_instrument_initialization=False, @@ -300,22 +491,37 @@ async def request_tip_presence(self) -> List[Optional[bool]]: return [self.head[ch].has_tip for ch in range(self.num_channels)] async def request_z_pos_channel_n(self, channel: int) -> float: - return 285.0 + """Return the configured simulated channel Z query value. + + This is not a measured stop-disk or tip-bottom position. Motion commands + do not update it; assign ``backend.chatterbox_state`` when a test needs a + specific reading. + """ + self._require_channel_index(channel, "channel") + positions = self._channel_state_vector( + self._chatterbox_state.channel_z_positions, "channel_z_positions" + ) + return positions[channel] async def channel_dispensing_drive_request_position( - self, channel_idx: int, simulated_value: float = 0.0 + self, channel_idx: int, simulated_value: Optional[float] = None ) -> float: - """Override to return mock dispensing drive position. + """Return the simulated dispensing-drive position for ``channel_idx``. - This method is called when the system needs to know the current position - of a channel's dispensing drive (e.g., before emptying tips). - - Returns a mock position with a default value of 0.0 for all channels. + Args: + channel_idx: 0-based channel index. + simulated_value: Optional per-call override. ``None`` (the default) uses + the instance ``chatterbox_state.dispensing_drive_positions`` value + (0.0 when unconfigured). An explicit ``0.0`` is an override, not an + omitted argument, and does not write back to instance state. """ - if not (0 <= channel_idx < self.num_channels): - raise ValueError(f"channel_idx must be between 0 and {self.num_channels - 1}") - - return simulated_value + self._require_channel_index(channel_idx, "channel_idx") + if simulated_value is not None: + return simulated_value + positions = self._channel_state_vector( + self._chatterbox_state.dispensing_drive_positions, "dispensing_drive_positions" + ) + return positions[channel_idx] async def channel_request_y_minimum_spacing(self, channel_idx: int) -> float: """Return mock minimum Y spacing for the given channel. @@ -420,8 +626,12 @@ async def head96_request_tip_presence(self) -> int: # # # # # # # # Extension: iSWAP # # # # # # # # async def request_iswap_initialization_status(self) -> bool: - """Return mock iSWAP initialization status.""" - return True + """Return the configured simulated iSWAP initialization reading. + + This is not a firmware QW result. Unconfigured default is ``True``. + ``setup()`` does not change this field. + """ + return self._chatterbox_state.iswap_initialization_status @property def iswap_parked(self) -> bool: diff --git a/pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_chatterbox_tests.py b/pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_chatterbox_tests.py new file mode 100644 index 00000000000..ee3629b6136 --- /dev/null +++ b/pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_chatterbox_tests.py @@ -0,0 +1,260 @@ +# mypy: disable-error-code="attr-defined,method-assign" + +import inspect +import unittest + +from pylabrobot.legacy.liquid_handling import LiquidHandler +from pylabrobot.resources import ( + PLT_CAR_L5AC_A00, + TIP_CAR_480_A00, + cor_96_wellplate_360uL_Fb, + set_tip_tracking, +) +from pylabrobot.resources.hamilton import STARLetDeck, hamilton_96_tiprack_300uL_filter + +from .STAR_backend import STARBackend +from .STAR_chatterbox import STARChatterboxBackend, STARChatterboxState + + +class TestSTARChatterboxState(unittest.IsolatedAsyncioTestCase): + """Per-instance STAR chatterbox query state. Unconfigured defaults match the query replies.""" + + async def test_default_backend_keeps_unconfigured_query_values(self): + backend = STARChatterboxBackend() + + self.assertEqual(await backend.request_z_pos_channel_n(0), 285.0) + self.assertEqual(await backend.request_z_pos_channel_n(7), 285.0) + self.assertTrue(await backend.request_iswap_initialization_status()) + self.assertEqual(await backend.channel_dispensing_drive_request_position(0), 0.0) + self.assertEqual(await backend.channel_dispensing_drive_request_position(7), 0.0) + + async def test_custom_per_instance_query_values(self): + z_positions = [100.0, 110.0, 120.0, 130.0, 140.0, 150.0, 160.0, 170.0] + drive_positions = [1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0] + backend = STARChatterboxBackend( + chatterbox_state=STARChatterboxState( + iswap_initialization_status=False, + channel_z_positions=z_positions, + dispensing_drive_positions=drive_positions, + ) + ) + + for channel, z in enumerate(z_positions): + self.assertEqual(await backend.request_z_pos_channel_n(channel), z) + self.assertFalse(await backend.request_iswap_initialization_status()) + for channel, vol in enumerate(drive_positions): + self.assertEqual(await backend.channel_dispensing_drive_request_position(channel), vol) + + async def test_independent_backend_instances(self): + shared = STARChatterboxState(channel_z_positions=[200.0] * 8) + first = STARChatterboxBackend(chatterbox_state=shared) + second = STARChatterboxBackend(chatterbox_state=shared) + + first.chatterbox_state = STARChatterboxState( + iswap_initialization_status=False, + channel_z_positions=[12.0] + [200.0] * 7, + dispensing_drive_positions=[0.0, 9.5] + [0.0] * 6, + ) + + self.assertEqual(await first.request_z_pos_channel_n(0), 12.0) + self.assertEqual(await second.request_z_pos_channel_n(0), 200.0) + self.assertFalse(await first.request_iswap_initialization_status()) + self.assertTrue(await second.request_iswap_initialization_status()) + self.assertEqual(await first.channel_dispensing_drive_request_position(1), 9.5) + self.assertEqual(await second.channel_dispensing_drive_request_position(1), 0.0) + + async def test_input_and_returned_collections_are_copied(self): + z_positions = [285.0] * 8 + drive_positions = [0.0] * 8 + backend = STARChatterboxBackend( + chatterbox_state=STARChatterboxState( + channel_z_positions=z_positions, + dispensing_drive_positions=drive_positions, + ) + ) + + z_positions[0] = 1.0 + drive_positions[0] = 99.0 + self.assertEqual(await backend.request_z_pos_channel_n(0), 285.0) + self.assertEqual(await backend.channel_dispensing_drive_request_position(0), 0.0) + + with self.assertRaises(TypeError): + backend.chatterbox_state.channel_z_positions[0] = 50.0 # type: ignore[index] + with self.assertRaises(TypeError): + backend.chatterbox_state.dispensing_drive_positions[0] = 7.0 # type: ignore[index] + + heights = await backend.request_pip_height_last_lld() + heights[0] = 123.0 + self.assertEqual(await backend.request_pip_height_last_lld(), [0.0] * 8) + + y_spacings = await backend.channels_request_y_minimum_spacing() + y_spacings[0] = 99.0 + self.assertNotEqual((await backend.channels_request_y_minimum_spacing())[0], 99.0) + + def test_rejects_wrong_channel_vector_length(self): + with self.assertRaisesRegex(ValueError, "channel_z_positions"): + STARChatterboxBackend( + num_channels=8, + chatterbox_state=STARChatterboxState(channel_z_positions=[285.0] * 4), + ) + with self.assertRaisesRegex(ValueError, "dispensing_drive_positions"): + STARChatterboxBackend( + num_channels=8, + chatterbox_state=STARChatterboxState(dispensing_drive_positions=[0.0] * 16), + ) + + async def test_rejects_out_of_range_channel_index(self): + backend = STARChatterboxBackend(num_channels=8) + with self.assertRaisesRegex(ValueError, "channel"): + await backend.request_z_pos_channel_n(-1) + with self.assertRaisesRegex(ValueError, "channel"): + await backend.request_z_pos_channel_n(8) + with self.assertRaisesRegex(ValueError, "channel_idx"): + await backend.channel_dispensing_drive_request_position(-1) + with self.assertRaisesRegex(ValueError, "channel_idx"): + await backend.channel_dispensing_drive_request_position(8) + + async def test_setup_preserves_configured_query_state(self): + backend = STARChatterboxBackend( + chatterbox_state=STARChatterboxState( + iswap_initialization_status=False, + channel_z_positions=[90.0] * 8, + dispensing_drive_positions=[3.0] * 8, + ) + ) + backend.set_deck(STARLetDeck()) + await backend.setup() + + self.assertFalse(await backend.request_iswap_initialization_status()) + self.assertEqual(await backend.request_z_pos_channel_n(2), 90.0) + self.assertEqual(await backend.channel_dispensing_drive_request_position(2), 3.0) + + async def test_replacing_state_is_visible_to_queries(self): + backend = STARChatterboxBackend() + backend.chatterbox_state = STARChatterboxState( + iswap_initialization_status=False, + channel_z_positions=[40.0] * 8, + dispensing_drive_positions=[11.0] * 8, + ) + + self.assertFalse(await backend.request_iswap_initialization_status()) + self.assertEqual(await backend.request_z_pos_channel_n(0), 40.0) + self.assertEqual(await backend.channel_dispensing_drive_request_position(0), 11.0) + + async def test_replacing_state_validates_channel_length(self): + backend = STARChatterboxBackend(num_channels=8) + with self.assertRaisesRegex(ValueError, "channel_z_positions"): + backend.chatterbox_state = STARChatterboxState(channel_z_positions=[1.0]) + + def test_rejects_non_state_replacement(self): + backend = STARChatterboxBackend() + with self.assertRaisesRegex(TypeError, "STARChatterboxState"): + backend.chatterbox_state = object() # type: ignore[assignment] + + async def test_replacing_state_refills_omitted_vectors(self): + backend = STARChatterboxBackend( + chatterbox_state=STARChatterboxState( + channel_z_positions=[40.0] * 8, + dispensing_drive_positions=[11.0] * 8, + ) + ) + backend.chatterbox_state = STARChatterboxState(iswap_initialization_status=False) + + self.assertFalse(await backend.request_iswap_initialization_status()) + self.assertEqual(await backend.request_z_pos_channel_n(0), 285.0) + self.assertEqual(await backend.channel_dispensing_drive_request_position(0), 0.0) + + async def test_explicit_zero_dispensing_override_is_not_omitted_argument(self): + backend = STARChatterboxBackend( + chatterbox_state=STARChatterboxState(dispensing_drive_positions=[5.0] * 8) + ) + + self.assertEqual(await backend.channel_dispensing_drive_request_position(0), 5.0) + self.assertEqual( + await backend.channel_dispensing_drive_request_position(0, simulated_value=0.0), + 0.0, + ) + self.assertEqual( + await backend.channel_dispensing_drive_request_position(0, simulated_value=12.5), + 12.5, + ) + self.assertEqual(await backend.channel_dispensing_drive_request_position(0), 5.0) + + def test_hardware_backend_does_not_define_chatterbox_state(self): + with self.assertRaises(AttributeError): + STARBackend.chatterbox_state + self.assertIsInstance(STARChatterboxBackend.chatterbox_state, property) + + +class TestSTARChatterboxStateDoesNotOwnTrackers(unittest.IsolatedAsyncioTestCase): + """Tip queries stay on TipTracker; last-LLD stays latched by simulated sensing.""" + + async def asyncSetUp(self): + self.backend = STARChatterboxBackend( + chatterbox_state=STARChatterboxState( + iswap_initialization_status=False, + channel_z_positions=[77.0] * 8, + ) + ) + self.deck = STARLetDeck() + self.lh = LiquidHandler(self.backend, deck=self.deck) + self.tip_car = TIP_CAR_480_A00(name="tip carrier") + self.tip_car[1] = self.tip_rack = hamilton_96_tiprack_300uL_filter(name="tip_rack_01") + self.deck.assign_child_resource(self.tip_car, rails=1) + self.plt_car = PLT_CAR_L5AC_A00(name="plate carrier") + self.plt_car[0] = self.plate = cor_96_wellplate_360uL_Fb(name="plate_01") + self.deck.assign_child_resource(self.plt_car, rails=9) + await self.lh.setup() + + async def asyncTearDown(self): + await self.lh.stop() + + def test_state_object_has_no_tip_fields(self): + params = inspect.signature(STARChatterboxState.__init__).parameters + self.assertNotIn("tip_presence", params) + self.assertNotIn("head96_tip_presence", params) + self.assertNotIn("tip_length", params) + self.assertNotIn("last_lld_heights", params) + + async def test_tip_presence_follows_tracker_not_chatterbox_state(self): + self.assertEqual(await self.backend.request_tip_presence(), [False] * 8) + + await self.lh.pick_up_tips(self.tip_rack["A1"], use_channels=[0]) + presence = await self.backend.request_tip_presence() + self.assertEqual(presence[0], True) + self.assertEqual(presence[1:], [False] * 7) + + tip = self.lh.head[0].get_tip() + self.assertEqual(await self.backend.request_tip_len_on_channel(0), tip.total_tip_length) + + async def test_head96_tip_presence_follows_tracker(self): + set_tip_tracking(enabled=True) + try: + self.assertEqual(await self.backend.head96_request_tip_presence(), 0) + await self.lh.pick_up_tips96(self.tip_rack) + self.assertEqual(await self.backend.head96_request_tip_presence(), 1) + finally: + set_tip_tracking(enabled=False) + + async def test_last_lld_latching_still_works_with_configured_state(self): + self.assertFalse(await self.backend.request_iswap_initialization_status()) + self.assertEqual(await self.backend.request_z_pos_channel_n(0), 77.0) + + channel = 3 + well = self.plate.get_item("D1") + await self.lh.pick_up_tips(self.tip_rack["D1"], use_channels=[channel]) + well.tracker.set_volume(150.0) + await self.backend.probe_liquid_heights(containers=[well], use_channels=[channel]) + + expected = well.get_absolute_location( + "c", "c", "cavity_bottom" + ).z + well.compute_height_from_volume(150.0) + heights = await self.backend.request_pip_height_last_lld() + self.assertAlmostEqual(heights[channel], expected) + self.assertEqual( + [h for i, h in enumerate(heights) if i != channel], + [0.0] * (self.backend.num_channels - 1), + ) + leaked = heights + leaked[channel] = 0.0 + self.assertAlmostEqual((await self.backend.request_pip_height_last_lld())[channel], expected)