diff --git a/docs/user_guide/00_liquid-handling/hamilton-prep/prep_basic_demo.ipynb b/docs/user_guide/00_liquid-handling/hamilton-prep/prep_basic_demo.ipynb index 95ee2afa899..2abb141071a 100644 --- a/docs/user_guide/00_liquid-handling/hamilton-prep/prep_basic_demo.ipynb +++ b/docs/user_guide/00_liquid-handling/hamilton-prep/prep_basic_demo.ipynb @@ -325,7 +325,7 @@ "\n", "await prep.pipettes.pick_up_tips(tip_spots, use_channels=channels)\n", "status(\"after pickup\")\n", - "await prep.pipettes.aspirate(source, vols=volumes, use_channels=channels, liquid_height=[3.0, 3.0])\n", + "await prep.pipettes.aspirate(source, piston_volumes=volumes, use_channels=channels, liquid_heights=[3.0, 3.0])\n", "status(\"after aspirate\")\n", "await prep.pipettes.dispense(destination, vols=volumes, use_channels=channels, liquid_height=[3.0, 3.0])\n", "status(\"after dispense\")\n", diff --git a/pylabrobot/hamilton/prep/driver/features/head8.py b/pylabrobot/hamilton/prep/driver/features/head8.py index 56f22167ee4..e34846d6f9f 100644 --- a/pylabrobot/hamilton/prep/driver/features/head8.py +++ b/pylabrobot/hamilton/prep/driver/features/head8.py @@ -544,7 +544,7 @@ async def drop_tips( for ch in use_channels: used = mounted[ch].tracker.get_used_volume() if not mounted[ch].tracker.is_disabled and used > 1e-6: - raise RuntimeError(f"Cannot drop tip on channel {ch} with volume {used} uL") + logger.warning("dropping the tip on channel %d with %s uL in it", ch, used) spots = [d for d in destinations if isinstance(d, TipSpot)] if len({id(spot) for spot in spots}) != len(spots): raise ValueError("each tip must go into a spot of its own") diff --git a/pylabrobot/hamilton/prep/driver/features/head8_tests.py b/pylabrobot/hamilton/prep/driver/features/head8_tests.py index 0bc76df5581..dfbd2dffa2a 100644 --- a/pylabrobot/hamilton/prep/driver/features/head8_tests.py +++ b/pylabrobot/hamilton/prep/driver/features/head8_tests.py @@ -471,6 +471,28 @@ async def _run() -> None: asyncio.run(_run()) +def test_head8_aspirate_pressure_without_p_lld_raises(): + """lld_mode=PRESSURE with no p_lld is refused and no aspirate is sent.""" + + async def _run() -> None: + deck, tip_rack, src_plate, _ = _make_deck() + p = PrepSimulationDriver(deck=deck, declared_configuration_json=RECORDING_PREP_HEAD8) + await p.setup() + assert p.head8 is not None + + await p.head8.pick_up_tips(tip_rack.column(0)) + captured, _ = _record_send(p) + with pytest.raises(ValueError, match="needs p_lld"): + await p.head8.aspirate( + wells=src_plate.column(0), volume=10, lld_mode=Pipettes.LLDMode.PRESSURE + ) + assert captured == [] + + await p.stop() + + asyncio.run(_run()) + + def test_head8_aspirate_lld_and_tadm_sends_mphaspirate_with_lld_tadm2(): """lld_mode=CAPACITIVE + tadm= → MphAspirateWithLldTadm2.""" diff --git a/pylabrobot/hamilton/prep/driver/features/pipettes.py b/pylabrobot/hamilton/prep/driver/features/pipettes.py index a23882936c4..f0031dfbfa0 100644 --- a/pylabrobot/hamilton/prep/driver/features/pipettes.py +++ b/pylabrobot/hamilton/prep/driver/features/pipettes.py @@ -53,13 +53,23 @@ from pylabrobot.hamilton.transport.tcp.messages import HoiParamsParser, parse_into_struct from pylabrobot.hamilton.transport.tcp.packets import Address from pylabrobot.legacy.liquid_handling.errors import ChannelizedError +from pylabrobot.lib.liquid_handling.channel_positioning import ( + compute_nonconsecutive_channel_offsets, +) +from pylabrobot.lib.liquid_handling.liquid_height import get_liquid_height_from_volume +from pylabrobot.lib.liquid_handling.mix import Mix from pylabrobot.lib.liquid_handling.pipette_batch_scheduling import ( ChannelBatch, plan_batches, validate_channel_selections, ) -from pylabrobot.resources import Container, Coordinate, Tip -from pylabrobot.resources.errors import HasTipError, NoTipError +from pylabrobot.resources import Container, Coordinate, Tip, does_volume_tracking +from pylabrobot.resources.errors import ( + HasTipError, + NoTipError, + TooLittleLiquidError, + TooLittleVolumeError, +) from pylabrobot.resources.hamilton import HamiltonTip, PrepDeck, TipSize from pylabrobot.resources.hamilton.core_grippers import HamiltonCoreGrippers from pylabrobot.resources.n_channel_pipettes import TipMountingShaft @@ -129,33 +139,33 @@ def default_lld_params( ) -> Pipettes._LldDefaults: """Build resolved pLLD / cLLD defaults. - When LLD is active and no caller override is given, returns non-default - parameters (``default_values=False``) so the firmware actually triggers - detection. Capacitive-only seeks leave the pressure block at firmware - defaults so the dispenser is not started at speed 0. Otherwise returns - firmware defaults. + With LLD active, cLLD gets non-default parameters so the firmware triggers detection; pLLD stays + on firmware defaults unless given. Args: effective_lld: Whether this call uses any LLD seek. - p_lld: Caller override for pressure LLD parameters. + p_lld: Caller override for pressure LLD parameters; required for PRESSURE and DUAL. c_lld: Caller override for capacitive LLD parameters. - lld_mode: Which LLD mode the call resolved to. ``CAPACITIVE`` keeps pLLD - on firmware defaults unless ``p_lld`` is set. + lld_mode: Which LLD mode the call resolved to. + + Raises: + ValueError: If a pressure mode has no `p_lld`, or a non-default `p_lld` seeks outside 1 to + 630 uL/s. """ + if p_lld is not None and not p_lld.default_values: + if not 1.0 <= p_lld.dispenser_seek_speed <= 630.0: + raise ValueError( + f"p_lld dispenser_seek_speed must be 1 to 630 uL/s, is {p_lld.dispenser_seek_speed}" + ) + resolved_p = p_lld or PrepCmd.PLldParameters.default() if not effective_lld: - resolved_p = p_lld or PrepCmd.PLldParameters.default() resolved_c = c_lld or PrepCmd.CLldParameters.default() return Pipettes._LldDefaults(p_lld=resolved_p, c_lld=resolved_c) - if lld_mode == Pipettes.LLDMode.CAPACITIVE: - resolved_p = p_lld or PrepCmd.PLldParameters.default() - else: - resolved_p = p_lld or PrepCmd.PLldParameters( - default_values=False, - sensitivity=1, - dispenser_seek_speed=0.0, - lld_height_difference=0.0, - detect_mode=0, + pressure = (Pipettes.LLDMode.PRESSURE, Pipettes.LLDMode.DUAL) + if lld_mode is not None and lld_mode in pressure and p_lld is None: + raise ValueError( + f"{lld_mode.name} LLD needs p_lld with a dispenser seek speed of 1 to 630 uL/s" ) resolved_c = c_lld or PrepCmd.CLldParameters( default_values=False, @@ -302,6 +312,52 @@ def lld_seek_timeout( return max(distance / speed + LLD_READ_TIMEOUT_PAD_S, LLD_READ_TIMEOUT_FLOOR_S) +def get_mix_parameters( + pre_mixes: Optional[Sequence[Optional[Mix]]], + n: int, + mix_positions_from_liquid_surface: Optional[Sequence[float]] = None, +) -> List[PrepCmd.MixParameters]: + """One mix block per container; the default, which mixes nothing, where there is no `Mix`. + + Args: + pre_mixes: a `Mix` per container, None for none. + n: how many containers. + mix_positions_from_liquid_surface: mixing depth under the aspirate height, in mm, per + container. 0.0 when None. + + Returns: + One mix block per container, in the containers' order. + + Raises: + ValueError: If a list and the containers differ in number, or a `Mix` repeats fewer than 1 or + more than 255 times. + """ + mixes = list(pre_mixes) if pre_mixes is not None else [None] * n + if len(mixes) != n: + raise ValueError(f"pre_mixes length must match containers ({n})") + depths = fill_in_defaults( + None if mix_positions_from_liquid_surface is None else list(mix_positions_from_liquid_surface), + [0.0] * n, + ) + blocks: List[PrepCmd.MixParameters] = [] + for m, depth in zip(mixes, depths): + if m is None: + blocks.append(PrepCmd.MixParameters.default()) + continue + if not 1 <= m.repetitions <= 255: + raise ValueError(f"a Mix repeats 1 to 255 times, is {m.repetitions}") + blocks.append( + PrepCmd.MixParameters( + default_values=False, + z_offset=depth, + volume=m.volume, + cycles=m.repetitions, + speed=m.flow_rate, + ) + ) + return blocks + + def _effective_radius(resource) -> float: """Effective radius for PrepCmd.CommonParameters.tube_radius. @@ -920,7 +976,7 @@ def _build_pipettor_gantry_move_parameters( class _AspirateChannelKit: """Pre-resolved per-channel values for one aspirate channel. - Computed once by ``_resolve_aspirate_channels``; the variant (LLD x monitoring + Built by ``_aspirate_in_one_move``; the variant (LLD x monitoring x v1/v2) only decides which fields get assembled into which wire dataclass. """ @@ -4285,11 +4341,11 @@ async def _drop_tips_in_one_move( if ch in indexed ] - # Nothing is dropped with liquid in it, two into one spot, or into a spot that is taken + # A tip holding liquid is dropped with a warning; two into one spot, or a taken spot, never for ch, tip in zip(use_channels, tips): if not tip.tracker.is_disabled and tip.tracker.get_used_volume() > 1e-6: - raise RuntimeError( - f"Cannot drop tip on channel {ch} with volume {tip.tracker.get_used_volume()} uL" + logger.warning( + "dropping the tip on channel %d with %s uL in it", ch, tip.tracker.get_used_volume() ) spots = [dest for dest in destinations if isinstance(dest, TipSpot)] if len({id(spot) for spot in spots}) != len(spots): @@ -4763,7 +4819,7 @@ def _resolve_channel_context( z_bottom_search_offset: Optional[List[float]] = None, container_segments: Optional[List[List[PrepCmd.SegmentDescriptor]]] = None, auto_container_geometry: bool = False, - hamilton_liquid_classes: Optional[List[HamiltonLiquidClass]] = None, + hamilton_liquid_classes: Optional[Sequence[Optional[HamiltonLiquidClass]]] = None, disable_volume_correction: Optional[List[bool]] = None, ) -> _ChannelContext[_OpT]: """Resolve shared per-channel state for aspirate or dispense. @@ -4840,115 +4896,7 @@ def _resolve_channel_context( ch_segments=ch_segments, ) - # -- aspirate: resolve, assemble, send ----------------------------------------------------------- - - def _resolve_aspirate_channels( - self, - ops: List[_PipetteTransfer], - use_channels: List[int], - effective_lld: bool, - *, - z_final: Optional[List[float]] = None, - z_fluid: Optional[List[float]] = None, - z_air: Optional[List[float]] = None, - settling_time: Optional[List[float]] = None, - transport_air_volume: Optional[List[float]] = None, - z_liquid_exit_speed: Optional[List[float]] = None, - prewet_volume: Optional[List[float]] = None, - z_minimum: Optional[List[float]] = None, - z_bottom_search_offset: Optional[List[float]] = None, - lld: Optional[PrepCmd.LldParameters] = None, - p_lld: Optional[PrepCmd.PLldParameters] = None, - c_lld: Optional[PrepCmd.CLldParameters] = None, - tadm: Optional[PrepCmd.TadmParameters] = None, - container_segments: Optional[List[List[PrepCmd.SegmentDescriptor]]] = None, - auto_container_geometry: bool = False, - hamilton_liquid_classes: Optional[List[HamiltonLiquidClass]] = None, - disable_volume_correction: Optional[List[bool]] = None, - lld_mode: Optional[Pipettes.LLDMode] = None, - ) -> list[_AspirateChannelKit]: - """Resolve all per-channel values for aspirate (pure computation, no I/O).""" - ctx = self._resolve_channel_context( - ops, - use_channels, - z_final=z_final, - z_fluid=z_fluid, - z_air=z_air, - z_minimum=z_minimum, - z_bottom_search_offset=z_bottom_search_offset, - container_segments=container_segments, - auto_container_geometry=auto_container_geometry, - hamilton_liquid_classes=hamilton_liquid_classes, - disable_volume_correction=disable_volume_correction, - ) - - # Aspirate-specific HLC defaults - hlcs = ctx.hlcs - settling_time = fill_in_defaults( - settling_time, [hlc.aspiration_settling_time if hlc is not None else 1.0 for hlc in hlcs] - ) - transport_air_volume = fill_in_defaults( - transport_air_volume, - [hlc.aspiration_air_transport_volume if hlc is not None else 0.0 for hlc in hlcs], - ) - z_liquid_exit_speed = fill_in_defaults( - z_liquid_exit_speed, [hlc.aspiration_swap_speed if hlc is not None else 10.0 for hlc in hlcs] - ) - prewet_volume = fill_in_defaults( - prewet_volume, - [hlc.aspiration_over_aspirate_volume if hlc is not None else 0.0 for hlc in hlcs], - ) - flow_rates = [ - op.flow_rate or (hlc.aspiration_flow_rate if hlc is not None else 100.0) - for op, hlc in zip(ops, hlcs) - ] - blowout_volumes = [ - op.blow_out_air_volume or (hlc.aspiration_blow_out_volume if hlc is not None else 0.0) - for op, hlc in zip(ops, hlcs) - ] - - lld_defaults = self._default_lld_params(effective_lld, p_lld, c_lld, lld_mode=lld_mode) - _tadm = tadm or PrepCmd.TadmParameters.default() - - kits: list[_AspirateChannelKit] = [] - for ch in range(self.num_channels): - if ch not in ctx.indexed_ops: - continue - idx = ctx.ch_to_idx[ch] - asp = ctx.indexed_ops[ch] - loc = asp.resource.get_location_wrt(self._require_deck(), "c", "c", "cavity_bottom") - radius = _effective_radius(asp.resource) - - kits.append( - _AspirateChannelKit( - channel=self.channel_enum(ch), - aspirate=PrepCmd.AspirateParameters.from_location( - loc, prewet_volume=prewet_volume[idx], blowout_volume=blowout_volumes[idx] - ), - common=PrepCmd.CommonParameters.for_op( - ctx.volumes[idx], - radius, - flow_rate=flow_rates[idx], - z_minimum=ctx.z_minimum[idx], - z_final=ctx.z_final[idx], - z_liquid_exit_speed=z_liquid_exit_speed[idx], - transport_air_volume=transport_air_volume[idx], - settling_time=settling_time[idx], - ), - segments=ctx.ch_segments[ch], - no_lld=PrepCmd.NoLldParameters.for_fixed_z( - ctx.z_fluid[idx], ctx.z_air[idx], z_bottom_search_offset=ctx.z_bottom_search_offset[idx] - ), - lld=self._lld_for_well(effective_lld, lld, ctx.well_geometry[idx].top_of_well), - p_lld=lld_defaults.p_lld, - c_lld=lld_defaults.c_lld, - monitoring=PrepCmd.AspirateMonitoringParameters.default(), - tadm=_tadm, - mix=PrepCmd.MixParameters.default(), - adc=PrepCmd.AdcParameters.default(), - ) - ) - return kits + # -- aspirate: assemble, send -------------------------------------------------------------------- @staticmethod def _assemble_aspirate_v2( @@ -5084,19 +5032,200 @@ def _assemble_aspirate_v1( (False, False, False): PrepCmd.PrepAspirateNoLldMonitoring, } - def _aspirate_command( + async def _unchecked_fw_aspirate( self, kits: list[_AspirateChannelKit], effective_lld: bool, is_tadm: bool, use_v2: bool, - ) -> TCPCommand[None]: - """The aspirate command for these channels, with the param types this firmware takes.""" + read_timeout: Optional[float] = None, + ) -> None: + """Send the aspiration as it is given: one entry per channel, its structs as they are. + + Args: + kits: each channel's firmware structs. + effective_lld: whether the LLD variant is sent. + is_tadm: whether the TADM variant is sent. + use_v2: whether the v2 command is sent, with its container description. + read_timeout: answer timeout in s. The link's when None. + """ cmd_cls = self._ASPIRATE_CMD[(effective_lld, is_tadm, use_v2)] assembler = self._assemble_aspirate_v2 if use_v2 else self._assemble_aspirate_v1 params = [assembler(k, effective_lld, is_tadm) for k in kits] command: TCPCommand[None] = cmd_cls(aspirate_parameters=params) # type: ignore[arg-type] - return command + await self._driver.send_command(command, read_timeout=read_timeout) + + async def _aspirate_in_one_move( + self, + use_channels: List[int], + locations: List[Coordinate], + lld_search_heights: List[float], + minimum_allowed_z_position_during: List[float], + piston_volumes: List[float], + *, + tube_radii: List[float], + lld_mode: Optional[Pipettes.LLDMode] = None, + clld_sensitivity: Optional[int] = None, + immersion_depths: Optional[List[float]] = None, + blow_out_air_volumes: Optional[List[float]] = None, + pre_wetting_volumes: Optional[List[float]] = None, + pre_mixes: Optional[Sequence[Optional[Mix]]] = None, + mix_positions_from_liquid_surface: Optional[List[float]] = None, + flow_rates: Optional[List[float]] = None, + settling_times: Optional[List[float]] = None, + swap_speeds: Optional[List[float]] = None, + pull_out_distance_transport_air: float = 10.0, + transport_air_volumes: Optional[List[float]] = None, + minimum_traverse_height_end: Optional[float] = None, + z_air: Optional[List[float]] = None, + z_bottom_search_offset: Optional[List[float]] = None, + container_segments: Optional[List[List[PrepCmd.SegmentDescriptor]]] = None, + lld: Optional[PrepCmd.LldParameters] = None, + p_lld: Optional[PrepCmd.PLldParameters] = None, + clot_detection_heights: Optional[List[float]] = None, + tadm: Optional[PrepCmd.TadmParameters] = None, + read_timeout: Optional[float] = None, + command_version: Optional[Literal["v1", "v2"]] = None, + check_only: bool = False, + ) -> None: + """Aspirate at each place given, the channels together, in one command. + + Positions and heights on the deck in mm, volumes in uL, speeds in mm/s or uL/s, times in s, one + entry per channel in `use_channels`' order. No model update: `aspirate` does that. + + Args: + use_channels: which channels, 0-indexed from the back. + locations: where each tip bottom goes; the z is the liquid surface. + lld_search_heights: where each LLD search starts. + minimum_allowed_z_position_during: how low each tip bottom may go. + piston_volumes: what each piston draws. + tube_radii: each container's radius, for the firmware's surface following. + lld_mode: how the liquid is found, one mode for every channel. None runs a search only when + `lld` is given. + clld_sensitivity: capacitive LLD sensitivity. 3 when None. + immersion_depths: how far under the surface each tip aspirates. With LLD the search's + `z_submerge`, 2.0 when None; without, off the location's z, 0.0 when None. + blow_out_air_volumes: air drawn before the liquid. 0.0 when None. + pre_wetting_volumes: drawn and returned first. 0.0 when None. + pre_mixes: a `Mix` per channel, mixed before the draw, None for no mixing. + mix_positions_from_liquid_surface: mixing depth under the aspirate height. 0.0 when None. + flow_rates: 100.0 when None. + settling_times: wait in the liquid. 1.0 when None. + swap_speeds: speed of leaving the liquid. 10.0 when None. + pull_out_distance_transport_air: rise above the surface before drawing transport air. + transport_air_volumes: air drawn after the liquid. 0.0 when None. + minimum_traverse_height_end: tip bottom height at the end. The traverse height less each + tip's overhang when None. + z_air: each tip bottom height to draw transport air at, in place of the pull-out distance. + z_bottom_search_offset: 2.0 when None. + container_segments: each container's cross-sections. None sends none. + lld: the LLD search's block as it is, in place of the one built here. + p_lld: the pressure LLD block as it is. + clot_detection_heights: how far a clot may hold each tip back, in mm. 0.0 when None; only 0.0 + until the check is verified on the device. + tadm: the TADM block as it is; given, the aspiration is monitored. + read_timeout: answer timeout in s. Long enough for the search when LLD runs and None. + command_version: "v1" or "v2". What the firmware supports when None. + check_only: refuse what would be refused, and send nothing. + + Raises: + ValueError: If the lists do not match or a channel is out of range. + """ + n = len(use_channels) + for name, length in ( + ("locations", len(locations)), + ("lld_search_heights", len(lld_search_heights)), + ("minimum_allowed_z_position_during", len(minimum_allowed_z_position_during)), + ("piston_volumes", len(piston_volumes)), + ("tube_radii", len(tube_radii)), + ): + if length != n: + raise ValueError(f"{name} length must match use_channels ({n})") + if any(not 0 <= ch < self.num_channels for ch in use_channels): + raise ValueError(f"use_channels index out of range (valid: 0..{self.num_channels - 1})") + effective_lld = self._resolve_effective_lld( + None if lld_mode is None else [lld_mode] * n, lld, n + ) + clot_detection_heights = fill_in_defaults(clot_detection_heights, [0.0] * n) + if any(h != 0 for h in clot_detection_heights): + raise ValueError("clot detection is not verified on the Prep yet; give 0.0") + c_lld = None + if clld_sensitivity is not None: + base = default_lld_params(True, lld_mode=Pipettes.LLDMode.CAPACITIVE).c_lld + c_lld = replace(base, sensitivity=clld_sensitivity) + lld_defaults = self._default_lld_params(effective_lld, p_lld, c_lld, lld_mode=lld_mode) + blow_out_air_volumes = fill_in_defaults(blow_out_air_volumes, [0.0] * n) + pre_wetting_volumes = fill_in_defaults(pre_wetting_volumes, [0.0] * n) + flow_rates = fill_in_defaults(flow_rates, [100.0] * n) + if any(f <= 0 for f in flow_rates): + raise ValueError(f"flow_rates must be above 0, got {flow_rates}") + settling_times = fill_in_defaults(settling_times, [1.0] * n) + swap_speeds = fill_in_defaults(swap_speeds, [10.0] * n) + transport_air_volumes = fill_in_defaults(transport_air_volumes, [0.0] * n) + z_bottom_search_offset = fill_in_defaults(z_bottom_search_offset, [2.0] * n) + z_air = fill_in_defaults(z_air, [loc.z + pull_out_distance_transport_air for loc in locations]) + if immersion_depths is not None and len(immersion_depths) != n: + raise ValueError(f"immersion_depths length must match use_channels ({n})") + if minimum_traverse_height_end is not None: + z_finals = [minimum_traverse_height_end] * n + else: + traverse = self._resolve_traverse_height(None) + tips = self._require_mounted_tips(use_channels) + z_finals = [traverse - (tip.get_size_z() - tip.fitting_depth) for tip in tips] + mix_blocks = get_mix_parameters(pre_mixes, n, mix_positions_from_liquid_surface) + + kits: list[_AspirateChannelKit] = [] + for i in sorted(range(n), key=lambda j: use_channels[j]): + loc = locations[i] + lld_block = self._lld_for_well(effective_lld, lld, lld_search_heights[i]) + if immersion_depths is not None: + lld_block = replace(lld_block, z_submerge=immersion_depths[i]) + z_fluid = loc.z - (immersion_depths[i] if immersion_depths is not None else 0.0) + kits.append( + _AspirateChannelKit( + channel=self.channel_enum(use_channels[i]), + aspirate=PrepCmd.AspirateParameters.from_location( + loc, prewet_volume=pre_wetting_volumes[i], blowout_volume=blow_out_air_volumes[i] + ), + common=PrepCmd.CommonParameters.for_op( + piston_volumes[i], + tube_radii[i], + flow_rate=flow_rates[i], + z_minimum=minimum_allowed_z_position_during[i], + z_final=z_finals[i], + z_liquid_exit_speed=swap_speeds[i], + transport_air_volume=transport_air_volumes[i], + settling_time=settling_times[i], + ), + segments=container_segments[i] if container_segments is not None else [], + no_lld=PrepCmd.NoLldParameters.for_fixed_z( + z_fluid, z_air[i], z_bottom_search_offset=z_bottom_search_offset[i] + ), + lld=lld_block, + p_lld=lld_defaults.p_lld, + c_lld=lld_defaults.c_lld, + monitoring=PrepCmd.AspirateMonitoringParameters.default(), + tadm=tadm or PrepCmd.TadmParameters.default(), + mix=mix_blocks[i], + adc=PrepCmd.AdcParameters.default(), + ) + ) + + if read_timeout is None and effective_lld: + read_timeout = lld_seek_timeout( + kits[0].lld, + min(minimum_allowed_z_position_during), + approach_from_z=self.default_minimum_traverse_height, + ) + if check_only: + return + await self._unchecked_fw_aspirate( + kits, + effective_lld, + tadm is not None, + self._resolve_command_version(command_version), + read_timeout=read_timeout if effective_lld else None, + ) # -- dispense: resolve, assemble, send ----------------------------------------------------------- @@ -5327,107 +5456,256 @@ def _build_transfers( for r, t, v, o, lh, fr, bav in zip(resources, tips, vols, offs, lhs, frs, bavs) ] - async def aspirate( + def _get_liquid_heights( self, - resources: Sequence[Container], - vols: Sequence[float], - use_channels: Optional[List[int]] = None, + containers: Sequence[Container], + liquid_heights: Optional[Sequence[Optional[float]]], + effective_lld: bool, + ) -> List[Optional[float]]: + """Each container's liquid height above its cavity bottom: as given, else from its volume. + + Args: + containers: the containers, in the call's order. + liquid_heights: the heights given, in mm, None where not given. + effective_lld: whether an LLD search runs; it finds the surface, so no height is needed. + + Returns: + The height of each container, in mm, None where the search is left to find it. + + Raises: + ValueError: If the heights and containers differ in number. + RuntimeError: If a height is missing, volume tracking is off and no LLD runs. + """ + heights = list(liquid_heights) if liquid_heights is not None else [None] * len(containers) + if len(heights) != len(containers): + raise ValueError(f"liquid_heights length must match containers ({len(containers)})") + tracking = does_volume_tracking() + resolved: List[Optional[float]] = [] + for container, height in zip(containers, heights): + if height is None and not effective_lld: + if not tracking: + raise RuntimeError( + f"no liquid height given for {container.name}, volume tracking is off and no LLD " + "runs, so nothing knows where its liquid is" + ) + height = get_liquid_height_from_volume(container, container.tracker.get_used_volume()) + resolved.append(height) + return resolved + + def _get_resource_offsets( + self, containers: Sequence[Container], use_channels: List[int] + ) -> List[Coordinate]: + """Each channel's offset in its container: channels sharing one spread across it in Y. + + Args: + containers: one per channel used. + use_channels: which channels, in the containers' order. + + Returns: + One offset per container; zero for a container only one channel goes to. + + Raises: + ValueError: If the channels sharing a container do not fit in it. + """ + offsets = [Coordinate.zero()] * len(containers) + for container in {id(c): c for c in containers}.values(): + jobs = [i for i, c in enumerate(containers) if c is container] + if len(jobs) < 2: + continue + jobs.sort(key=lambda i: use_channels[i]) + spread = compute_nonconsecutive_channel_offsets( + container, [use_channels[i] for i in jobs], self.minimum_y_spacings + ) + if spread is None: + raise ValueError( + f"channels {[use_channels[i] for i in jobs]} do not fit in {container.name}" + ) + for i, offset in zip(jobs, spread): + offsets[i] = offset + return offsets + + async def _aspirate_batch( + self, + x_position: float, + containers: List[Container], + drawn: List[float], + use_channels: List[int], + resource_offsets: List[Coordinate], + effective_lld: bool, + by_liquid_class: bool, + liquid_heights: Optional[List[Optional[float]]], + lld_mode: Optional[Pipettes.LLDMode], + flow_rates: Optional[List[Optional[float]]], *, - flow_rates: Optional[List[Optional[float]]] = None, - offsets: Optional[List[Coordinate]] = None, - liquid_height: Optional[List[Optional[float]]] = None, - blow_out_air_volume: Optional[List[Optional[float]]] = None, - z_final: Optional[List[float]] = None, - z_fluid: Optional[List[float]] = None, - z_air: Optional[List[float]] = None, - settling_time: Optional[List[float]] = None, - transport_air_volume: Optional[List[float]] = None, - z_liquid_exit_speed: Optional[List[float]] = None, - prewet_volume: Optional[List[float]] = None, - z_minimum: Optional[List[float]] = None, - z_bottom_search_offset: Optional[List[float]] = None, - lld_mode: Optional[List[Any]] = None, - lld: Optional[PrepCmd.LldParameters] = None, - p_lld: Optional[PrepCmd.PLldParameters] = None, - c_lld: Optional[PrepCmd.CLldParameters] = None, - tadm: Optional[PrepCmd.TadmParameters] = None, - container_segments: Optional[List[List[PrepCmd.SegmentDescriptor]]] = None, - auto_container_geometry: bool = False, - hamilton_liquid_classes: Optional[List[HamiltonLiquidClass]] = None, - disable_volume_correction: Optional[List[bool]] = None, - read_timeout: Optional[float] = None, - command_version: Optional[Literal["v1", "v2"]] = None, - ): - """Aspirate from containers using mounted tips. + hamilton_liquid_classes: Optional[List[HamiltonLiquidClass]], + clld_sensitivity: Optional[int], + immersion_depths: Optional[List[float]], + blow_out_air_volumes: Optional[List[Optional[float]]], + pre_wetting_volumes: Optional[List[float]], + lld: Optional[PrepCmd.LldParameters], + p_lld: Optional[PrepCmd.PLldParameters], + clot_detection_heights: Optional[List[float]], + z_fluid: Optional[List[float]], + minimum_allowed_z_positions_during: Optional[List[float]], + z_bottom_search_offset: Optional[List[float]], + pre_mixes: Optional[List[Optional[Mix]]], + mix_positions_from_liquid_surface: Optional[List[float]], + settling_times: Optional[List[float]], + swap_speeds: Optional[List[float]], + transport_air_volumes: Optional[List[float]], + z_air: Optional[List[float]], + minimum_traverse_height_end: Optional[float], + tadm: Optional[PrepCmd.TadmParameters], + container_segments: Optional[List[List[PrepCmd.SegmentDescriptor]]], + surface_following_distances: Optional[List[float]], + read_timeout: Optional[float], + command_version: Optional[Literal["v1", "v2"]], + check_only: bool, + ) -> List[float]: + """Aspirate one batch in one command, the channels already over it; book and settle volumes. - Explicit kwargs override Hamilton liquid-class defaults; HLC supplies - unspecified fields and the volume correction curve unless disabled. + Every other argument is `aspirate`'s, one entry per container of the batch where per container. + + Args: + x_position: the batch's X, sent for every channel, in mm. + drawn: the volumes, or the piston volumes, per container, in uL. + resource_offsets: as `aspirate` resolved them. + effective_lld: whether an LLD search runs. + by_liquid_class: whether `drawn` are liquid volumes, corrected by a liquid class. + minimum_traverse_height_end: the tip bottom height every tip is left at, in mm. + check_only: refuse what would be refused; nothing is booked or sent. + + Returns: + The liquid volume each channel takes, in uL. """ - resources = list(resources) - use_channels = use_channels if use_channels is not None else list(range(len(resources))) + n = len(containers) ops = self._build_transfers( - resources, - vols, + containers, + drawn, use_channels, - offsets=offsets, - liquid_height=liquid_height, + offsets=resource_offsets, + liquid_height=self._get_liquid_heights(containers, liquid_heights, effective_lld), flow_rates=flow_rates, - blow_out_air_volume=blow_out_air_volume, + blow_out_air_volume=blow_out_air_volumes, ) - effective_lld = self._resolve_effective_lld(lld_mode, lld, len(ops)) - is_tadm = tadm is not None - use_v2 = self._resolve_command_version(command_version) - - kits = self._resolve_aspirate_channels( + classes: List[Optional[HamiltonLiquidClass]] = [None] * n + if by_liquid_class: + classes = resolve_hamilton_liquid_classes( + None if hamilton_liquid_classes is None else list(hamilton_liquid_classes), + ops, + jet=False, + blow_out=False, + ) + for op, hlc in zip(ops, classes): + if hlc is None: + raise ValueError( + f"no liquid class for {op.tip}; give hamilton_liquid_classes, or piston_volumes" + ) + ctx = self._resolve_channel_context( ops, use_channels, - effective_lld, - z_final=z_final, z_fluid=z_fluid, z_air=z_air, - settling_time=settling_time, - transport_air_volume=transport_air_volume, - z_liquid_exit_speed=z_liquid_exit_speed, - prewet_volume=prewet_volume, - z_minimum=z_minimum, + z_minimum=minimum_allowed_z_positions_during, z_bottom_search_offset=z_bottom_search_offset, - lld=lld, - p_lld=p_lld, - c_lld=c_lld, - tadm=tadm, container_segments=container_segments, - auto_container_geometry=auto_container_geometry, - hamilton_liquid_classes=hamilton_liquid_classes, - disable_volume_correction=disable_volume_correction, - lld_mode=self._single_lld_mode(lld_mode), + hamilton_liquid_classes=classes, ) - - lld_read_timeout = read_timeout - if lld_read_timeout is None and effective_lld and kits: - min_z_min = min(k.common.z_minimum for k in kits) - lld_read_timeout = lld_seek_timeout( - kits[0].lld, - min_z_min, - approach_from_z=self.default_minimum_traverse_height, + deck = self._require_deck() + # The firmware reads the profile at the tip's height, counted from z_minimum + bottoms = [op.resource.get_location_wrt(deck, "c", "c", "cavity_bottom").z for op in ops] + distances = [ + None if surface_following_distances is None else surface_following_distances[i] + for i in range(len(ops)) + ] + segments = [ + container_segments[i] + if container_segments is not None + else _get_container_segments( + op.resource, + liquid_height=ctx.z_fluid[i] - bottoms[i], + piston_volume=ctx.volumes[i], + surface_following_distance=distances[i], + profile_start=ctx.z_minimum[i] - bottoms[i], ) - + for i, op in enumerate(ops) + ] + locations = [ + Coordinate( + x_position, + op.resource.get_location_wrt(deck, "c", "c", "cavity_bottom").y + op.offset.y, + ctx.z_fluid[i], + ) + for i, op in enumerate(ops) + ] volume_intents = [ VolumeTransferIntent( channel=ch, container=op.resource, tip=op.tip, - volume_ul=next(k.common.liquid_volume for k in kits if k.channel == self.channel_enum(ch)), + volume_ul=ctx.volumes[i], direction="aspirate", ) - for ch, op in zip(use_channels, ops) + for i, (ch, op) in enumerate(zip(use_channels, ops)) ] - queue_volume_transfers(volume_intents) + if not check_only: + queue_volume_transfers(volume_intents) aspirated = {ch: False for ch in use_channels} try: - await self._driver.send_command( - self._aspirate_command(kits, effective_lld, is_tadm, use_v2), - read_timeout=lld_read_timeout if effective_lld else None, + await self._aspirate_in_one_move( + use_channels, + locations, + [g.top_of_well for g in ctx.well_geometry], + ctx.z_minimum, + ctx.volumes, + # Without segments the firmware follows tube_radius, and 0 does not follow + tube_radii=[ + 0.0 if d == 0 else _effective_radius(op.resource) for d, op in zip(distances, ops) + ], + lld_mode=lld_mode, + clld_sensitivity=clld_sensitivity, + immersion_depths=None if immersion_depths is None else list(immersion_depths), + blow_out_air_volumes=[ + op.blow_out_air_volume + if op.blow_out_air_volume is not None + else (hlc.aspiration_blow_out_volume if hlc is not None else 0.0) + for op, hlc in zip(ops, classes) + ], + pre_wetting_volumes=fill_in_defaults( + pre_wetting_volumes, + [hlc.aspiration_over_aspirate_volume if hlc is not None else 0.0 for hlc in classes], + ), + pre_mixes=pre_mixes, + mix_positions_from_liquid_surface=mix_positions_from_liquid_surface, + flow_rates=[ + op.flow_rate + if op.flow_rate is not None + else (hlc.aspiration_flow_rate if hlc is not None else 100.0) + for op, hlc in zip(ops, classes) + ], + settling_times=fill_in_defaults( + settling_times, + [hlc.aspiration_settling_time if hlc is not None else 1.0 for hlc in classes], + ), + swap_speeds=fill_in_defaults( + swap_speeds, [hlc.aspiration_swap_speed if hlc is not None else 10.0 for hlc in classes] + ), + transport_air_volumes=fill_in_defaults( + transport_air_volumes, + [hlc.aspiration_air_transport_volume if hlc is not None else 0.0 for hlc in classes], + ), + minimum_traverse_height_end=minimum_traverse_height_end, + z_air=ctx.z_air, + z_bottom_search_offset=ctx.z_bottom_search_offset, + container_segments=segments, + lld=lld, + p_lld=p_lld, + clot_detection_heights=clot_detection_heights, + tadm=tadm, + read_timeout=read_timeout, + command_version=command_version, + check_only=check_only, ) aspirated = all_channels_succeeded(use_channels) except ChannelizedError as e: @@ -5436,7 +5714,303 @@ async def aspirate( raise finally: # What each channel took is what its tip now holds, and the well no longer does - finalize_volume_ops(volume_intents, aspirated) + if not check_only: + finalize_volume_ops(volume_intents, aspirated) + return ctx.volumes + + def _check_volumes_suffice( + self, containers: Sequence[Container], tips: Sequence[Tip], volumes: Sequence[float] + ) -> None: + """Refuse more than a container holds over all its jobs, or than a tip has room for. + + Args: + containers: per job. + tips: the tip each job fills, per job. + volumes: the liquid volume each job takes, in uL. + + Raises: + TooLittleLiquidError: If a container holds less than all its jobs take. + TooLittleVolumeError: If a tip has less room than its job takes. + """ + if not does_volume_tracking(): + return + asked: Dict[int, float] = {} + for container, volume in zip(containers, volumes): + asked[id(container)] = asked.get(id(container), 0.0) + volume + for container in {id(c): c for c in containers}.values(): + held = container.tracker.get_used_volume() + if not container.tracker.is_disabled and asked[id(container)] - held > 1e-6: + raise TooLittleLiquidError( + f"{container.name} holds {held} uL, {asked[id(container)]} uL asked for" + ) + for tip, volume in zip(tips, volumes): + room = tip.tracker.get_free_volume() + if not tip.tracker.is_disabled and volume - room > 1e-6: + raise TooLittleVolumeError(f"a tip with room for {room} uL asked to take {volume} uL") + + def _get_lld_modes( + self, lld_mode: Union[Pipettes.LLDMode, Sequence[Pipettes.LLDMode], None], n: int + ) -> Optional[List[Pipettes.LLDMode]]: + """One LLD mode per container, or None when none is given. + + Args: + lld_mode: one for all, or one per container. + n: how many containers. + + Raises: + ValueError: If a mode is not an `LLDMode`, the list is not one per container, or a pressure + mode is mixed with another mode. + """ + if lld_mode is None: + return None + if isinstance(lld_mode, self.LLDMode): + return [lld_mode] * n + if isinstance(lld_mode, str) or not isinstance(lld_mode, Sequence): + raise ValueError(f"lld_mode must be an LLDMode or one per container, is {lld_mode!r}") + modes = list(lld_mode) + if len(modes) != n: + raise ValueError(f"{len(modes)} lld modes for {n} containers") + for mode in modes: + if not isinstance(mode, self.LLDMode): + raise ValueError(f"lld_mode entries must be LLDMode, got {mode!r}") + pressure = {self.LLDMode.PRESSURE, self.LLDMode.DUAL} + if len(set(modes)) > 1 and pressure & set(modes): + raise ValueError(f"a pressure LLD mode cannot be mixed with another in one call: {modes}") + return modes + + async def aspirate( + self, + containers: Sequence[Container], + volumes: Optional[Sequence[float]] = None, + use_channels: Optional[List[int]] = None, + resource_offsets: Optional[List[Coordinate]] = None, + liquid_heights: Optional[Sequence[Optional[float]]] = None, + lld_mode: Union[Pipettes.LLDMode, Sequence[Pipettes.LLDMode], None] = None, + flow_rates: Optional[Sequence[Optional[float]]] = None, + *, + hamilton_liquid_classes: Optional[List[HamiltonLiquidClass]] = None, + piston_volumes: Optional[Sequence[float]] = None, + blow_out_air_volumes: Optional[Sequence[Optional[float]]] = None, + immersion_depths: Optional[Sequence[float]] = None, + minimum_allowed_z_positions_during: Optional[List[float]] = None, + pre_wetting_volumes: Optional[List[float]] = None, + pre_mixes: Optional[Sequence[Optional[Mix]]] = None, + mix_positions_from_liquid_surface: Optional[Sequence[float]] = None, + surface_following_distances: Optional[Sequence[float]] = None, + settling_times: Optional[List[float]] = None, + swap_speeds: Optional[List[float]] = None, + clot_detection_heights: Optional[Sequence[float]] = None, + transport_air_volumes: Optional[List[float]] = None, + minimum_traverse_height_start: Optional[float] = None, + minimum_traverse_height_during: Optional[float] = None, + minimum_traverse_height_end: Optional[float] = None, + x_grouping_tolerance: Optional[float] = None, + clld_sensitivity: Optional[int] = None, + lld: Optional[PrepCmd.LldParameters] = None, + p_lld: Optional[PrepCmd.PLldParameters] = None, + z_fluid: Optional[List[float]] = None, + z_bottom_search_offset: Optional[List[float]] = None, + z_air: Optional[List[float]] = None, + tadm: Optional[PrepCmd.TadmParameters] = None, + container_segments: Optional[List[List[PrepCmd.SegmentDescriptor]]] = None, + read_timeout: Optional[float] = None, + command_version: Optional[Literal["v1", "v2"]] = None, + ) -> None: + """Draw liquid from each container with a channel's tip, one command per batch. + + One channel per container. Containers within `x_grouping_tolerance` of one X share a batch; + the channels go to each batch in ascending X, and its volumes are booked when it aspirates. + + Args: + containers: one per channel used, at most as many as there are channels. + volumes: how much liquid to take from each container, in uL, corrected by the liquid class. + One of this and `piston_volumes`. + use_channels: which channels, 0-indexed from the back. The first len(containers) when None. + resource_offsets: added to where each channel goes in its container, in mm. The z shifts + the heights. Channels sharing a container spread across it in Y when None. + liquid_heights: where the liquid stands above each cavity bottom, in mm. None takes it from + the tracked volume, or, with an LLD mode, leaves it to the search. + lld_mode: how to search, one for all or one per container. None runs a search only when + `lld` is given. + flow_rates: in uL/s, per container. The liquid class's, else 100.0, when None. + hamilton_liquid_classes: the class for each container's volume. Looked up for the + channel's tip, water, when None. + piston_volumes: how much each piston draws, in uL, per container, as given, with no liquid + class. One of this and `volumes`. + blow_out_air_volumes: air drawn before the liquid, in uL, per container. The liquid + class's, else 0.0, when None. + immersion_depths: how far under the surface each tip aspirates, in mm, per container. + With LLD the search's `z_submerge`, 2.0 when None; without, off `z_fluid`, 0.0 when None. + minimum_allowed_z_positions_during: how low each tip bottom may go, in mm, per container. + The cavity bottom when None. + pre_wetting_volumes: drawn and returned first, in uL, per container. The liquid class's + over-aspirate volume, else 0.0, when None. + pre_mixes: a `Mix` per container, mixed before the draw, None for no mixing. Its + `surface_following_distance` is not sent. + mix_positions_from_liquid_surface: mixing depth under the aspirate height, in mm, per + container. 0.0 when None. + surface_following_distances: how far each tip follows the sinking surface, in mm, per + container: its profile scaled to that. None follows the profile as it is; 0 does not follow. + settling_times: how long the tip waits in the liquid, in s, per container. The liquid + class's, else 1.0, when None. + swap_speeds: how fast the tip leaves the liquid, in mm/s, per container. The liquid + class's, else 10.0, when None. + clot_detection_heights: how far a clot may hold each tip back, in mm, per container. 0.0 when + None; only 0.0 until the check is verified on the device. + transport_air_volumes: air drawn after the liquid, in uL, per container. The liquid + class's, else 0.0, when None. + minimum_traverse_height_start: the height every low channel's tip bottom is raised to before + the first batch, in mm. Z safety when None. + minimum_traverse_height_during: each tip bottom's height at the end of every batch but the + last, in mm. The traverse height less each tip's overhang when None, then Z safety. + minimum_traverse_height_end: the tip bottom height every tip is left at, in mm. The + traverse height less each tip's overhang when None. + x_grouping_tolerance: containers within this X distance share a batch, in mm. + `default_x_grouping_tolerance` when None. + clld_sensitivity: capacitive LLD sensitivity for every channel. 3 when None. + lld: the LLD search's start, speed and submerge depth. From the container's top when None. + p_lld: pressure LLD settings, seeking at 1 to 630 uL/s; needed for PRESSURE and DUAL. The + firmware's own when None. + z_fluid: the tip bottom height to aspirate at without LLD, in mm, per container. The cavity + bottom plus the liquid height when None. + z_bottom_search_offset: in mm, per container. 2.0 when None. + z_air: the tip bottom height above each container the tip leaves from, in mm. 2 mm over + the container's top when None. + tadm: TADM settings; given, the aspiration is monitored. + container_segments: each container's cross-sections, sent as they are, per container. None + builds them from each container's profile. + read_timeout: how long to wait for the answer, in s. Long enough for the search when an + LLD search runs and this is None. + command_version: "v1" or "v2" aspirate commands. What the firmware supports when None. + + Raises: + ValueError: If an argument is out of range, the lists do not match, a channel repeats, there + are more containers than channels, both or neither of `volumes` and `piston_volumes` are + given, a class is given with `piston_volumes`, no class is known for a channel's tip, a + mode is not an `LLDMode`, a pressure mode is mixed with another or has no `p_lld`. + RuntimeError: If a channel used carries no tip, or nothing knows where a container's + liquid stands: no height given, volume tracking off, no LLD. + TooLittleLiquidError: If a container holds less than it is asked for. + """ + containers = list(containers) + n = len(containers) + use_channels = use_channels if use_channels is not None else list(range(n)) + drawn = volumes if volumes is not None else piston_volumes + if drawn is None or (volumes is not None and piston_volumes is not None): + raise ValueError("give one of volumes and piston_volumes") + if piston_volumes is not None and hamilton_liquid_classes is not None: + raise ValueError("piston_volumes are sent as given; no liquid class applies") + if not containers: + raise ValueError("no containers to aspirate from") + if n > self.num_channels: + raise ValueError(f"{n} containers for {self.num_channels} channels, one channel each") + if len(use_channels) != n or len(set(use_channels)) != n: + raise ValueError(f"use_channels must name one distinct channel per container: {use_channels}") + per_container: Dict[str, Optional[Sequence[Any]]] = { + "volumes" if volumes is not None else "piston_volumes": drawn, + "resource_offsets": resource_offsets, + "liquid_heights": liquid_heights, + "flow_rates": flow_rates, + "hamilton_liquid_classes": hamilton_liquid_classes, + "immersion_depths": immersion_depths, + "blow_out_air_volumes": blow_out_air_volumes, + "pre_wetting_volumes": pre_wetting_volumes, + "clot_detection_heights": clot_detection_heights, + "z_fluid": z_fluid, + "minimum_allowed_z_positions_during": minimum_allowed_z_positions_during, + "z_bottom_search_offset": z_bottom_search_offset, + "pre_mixes": pre_mixes, + "mix_positions_from_liquid_surface": mix_positions_from_liquid_surface, + "settling_times": settling_times, + "swap_speeds": swap_speeds, + "transport_air_volumes": transport_air_volumes, + "z_air": z_air, + "container_segments": container_segments, + "surface_following_distances": surface_following_distances, + } + for name, values in per_container.items(): + if values is not None and len(values) != n: + raise ValueError(f"{name} length must match containers ({n})") + modes = self._get_lld_modes(lld_mode, n) + offsets = ( + resource_offsets + if resource_offsets is not None + else self._get_resource_offsets(containers, use_channels) + ) + # One command carries one LLD category, so each mode is planned on its own. + groups = ( + [list(range(n))] + if modes is None + else [[job for job in range(n) if modes[job] == mode] for mode in dict.fromkeys(modes)] + ) + batches: List[ChannelBatch] = [] + for group in groups: + _, planned = self._plan_batched( + self._require_deck(), + [containers[job] for job in group], + [use_channels[job] for job in group], + [offsets[job] for job in group], + x_grouping_tolerance, + minimum_traverse_height_end, + ) + batches.extend(replace(b, indices=[group[job] for job in b.indices]) for b in planned) + batches.sort(key=lambda b: b.x_position) + + def pick(values: Optional[Sequence[_T]], batch: ChannelBatch) -> Optional[List[_T]]: + """The batch's entries of a per-container list, None when it is None.""" + return None if values is None else [values[job] for job in batch.indices] + + async def aspirate_batch(batch: ChannelBatch, check_only: bool = False) -> List[float]: + """Aspirate the batch's containers in one command; the last leaves the tips at the end.""" + last = batch is batches[-1] + batch_modes = pick(modes, batch) + return await self._aspirate_batch( + batch.x_position, + [containers[job] for job in batch.indices], + [drawn[job] for job in batch.indices], + batch.channels, + [offsets[job] for job in batch.indices], + self._resolve_effective_lld(batch_modes, lld, len(batch.indices)), + piston_volumes is None, + pick(liquid_heights, batch), + None if batch_modes is None else batch_modes[0], + pick(flow_rates, batch), + hamilton_liquid_classes=pick(hamilton_liquid_classes, batch), + clld_sensitivity=clld_sensitivity, + immersion_depths=pick(immersion_depths, batch), + blow_out_air_volumes=pick(blow_out_air_volumes, batch), + pre_wetting_volumes=pick(pre_wetting_volumes, batch), + lld=lld, + p_lld=p_lld, + clot_detection_heights=pick(clot_detection_heights, batch), + z_fluid=pick(z_fluid, batch), + minimum_allowed_z_positions_during=pick(minimum_allowed_z_positions_during, batch), + z_bottom_search_offset=pick(z_bottom_search_offset, batch), + pre_mixes=pick(pre_mixes, batch), + mix_positions_from_liquid_surface=pick(mix_positions_from_liquid_surface, batch), + settling_times=pick(settling_times, batch), + swap_speeds=pick(swap_speeds, batch), + transport_air_volumes=pick(transport_air_volumes, batch), + z_air=pick(z_air, batch), + minimum_traverse_height_end=( + minimum_traverse_height_end if last else minimum_traverse_height_during + ), + tadm=tadm, + container_segments=pick(container_segments, batch), + surface_following_distances=pick(surface_following_distances, batch), + read_timeout=read_timeout, + command_version=command_version, + check_only=check_only, + ) + + # Every refusal the model can decide comes before the first command. + taken = [0.0] * n + for batch in batches: + for job, volume in zip(batch.indices, await aspirate_batch(batch, check_only=True)): + taken[job] = volume + self._check_volumes_suffice(containers, self._require_mounted_tips(use_channels), taken) + await self._check_tips_and_raise(use_channels, minimum_traverse_height_start) + await self._execute_batched(aspirate_batch, batches, minimum_traverse_height_during) async def dispense( self, diff --git a/pylabrobot/hamilton/prep/driver/features/pipettes_tests.py b/pylabrobot/hamilton/prep/driver/features/pipettes_tests.py index eee35d0e6f4..47df0281d45 100644 --- a/pylabrobot/hamilton/prep/driver/features/pipettes_tests.py +++ b/pylabrobot/hamilton/prep/driver/features/pipettes_tests.py @@ -4,7 +4,8 @@ import asyncio import functools -from typing import Any, List +import hashlib +from typing import Any, Dict, List, Optional, Sequence, Tuple from unittest.mock import AsyncMock, patch import pytest @@ -18,6 +19,7 @@ _get_profile_drop, ) from pylabrobot.hamilton.prep.driver.features.pipettes import logger as pipettes_logger +from pylabrobot.hamilton.prep.driver.master import _ResolvedPrepCommand from pylabrobot.hamilton.prep.driver.simulator import ( SIMULATED_X_AXIS_OFFSET, SIMULATED_X_SPEED, @@ -25,14 +27,23 @@ SIMULATED_Z_DRIVE_OFFSETS, _SimulatedIO, ) +from pylabrobot.hamilton.star.liquid_classes.mapping import ( + StandardVolumeFilter_Water_DispenseJet_Empty as _WATER_50, +) from pylabrobot.hamilton.transport.tcp.hoi_error import HoiError from pylabrobot.hamilton.transport.tcp.packets import Address from pylabrobot.hamilton.transport.tcp.wire_types import HcResultEntry +from pylabrobot.lib.liquid_handling.mix import Mix from pylabrobot.lib.liquid_handling.pipette_batch_scheduling import ChannelBatch from pylabrobot.resources import Container, Coordinate, PetriDish, Resource, Well from pylabrobot.resources.corning.axygen.plates import cor_axy_96_wellplate_500uL_Ub from pylabrobot.resources.corning.plates import cor_96_wellplate_360uL_Fb -from pylabrobot.resources.errors import HasTipError, NoTipError +from pylabrobot.resources.errors import ( + HasTipError, + NoTipError, + TooLittleLiquidError, + TooLittleVolumeError, +) from pylabrobot.resources.hamilton import ( PrepDeck, STARLetDeck, @@ -131,9 +142,9 @@ async def _t(): assert tip is not None and tip.parent is p.pipettes.shaft(i) await p.pipettes.aspirate( src, - vols=vols, + piston_volumes=vols, use_channels=use, - disable_volume_correction=[True] * 2, + liquid_heights=[1.0] * 2, ) for well in src: assert well.tracker.get_used_volume() == pytest.approx(80.0) @@ -166,6 +177,298 @@ async def _t(): _run(_t()) +_ASPIRATE_COMMANDS: Tuple[Any, ...] = tuple(Pipettes._ASPIRATE_CMD.values()) + + +def test_aspirate_takes_a_missing_liquid_height_from_the_tracked_volume(): + """No height given: the tip goes to where the tracked volume stands; untracked, it refuses.""" + + async def _t(): + set_tip_tracking(True) + set_volume_tracking(True) + try: + deck = PrepDeck() + tip_rack = deck[3] = hamilton_96_tiprack_50uL_NTR(name="ntr", with_tips=True) + plate = deck[0] = cor_96_wellplate_360uL_Fb(name="plate") + p = PrepSimulationDriver(deck=deck) + await p.setup() + assert p.pipettes is not None + well = plate.get_item("A1") + well.tracker.set_volume(100.0) + await p.pipettes.pick_up_tips([tip_rack.get_item("A1")], use_channels=[0]) + sent = _record(p) + await p.pipettes.aspirate([well], piston_volumes=[20.0], use_channels=[0]) + entry = next(c for c in sent if isinstance(c, _ASPIRATE_COMMANDS)).aspirate_parameters[0] + bottom = well.get_location_wrt(deck, "c", "c", "cavity_bottom").z + height = well.compute_height_from_volume(100.0) + assert entry.no_lld.z_fluid == pytest.approx(bottom + height, abs=0.01) + + set_volume_tracking(False) + with pytest.raises(RuntimeError, match="nothing knows where its liquid is"): + await p.pipettes.aspirate([well], piston_volumes=[20.0], use_channels=[0]) + await p.stop() + finally: + set_tip_tracking(False) + set_volume_tracking(False) + + _run(_t()) + + +def test_aspirate_immerses_the_tip_below_the_surface_with_and_without_lld(): + """Without LLD the aspirate height drops by the depth; with LLD the depth is the search's submerge.""" + + async def _t(): + deck = PrepDeck() + tip_rack = deck[3] = hamilton_96_tiprack_50uL_NTR(name="ntr", with_tips=True) + plate = deck[0] = cor_96_wellplate_360uL_Fb(name="plate") + p = PrepSimulationDriver(deck=deck) + await p.setup() + assert p.pipettes is not None + well = plate.get_item("A1") + bottom = well.get_location_wrt(deck, "c", "c", "cavity_bottom").z + await p.pipettes.pick_up_tips([tip_rack.get_item("A1")], use_channels=[0]) + sent = _record(p) + await p.pipettes.aspirate( + [well], piston_volumes=[0.0], use_channels=[0], liquid_heights=[5.0], immersion_depths=[2.0] + ) + await p.pipettes.aspirate( + [well], + piston_volumes=[0.0], + use_channels=[0], + lld_mode=Pipettes.LLDMode.CAPACITIVE, + immersion_depths=[1.5], + ) + plain, searched = [c.aspirate_parameters[0] for c in sent if isinstance(c, _ASPIRATE_COMMANDS)] + assert plain.no_lld.z_fluid == pytest.approx(bottom + 3.0, abs=0.01) + assert searched.lld.z_submerge == pytest.approx(1.5) + await p.stop() + + _run(_t()) + + +def test_aspirate_takes_volumes_by_a_liquid_class_or_piston_volumes_as_given(): + """One of volumes and piston_volumes; a tip without a class is refused before anything moves.""" + + async def _t(): + deck = PrepDeck() + tip_rack = deck[3] = hamilton_96_tiprack_50uL_NTR(name="ntr", with_tips=True) + plate = deck[0] = cor_96_wellplate_360uL_Fb(name="plate") + p = PrepSimulationDriver(deck=deck) + await p.setup() + assert p.pipettes is not None + well = plate.get_item("A1") + await p.pipettes.pick_up_tips([tip_rack.get_item("A1")], use_channels=[0]) + for kwargs, match in ( + ({}, "one of volumes and piston_volumes"), + ({"volumes": [5.0], "piston_volumes": [5.0]}, "one of volumes and piston_volumes"), + ({"piston_volumes": [5.0], "hamilton_liquid_classes": [None]}, "no liquid class applies"), + ({"volumes": [5.0]}, "no liquid class for"), + ): + with pytest.raises(ValueError, match=match): + await p.pipettes.aspirate([well], use_channels=[0], liquid_heights=[2.0], **kwargs) + sent = _record(p) + await p.pipettes.aspirate([well], piston_volumes=[5.0], use_channels=[0], liquid_heights=[2.0]) + entry = next(c for c in sent if isinstance(c, _ASPIRATE_COMMANDS)).aspirate_parameters[0] + assert entry.common.liquid_volume == pytest.approx(5.0) + await p.stop() + + _run(_t()) + + +def test_aspirate_sends_an_explicit_zero_blow_out_and_refuses_a_zero_flow_rate(): + """A given 0.0 blow-out is sent over the class's; a flow rate of 0.0 is refused before sending.""" + + from pylabrobot.hamilton.star.liquid_classes.mapping import ( + StandardVolumeFilter_Water_DispenseJet_Empty as water, + ) + + async def _t(): + deck = PrepDeck() + tip_rack = deck[3] = hamilton_96_tiprack_50uL_NTR(name="ntr", with_tips=True) + plate = deck[0] = cor_96_wellplate_360uL_Fb(name="plate") + p = PrepSimulationDriver(deck=deck) + await p.setup() + assert p.pipettes is not None + well = plate.get_item("A1") + await p.pipettes.pick_up_tips([tip_rack.get_item("A1")], use_channels=[0]) + assert water.aspiration_blow_out_volume > 0 + classes: Dict[str, Any] = { + "hamilton_liquid_classes": [water], + "liquid_heights": [2.0], + "use_channels": [0], + } + sent = _record(p) + await p.pipettes.aspirate([well], volumes=[5.0], blow_out_air_volumes=[0.0], **classes) + entry = next(c for c in sent if isinstance(c, _ASPIRATE_COMMANDS)).aspirate_parameters[0] + assert entry.aspirate.blowout_volume == 0.0 + sent.clear() + with pytest.raises(ValueError, match="flow_rates must be above 0"): + await p.pipettes.aspirate([well], volumes=[5.0], flow_rates=[0.0], **classes) + assert not any(isinstance(c, _ASPIRATE_COMMANDS) for c in sent) + await p.stop() + + _run(_t()) + + +def test_a_tip_holding_liquid_is_dropped_with_a_warning(): + """What a tip still holds is the caller's to decide: it is dropped, and the log says so.""" + + async def _t(): + set_tip_tracking(True) + set_volume_tracking(True) + try: + deck = PrepDeck() + tip_rack = deck[3] = hamilton_96_tiprack_50uL_NTR(name="ntr", with_tips=True) + plate = deck[0] = cor_96_wellplate_360uL_Fb(name="plate") + p = PrepSimulationDriver(deck=deck) + await p.setup() + assert p.pipettes is not None + well, spot = plate.get_item("A1"), tip_rack.get_item("A1") + well.tracker.set_volume(100.0) + await p.pipettes.pick_up_tips([spot], use_channels=[0]) + await p.pipettes.aspirate([well], piston_volumes=[5.0], use_channels=[0]) + with patch.object(pipettes_logger, "warning") as warning: + await p.pipettes.return_tips() + assert p.pipettes.get_mounted_tip(0) is None and spot.has_tip() + assert "with 5.0 uL in it" in warning.call_args.args[0] % warning.call_args.args[1:] + await p.stop() + finally: + set_volume_tracking(False) + set_tip_tracking(False) + + _run(_t()) + + +def test_aspirate_refuses_a_clot_check_until_it_is_verified(): + """0.0 sends the capacitive block as before; any other height is refused before sending.""" + + async def _t(): + deck = PrepDeck() + tip_rack = deck[3] = hamilton_96_tiprack_50uL_NTR(name="ntr", with_tips=True) + plate = deck[0] = cor_96_wellplate_360uL_Fb(name="plate") + p = PrepSimulationDriver(deck=deck) + await p.setup() + assert p.pipettes is not None + await p.pipettes.pick_up_tips([tip_rack.get_item("A1")], use_channels=[0]) + well = plate.get_item("A1") + kwargs: Dict[str, Any] = { + "piston_volumes": [5.0], + "use_channels": [0], + "lld_mode": Pipettes.LLDMode.CAPACITIVE, + } + sent = _record(p) + await p.pipettes.aspirate([well], clot_detection_heights=[0.0], **kwargs) + c_lld = next(c for c in sent if isinstance(c, _ASPIRATE_COMMANDS)).aspirate_parameters[0].c_lld + assert (c_lld.clot_check_enable, c_lld.z_clot_check) == (False, 0.0) + sent.clear() + with pytest.raises(ValueError, match="clot detection is not verified"): + await p.pipettes.aspirate([well], clot_detection_heights=[1.5], **kwargs) + assert not any(isinstance(c, _ASPIRATE_COMMANDS) for c in sent) + await p.stop() + + _run(_t()) + + +def test_aspirate_sends_the_clld_sensitivity_it_is_given(): + """clld_sensitivity replaces only the sensitivity in the capacitive LLD block.""" + + async def _t(): + deck = PrepDeck() + tip_rack = deck[3] = hamilton_96_tiprack_50uL_NTR(name="ntr", with_tips=True) + plate = deck[0] = cor_96_wellplate_360uL_Fb(name="plate") + p = PrepSimulationDriver(deck=deck) + await p.setup() + assert p.pipettes is not None + await p.pipettes.pick_up_tips([tip_rack.get_item("A1")], use_channels=[0]) + sent = _record(p) + await p.pipettes.aspirate( + [plate.get_item("A1")], + piston_volumes=[0.0], + use_channels=[0], + lld_mode=Pipettes.LLDMode.CAPACITIVE, + clld_sensitivity=2, + ) + c_lld = next(c for c in sent if isinstance(c, _ASPIRATE_COMMANDS)).aspirate_parameters[0].c_lld + assert (c_lld.default_values, c_lld.sensitivity, c_lld.detect_mode) == (False, 2, 0) + await p.stop() + + _run(_t()) + + +def test_channels_sharing_a_container_spread_across_it(): + """Two channels into one dish go either side of its centre, at least their spacing apart.""" + + async def _t(): + deck = PrepDeck() + tip_rack = deck[3] = hamilton_96_tiprack_50uL_NTR(name="ntr", with_tips=True) + dish = PetriDish(name="dish", diameter=77.0, height=30.0, material_z_thickness=2.0) + deck[6].assign_child_by_anchor( + dish, parent_anchor=("c", "c", "t"), child_anchor=("c", "c", "b") + ) + p = PrepSimulationDriver(deck=deck) + await p.setup() + assert p.pipettes is not None + await p.pipettes.pick_up_tips(tip_rack["A1:B1"], use_channels=[0, 1]) + sent = _record(p) + await p.pipettes.aspirate( + [dish] * 2, piston_volumes=[5.0] * 2, use_channels=[0, 1], liquid_heights=[5.0] * 2 + ) + rear, front = [e.aspirate.y_position for e in sent[-1].aspirate_parameters] + centre = dish.get_location_wrt(deck, "c", "c", "c").y + assert rear - centre == pytest.approx(centre - front, abs=0.01) + assert rear - front >= 9.0 + await p.stop() + + _run(_t()) + + +def test_aspirate_sends_a_mix_per_container_and_the_default_block_where_none(): + """A `Mix` becomes that channel's mix block; a container without one is sent the default.""" + + async def _t(): + deck = PrepDeck() + tip_rack = deck[3] = hamilton_96_tiprack_50uL_NTR(name="ntr", with_tips=True) + plate = deck[0] = cor_96_wellplate_360uL_Fb(name="plate") + p = PrepSimulationDriver(deck=deck) + await p.setup() + assert p.pipettes is not None + await p.pipettes.pick_up_tips(tip_rack["A1:B1"], use_channels=[0, 1]) + sent = _record(p) + await p.pipettes.aspirate( + plate["A1:B1"], + piston_volumes=[10.0, 10.0], + use_channels=[0, 1], + liquid_heights=[2.0, 2.0], + pre_mixes=[Mix(volume=30.0, repetitions=3, flow_rate=50.0), None], + mix_positions_from_liquid_surface=[1.5, 1.5], + ) + entries = next(c for c in sent if isinstance(c, _ASPIRATE_COMMANDS)).aspirate_parameters + mixed, plain = entries + assert (mixed.mix.default_values, mixed.mix.volume, mixed.mix.cycles) == (False, 30.0, 3) + assert (mixed.mix.speed, mixed.mix.z_offset) == (50.0, 1.5) + assert plain.mix == PrepCmd.MixParameters.default() + + with pytest.raises(ValueError, match="pre_mixes length"): + await p.pipettes.aspirate( + plate["A1:B1"], + piston_volumes=[10.0, 10.0], + use_channels=[0, 1], + liquid_heights=[2.0, 2.0], + pre_mixes=[None], + ) + with pytest.raises(ValueError, match="mix_positions_from_liquid_surface length"): + await p.pipettes.aspirate( + plate["A1:B1"], + piston_volumes=[10.0, 10.0], + use_channels=[0, 1], + liquid_heights=[2.0, 2.0], + mix_positions_from_liquid_surface=[1.5], + ) + await p.stop() + + _run(_t()) + + def test_default_minimum_traverse_height_is_plrs_then_the_devices_then_the_one_setup_is_given(): async def _t(): p = PrepSimulationDriver(deck=STARLetDeck(), default_minimum_traverse_height=160.0) @@ -2548,3 +2851,515 @@ async def _t(): await p.stop() _run(_t()) + + +def _aspirate_following_setup(): + """A simulated Prep with a tip on channel 0 and a Corning 3603 well of 200 uL.""" + deck = PrepDeck() + tip_rack = deck[3] = hamilton_96_tiprack_50uL_NTR(name="ntr", with_tips=True) + plate = deck[0] = cor_96_wellplate_360uL_Fb(name="plate") + return PrepSimulationDriver(deck=deck), tip_rack, plate + + +def test_aspirate_sends_the_containers_profile_from_z_minimum(): + """By default the segments are the well's own profile, cut to begin at z_minimum.""" + + async def _t(): + p, tip_rack, plate = _aspirate_following_setup() + await p.setup() + assert p.pipettes is not None + well = plate.get_item("A1") + await p.pipettes.pick_up_tips([tip_rack.get_item("A1")], use_channels=[0]) + sent = _record(p) + await p.pipettes.aspirate( + [well], + piston_volumes=[20.0], + use_channels=[0], + liquid_heights=[3.0], + minimum_allowed_z_positions_during=[ + well.get_location_wrt(p.deck, "c", "c", "cavity_bottom").z + 1.0 + ], + ) + entry = next(c for c in sent if isinstance(c, _ASPIRATE_COMMANDS)).aspirate_parameters[0] + expected = _get_container_segments(well, profile_start=1.0) + assert [(s.area_bottom, s.height) for s in entry.container_description] == [ + (pytest.approx(s.area_bottom), pytest.approx(s.height)) for s in expected + ] + await p.stop() + + _run(_t()) + + +def test_aspirate_scales_the_profile_to_a_surface_following_distance(): + """A distance scales the profile so the tip sinks exactly that; 0 sends none and tube_radius 0.""" + + async def _t(): + p, tip_rack, plate = _aspirate_following_setup() + await p.setup() + assert p.pipettes is not None + well = plate.get_item("A1") + await p.pipettes.pick_up_tips([tip_rack.get_item("A1")], use_channels=[0]) + sent = _record(p) + await p.pipettes.aspirate( + [well], + piston_volumes=[20.0], + use_channels=[0], + liquid_heights=[3.0], + surface_following_distances=[0.5], + ) + entry = next(c for c in sent if isinstance(c, _ASPIRATE_COMMANDS)).aspirate_parameters[0] + assert _get_profile_drop(entry.container_description, 3.0, 20.0) == pytest.approx(0.5, abs=1e-3) + await p.pipettes.dispense( + [well], vols=[20.0], use_channels=[0], liquid_height=[3.0], disable_volume_correction=[True] + ) + sent.clear() + await p.pipettes.aspirate( + [well], + piston_volumes=[20.0], + use_channels=[0], + liquid_heights=[3.0], + surface_following_distances=[0.0], + ) + entry = next(c for c in sent if isinstance(c, _ASPIRATE_COMMANDS)).aspirate_parameters[0] + assert entry.container_description == [] + assert entry.common.tube_radius == 0.0 + with pytest.raises(ValueError, match="surface_following_distances length"): + await p.pipettes.aspirate( + [well], + piston_volumes=[20.0], + use_channels=[0], + liquid_heights=[3.0], + surface_following_distances=[0.5, 0.5], + ) + await p.stop() + + _run(_t()) + + +_ASPIRATE_TWO = {"piston_volumes": [20.0, 30.0], "liquid_heights": [3.0, 4.0]} +_ASPIRATE_SEGMENTS = [ + [ + PrepCmd.SegmentDescriptor(area_top=30.0, area_bottom=10.0, height=4.0), + PrepCmd.SegmentDescriptor(area_top=30.0, area_bottom=30.0, height=6.0), + ], + [], +] + +# name: (tip volume, where, use_channels, keyword arguments) +_GOLDEN_ASPIRATE_CALLS: Dict[str, Tuple[int, str, Optional[List[int]], Dict[str, Any]]] = { + "piston volumes, heights": (300, "A1:B1", [0, 1], _ASPIRATE_TWO), + "class volumes, tracked heights": (300, "A1:B1", [0, 1], {"volumes": [20.0, 30.0]}), + "50 uL tips, classes given": ( + 50, + "A1:B1", + [0, 1], + {"volumes": [10.0, 20.0], "hamilton_liquid_classes": [_WATER_50, _WATER_50]}, + ), + "channel 0 alone": (300, "A1", [0], {"piston_volumes": [20.0], "liquid_heights": [3.0]}), + "channel 1 alone": (300, "B1", [1], {"piston_volumes": [20.0], "liquid_heights": [3.0]}), + "channels given front first": (300, "B1,A1", [1, 0], _ASPIRATE_TWO), + "use_channels None": (300, "A1:B1", None, _ASPIRATE_TWO), + "v1": (300, "A1:B1", [0, 1], {**_ASPIRATE_TWO, "command_version": "v1"}), + "v2 given": (300, "A1:B1", [0, 1], {**_ASPIRATE_TWO, "command_version": "v2"}), + "capacitive LLD": ( + 300, + "A1:B1", + [0, 1], + {"volumes": [10.0, 10.0], "lld_mode": Pipettes.LLDMode.CAPACITIVE}, + ), + "capacitive LLD, one per container": ( + 300, + "A1:B1", + [0, 1], + {"volumes": [10.0, 10.0], "lld_mode": [Pipettes.LLDMode.CAPACITIVE] * 2}, + ), + "capacitive LLD, v1, sensitivity, immersion": ( + 300, + "A1:B1", + [0, 1], + { + "volumes": [10.0, 10.0], + "lld_mode": Pipettes.LLDMode.CAPACITIVE, + "command_version": "v1", + "clld_sensitivity": 2, + "immersion_depths": [1.0, 1.5], + }, + ), + "capacitive LLD, read_timeout given": ( + 300, + "A1:B1", + [0, 1], + {"volumes": [10.0, 10.0], "lld_mode": Pipettes.LLDMode.CAPACITIVE, "read_timeout": 30.0}, + ), + "mix, immersion": ( + 300, + "A1:B1", + [0, 1], + { + **_ASPIRATE_TWO, + "immersion_depths": [1.0, 1.0], + "pre_mixes": [Mix(20.0, 2, 40.0), None], + "mix_positions_from_liquid_surface": [1.0, 1.0], + }, + ), + "tadm, end height, z_air": ( + 300, + "A1:B1", + [0, 1], + { + **_ASPIRATE_TWO, + "tadm": PrepCmd.TadmParameters( + default_values=False, + limit_curve_index=1, + recording_mode=PrepCmd.TadmRecordingModes.Errors, + ), + "minimum_traverse_height_end": 110.0, + "z_air": [60.0, 61.0], + }, + ), + "z values given, v1": ( + 300, + "A1:B1", + [0, 1], + { + **_ASPIRATE_TWO, + "command_version": "v1", + "z_fluid": [10.0, 11.0], + "minimum_allowed_z_positions_during": [5.0, 6.0], + "z_bottom_search_offset": [1.0, 1.5], + }, + ), + "times, speeds, volumes given": ( + 300, + "A1:B1", + [0, 1], + { + "volumes": [20.0, 20.0], + "liquid_heights": [3.0, 4.0], + "flow_rates": [50.0, None], + "settling_times": [0.5, 0.5], + "swap_speeds": [20.0, 20.0], + "transport_air_volumes": [5.0, 5.0], + "pre_wetting_volumes": [2.0, 2.0], + "blow_out_air_volumes": [3.0, None], + }, + ), + "offsets of one X": ( + 300, + "A1:B1", + [0, 1], + {**_ASPIRATE_TWO, "resource_offsets": [Coordinate(0.5, 0.5, 1.0), Coordinate(0.5, 0, 0)]}, + ), + "channels sharing a dish": ( + 300, + "dish", + [0, 1], + {"piston_volumes": [20.0, 30.0], "liquid_heights": [5.0, 5.0]}, + ), + "container segments given": ( + 300, + "A1:B1", + [0, 1], + {**_ASPIRATE_TWO, "container_segments": _ASPIRATE_SEGMENTS}, + ), + "surface following distances": ( + 300, + "A1:B1", + [0, 1], + {**_ASPIRATE_TWO, "surface_following_distances": [0.0, 2.0]}, + ), +} + + +async def _capture_aspirate(name: str) -> List[str]: + """Each aspirate frame a golden call sends: its read timeout, then its bytes as hex. + + The frame is built with source 2.1.65535 and sequence 0. Volume tracking is on, 200 uL a well. + """ + tip_volume, where, use_channels, kwargs = _GOLDEN_ASPIRATE_CALLS[name] + set_volume_tracking(True) + try: + deck = PrepDeck() + tip_rack = hamilton_96_tiprack_300uL_NTR if tip_volume == 300 else hamilton_96_tiprack_50uL_NTR + rack = deck[1] = tip_rack(name="tips", with_tips=True) + plate = deck[0] = cor_96_wellplate_360uL_Fb(name="plate") + dish = PetriDish(name="dish", diameter=77.0, height=30.0, material_z_thickness=2.0) + deck[6].assign_child_by_anchor( + dish, parent_anchor=("c", "c", "t"), child_anchor=("c", "c", "b") + ) + for well in plate.get_all_items(): + well.tracker.set_volume(200.0) + dish.tracker.set_volume(20000.0) + containers: List[Container] = ( + [dish, dish] if where == "dish" else [w for part in where.split(",") for w in plate[part]] + ) + channels = sorted(use_channels if use_channels is not None else range(len(containers))) + p = PrepSimulationDriver(deck=deck) + await p.setup() + assert p.pipettes is not None + await p.pipettes.pick_up_tips([rack.get_item(c) for c in channels], use_channels=channels) + session = p.io._session + exchange = session.exchange + frames: List[str] = [] + + async def record(command, *args, **kw): + request = command.request if isinstance(command, _ResolvedPrepCommand) else command + if isinstance(request, _ASPIRATE_COMMANDS): + frames.append(f"{kw.get('read_timeout')} {command.build(Address(2, 1, 65535), 0).hex()}") + return await exchange(command, *args, **kw) + + session.exchange = record # type: ignore[method-assign] + await p.pipettes.aspirate(containers, use_channels=use_channels, **kwargs) + finally: + set_volume_tracking(False) + return frames + + +# name: each aspirate frame's read timeout, then the SHA-256 of its bytes +_GOLDEN_ASPIRATE_FRAMES: Dict[str, List[str]] = { + "piston volumes, heights": [ + "60.0 4a9e16bf10ef7f31d6d6e7b1312115061c5c6950d4ee9205e6c6c774bd4dd740", + ], + "class volumes, tracked heights": [ + "60.0 38d6bfbc3a9e4d36258e6a98791191f1dfbc44d52b4d9ac69a6cbdb5ae556b6e", + ], + "50 uL tips, classes given": [ + "60.0 f8fe364b8b35b10c77daa56cf99db9fe1ad5bab2366339f425f8c4efb7135ae3", + ], + "channel 0 alone": ["60.0 e5cdcb46ec932553d7901c07a227d03769424d7c694e3df42d14aeea7540b68a"], + "channel 1 alone": ["60.0 fd6b68d105f381b4e96a8423d7509b9e122c251ca112def419e9e27cae4578f7"], + "channels given front first": [ + "60.0 b5c6148899a435077a8ab70d22581417679a4865b7fabe1a9f881a3b8a09cd60", + ], + "use_channels None": ["60.0 4a9e16bf10ef7f31d6d6e7b1312115061c5c6950d4ee9205e6c6c774bd4dd740"], + "v1": ["60.0 ee7d3df414101f7df8178f8979539534afabc3d89e7af8cbe1a1749391d16071"], + "v2 given": ["60.0 4a9e16bf10ef7f31d6d6e7b1312115061c5c6950d4ee9205e6c6c774bd4dd740"], + "capacitive LLD": ["62.094 f26ce6d307a03e855b6c46edb53e2c82ec87d7ed83efafd48d517a2c6e77b4ef"], + "capacitive LLD, one per container": [ + "62.094 f26ce6d307a03e855b6c46edb53e2c82ec87d7ed83efafd48d517a2c6e77b4ef", + ], + "capacitive LLD, v1, sensitivity, immersion": [ + "62.094 48befe0ba080ccf5e048dfe4549889bf71683f06e26ae81ae27e2ff9b5ca019f", + ], + "capacitive LLD, read_timeout given": [ + "30.0 f26ce6d307a03e855b6c46edb53e2c82ec87d7ed83efafd48d517a2c6e77b4ef", + ], + "mix, immersion": ["60.0 920c42c02c8749b7b00d022c097838e57086103dadcb94ecaf967e990c83cbda"], + "tadm, end height, z_air": [ + "60.0 bfad748ab52d3c51608b0985da1030ce83a3107349303a9648cd3abc579af733", + ], + "z values given, v1": ["60.0 d9fe620d4ba996942e435c6977876e05bc26acf4a5d388604e92e619e910ba14"], + "times, speeds, volumes given": [ + "60.0 da7e1372a80c75a5fecbc12a41113b74712b5bcf83622c69f4dc5d01886ab66c", + ], + "offsets of one X": ["60.0 5d7b5dc2a0df517e456fda99f4c3391d762d8c9c18ec585afef53f6bf2fc051b"], + "channels sharing a dish": [ + "60.0 dffe23c5435a3ee76833db413d916081ae356229328b223b71cff8df56572891", + ], + "container segments given": [ + "60.0 f5f9d38f58411c840e3f5b558c7f667d31a5e258b5e3792ebabef4a37fc9d57f", + ], + "surface following distances": [ + "60.0 5763eca12d07f7f4c0a21d83b7608adaa1bc5fe3c5b2a6caa0818b54576b50d9", + ], +} + + +def _digest(frames: List[str]) -> List[str]: + """Each frame's read timeout, then the SHA-256 of its bytes.""" + digests = [] + for frame in frames: + timeout, hexed = frame.split() + digests.append(f"{timeout} {hashlib.sha256(bytes.fromhex(hexed)).hexdigest()}") + return digests + + +def test_aspirate_sends_the_golden_frames(): + """Every golden aspirate call sends exactly the frames and read timeouts recorded for it.""" + assert list(_GOLDEN_ASPIRATE_FRAMES) == list(_GOLDEN_ASPIRATE_CALLS) + for name, frames in _GOLDEN_ASPIRATE_FRAMES.items(): + assert _digest(asyncio.run(_capture_aspirate(name))) == frames, name + + +def test_aspirate_runs_one_command_per_x_and_books_each_batch_on_its_own(): + """Two X: a move before one command per X; each batch booked on its own, a failed one undone.""" + + async def _t(): + set_volume_tracking(True) + try: + deck = PrepDeck() + rack = deck[1] = hamilton_96_tiprack_300uL_NTR(name="tips", with_tips=True) + plate = deck[0] = cor_96_wellplate_360uL_Fb(name="plate") + for well in plate.get_all_items(): + well.tracker.set_volume(200.0) + p = PrepSimulationDriver(deck=deck) + await p.setup() + assert p.pipettes is not None + await p.pipettes.pick_up_tips(rack["A1:B1"], use_channels=[0, 1]) + tips = [p.pipettes.get_mounted_tip(ch) for ch in (0, 1)] + sent: List[Any] = [] + at_second: List[float] = [] + fail_second = False + send = p.send_command + + async def record(command, *args, **kwargs): + if isinstance(command, _ASPIRATE_COMMANDS) and any( + isinstance(c, _ASPIRATE_COMMANDS) for c in sent + ): + at_second.extend([rear.tracker.volume, front.tracker.volume]) + at_second.append(front.tracker.pending_volume) + if fail_second: + sent.append(command) + raise RuntimeError("refused") + sent.append(command) + return await send(command, *args, **kwargs) + + p.send_command = record # type: ignore[method-assign] + rear, front = plate.get_item("A1"), plate.get_item("B2") + heights: Dict[str, Any] = {"piston_volumes": [10.0, 20.0], "liquid_heights": [3.0, 3.0]} + await p.pipettes.aspirate( + [rear, front], + use_channels=[0, 1], + minimum_traverse_height_during=100.0, + minimum_traverse_height_end=110.0, + **heights, + ) + first, second = [i for i, c in enumerate(sent) if isinstance(c, _ASPIRATE_COMMANDS)] + for index, well, z_final in ((first, rear, 100.0), (second, front, 110.0)): + (entry,) = sent[index].aspirate_parameters + x = well.get_location_wrt(deck, "c", "c", "cavity_bottom").x + assert entry.aspirate.x_position == pytest.approx(x) + assert entry.common.z_final == z_final + moves = [i for i, c in enumerate(sent) if isinstance(c, PrepCmd.PrepMoveToPosition)] + assert len(moves) == 2 and moves[0] < first < moves[1] < second + # At the second command the first batch is committed and the second only pending. + assert at_second == [190.0, 200.0, 180.0] + assert (rear.tracker.volume, front.tracker.volume) == (190.0, 180.0) + + rear, front = plate.get_item("A5"), plate.get_item("B6") + sent.clear() + at_second.clear() + fail_second = True + with pytest.raises(RuntimeError, match="refused"): + await p.pipettes.aspirate([rear, front], use_channels=[0, 1], **heights) + assert (rear.tracker.volume, front.tracker.volume) == (190.0, 200.0) + assert front.tracker.pending_volume == 200.0 + assert [tip.tracker.get_used_volume() for tip in tips if tip is not None] == [20.0, 20.0] + failed = max(i for i, c in enumerate(sent) if isinstance(c, _ASPIRATE_COMMANDS)) + assert any(isinstance(c, PrepCmd.PrepMoveZUpToSafe) for c in sent[failed + 1 :]) + finally: + set_volume_tracking(False) + + _run(_t()) + + +def test_aspirate_refuses_what_the_model_decides_before_any_command(): + """Each refusal the model can decide is raised with no command sent, not a tip sense or a move.""" + + async def _t(): + set_volume_tracking(True) + try: + deck = PrepDeck() + rack_300 = deck[1] = hamilton_96_tiprack_300uL_NTR(name="tips_300", with_tips=True) + rack_50 = deck[3] = hamilton_96_tiprack_50uL_NTR(name="tips_50", with_tips=True) + plate = deck[0] = cor_96_wellplate_360uL_Fb(name="plate") + dish = PetriDish(name="dish", diameter=77.0, height=30.0, material_z_thickness=2.0) + deck[6].assign_child_by_anchor( + dish, parent_anchor=("c", "c", "t"), child_anchor=("c", "c", "b") + ) + for well in plate.get_all_items(): + well.tracker.set_volume(200.0) + dish.tracker.set_volume(180.0) + p = PrepSimulationDriver(deck=deck) + await p.setup() + assert p.pipettes is not None + await p.pipettes.pick_up_tips([rack_300.get_item("A1")], use_channels=[0]) + two = plate["A1:B1"] + one: Dict[str, Any] = {"use_channels": [0], "piston_volumes": [10.0], "liquid_heights": [3.0]} + both: Dict[str, Any] = { + "use_channels": [0, 1], + "piston_volumes": [10.0, 10.0], + "liquid_heights": [3.0] * 2, + } + apart = [Coordinate(-5.0, 0, 0), Coordinate(5.0, 0, 0)] + capacitive, pressure = Pipettes.LLDMode.CAPACITIVE, Pipettes.LLDMode.PRESSURE + seek_0 = PrepCmd.PLldParameters( + default_values=False, + sensitivity=1, + dispenser_seek_speed=0.0, + lld_height_difference=0.0, + detect_mode=0, + ) + sent = _record(p) + refusals: List[Tuple[Any, str, Sequence[Container], Dict[str, Any]]] = [ + (NoTipError, "no tip is mounted", two, both), + ] + for error, match, containers, kwargs in refusals: + with pytest.raises(error, match=match): + await p.pipettes.aspirate(containers, **kwargs) + assert sent == [], match + await p.pipettes.pick_up_tips([rack_50.get_item("B1")], use_channels=[1]) + sent.clear() + refusals = [ + (ValueError, "no liquid class", two, {"use_channels": [0, 1], "volumes": [10.0, 10.0]}), + (ValueError, "one distinct channel", two, {**both, "use_channels": [0, 0]}), + (ValueError, "3 containers for 2 channels", plate["A1:C1"], {"piston_volumes": [1.0] * 3}), + (ValueError, "flow_rates length", plate["A1"], {**one, "flow_rates": [50.0, 50.0]}), + (ValueError, "flow_rates must be above 0", plate["A1"], {**one, "flow_rates": [0.0]}), + (ValueError, "clot detection", plate["A1"], {**one, "clot_detection_heights": [1.5]}), + (ValueError, "outside channel", two, {**both, "minimum_traverse_height_end": 200.0}), + ( + TooLittleLiquidError, + "dish holds 180.0 uL, 190.0 uL asked for", + [dish, dish], + {**both, "piston_volumes": [150.0, 40.0], "resource_offsets": apart}, + ), + (TooLittleVolumeError, "room for", two, {**both, "piston_volumes": [10.0, 70.0]}), + (ValueError, "PRESSURE LLD needs p_lld", two, {**both, "lld_mode": pressure}), + (ValueError, "1 to 630", two, {**both, "lld_mode": capacitive, "p_lld": seek_0}), + (ValueError, "must be LLDMode", two, {**both, "lld_mode": [capacitive, "capacitive"]}), + (ValueError, "1 lld modes for 2", two, {**both, "lld_mode": [capacitive]}), + (ValueError, "cannot be mixed", two, {**both, "lld_mode": [capacitive, pressure]}), + ] + for error, match, containers, kwargs in refusals: + with pytest.raises(error, match=match): + await p.pipettes.aspirate(containers, **kwargs) + assert sent == [], match + set_volume_tracking(False) + with pytest.raises(RuntimeError, match="nothing knows where its liquid is"): + await p.pipettes.aspirate(two, use_channels=[0, 1], piston_volumes=[10.0, 10.0]) + assert sent == [] + finally: + set_volume_tracking(False) + + _run(_t()) + + +def test_aspirate_sends_one_lld_category_per_command(): + """Modes OFF and CAPACITIVE at one X: one command each, the pressure block the firmware's own.""" + + async def _t(): + deck = PrepDeck() + rack = deck[1] = hamilton_96_tiprack_300uL_NTR(name="tips", with_tips=True) + plate = deck[0] = cor_96_wellplate_360uL_Fb(name="plate") + p = PrepSimulationDriver(deck=deck) + await p.setup() + assert p.pipettes is not None + await p.pipettes.pick_up_tips(rack["A1:B1"], use_channels=[0, 1]) + sent = _record(p, only=_ASPIRATE_COMMANDS) + await p.pipettes.aspirate( + plate["A1:B1"], + use_channels=[0, 1], + piston_volumes=[10.0, 20.0], + liquid_heights=[3.0, 3.0], + lld_mode=[Pipettes.LLDMode.OFF, Pipettes.LLDMode.CAPACITIVE], + ) + assert [type(c) for c in sent] == [ + PrepCmd.PrepAspirateNoLldMonitoringV2, + PrepCmd.PrepAspirateWithLldV2, + ] + assert [len(c.aspirate_parameters) for c in sent] == [1, 1] + assert sent[1].aspirate_parameters[0].p_lld == PrepCmd.PLldParameters.default() + await p.stop() + + _run(_t())