diff --git a/pylabrobot/hamilton/prep/driver/features/head8.py b/pylabrobot/hamilton/prep/driver/features/head8.py index 20e17d1d6fc..56f22167ee4 100644 --- a/pylabrobot/hamilton/prep/driver/features/head8.py +++ b/pylabrobot/hamilton/prep/driver/features/head8.py @@ -64,8 +64,8 @@ PipetteChannel, Pipettes, _absolute_z_from_well, - _build_container_segments, _effective_radius, + _get_container_segments, ) from .pipettes import ( default_lld_params as _default_lld_params_fn, @@ -1115,9 +1115,6 @@ def __init__(self, tip: Tip, volume: float): loc = container.get_location_wrt(self._require_deck(), "c", "c", "cavity_bottom") ref_x, ref_y = loc.x, loc.y + 3.5 * PROBE_PITCH_MM wg = _absolute_z_from_well(container, self._require_deck(), liquid_height) - ref_segments = container_segments or ( - _build_container_segments(container) if auto_container_geometry else [] - ) ref_resource = container else: wells_list = list(wells) # type: ignore[arg-type] @@ -1131,14 +1128,20 @@ def __init__(self, tip: Tip, volume: float): ref_loc = wells_list[0].get_location_wrt(self._require_deck(), "c", "c", "cavity_bottom") ref_x, ref_y = ref_loc.x, ref_loc.y wg = _absolute_z_from_well(wells_list[0], self._require_deck(), liquid_height) - ref_segments = container_segments or ( - _build_container_segments(wells_list[0]) if auto_container_geometry else [] - ) ref_resource = wells_list[0] resolved_z_fluid = z_fluid if z_fluid is not None else wg.liquid_surface resolved_z_air = z_air if z_air is not None else wg.z_air resolved_z_minimum = z_minimum if z_minimum is not None else wg.well_bottom + # the firmware counts segment 0 from z_minimum + cavity_bottom_z = ref_resource.get_location_wrt( + self._require_deck(), "c", "c", "cavity_bottom" + ).z + ref_segments = container_segments or ( + _get_container_segments(ref_resource, profile_start=resolved_z_minimum - cavity_bottom_z) + if auto_container_geometry + else [] + ) resolved_z_bottom_search_offset = ( z_bottom_search_offset if z_bottom_search_offset is not None else 2.0 ) @@ -1337,9 +1340,6 @@ def __init__(self, tip: Tip, volume: float): loc = container.get_location_wrt(self._require_deck(), "c", "c", "cavity_bottom") ref_x, ref_y = loc.x, loc.y + 3.5 * PROBE_PITCH_MM wg = _absolute_z_from_well(container, self._require_deck(), liquid_height) - ref_segments = container_segments or ( - _build_container_segments(container) if auto_container_geometry else [] - ) ref_resource = container else: wells_list = list(wells) # type: ignore[arg-type] @@ -1353,14 +1353,20 @@ def __init__(self, tip: Tip, volume: float): ref_loc = wells_list[0].get_location_wrt(self._require_deck(), "c", "c", "cavity_bottom") ref_x, ref_y = ref_loc.x, ref_loc.y wg = _absolute_z_from_well(wells_list[0], self._require_deck(), liquid_height) - ref_segments = container_segments or ( - _build_container_segments(wells_list[0]) if auto_container_geometry else [] - ) ref_resource = wells_list[0] resolved_z_fluid = z_fluid if z_fluid is not None else wg.liquid_surface resolved_z_air = z_air if z_air is not None else wg.z_air resolved_z_minimum = z_minimum if z_minimum is not None else wg.well_bottom + # the firmware counts segment 0 from z_minimum + cavity_bottom_z = ref_resource.get_location_wrt( + self._require_deck(), "c", "c", "cavity_bottom" + ).z + ref_segments = container_segments or ( + _get_container_segments(ref_resource, profile_start=resolved_z_minimum - cavity_bottom_z) + if auto_container_geometry + else [] + ) resolved_z_bottom_search_offset = ( z_bottom_search_offset if z_bottom_search_offset is not None else 2.0 ) diff --git a/pylabrobot/hamilton/prep/driver/features/head8_tests.py b/pylabrobot/hamilton/prep/driver/features/head8_tests.py index 05789a2145a..0bc76df5581 100644 --- a/pylabrobot/hamilton/prep/driver/features/head8_tests.py +++ b/pylabrobot/hamilton/prep/driver/features/head8_tests.py @@ -20,6 +20,7 @@ from pylabrobot.hamilton.prep.driver.features.pipettes import ( Pipettes, _build_pipettor_gantry_move_parameters, + _get_container_segments, ) from pylabrobot.hamilton.prep.driver.simulator import RECORDING_PREP_HEAD8 from pylabrobot.resources import Coordinate, Resource @@ -301,6 +302,36 @@ async def _run() -> None: asyncio.run(_run()) +def test_head8_aspirate_container_segments_start_at_z_minimum(): + """With auto_container_geometry, segment 0 begins at the z_minimum the command sends.""" + + 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 + + captured, _ = _record_send(p) + + await p.head8.pick_up_tips(tip_rack.column(0)) + wells = src_plate.column(0) + cavity_bottom_z = wells[0].get_location_wrt(deck, "c", "c", "cavity_bottom").z + profile_top = sum(s.height for s in _get_container_segments(wells[0])) + await p.head8.aspirate( + wells=wells, volume=10, z_minimum=cavity_bottom_z + 1.5, auto_container_geometry=True + ) + + asp = [c for c in captured if isinstance(c, PrepCmd.MphAspirateNoLldMonitoring2)] + params = asp[0].aspirate_parameters[0] + assert params.common.z_minimum == pytest.approx(cavity_bottom_z + 1.5) + sent_height = sum(s.height for s in params.container_description) + assert sent_height == pytest.approx(profile_top - 1.5) + + await p.stop() + + asyncio.run(_run()) + + def test_head8_v2_dispense_sends_mphdispensetnolld2(): """Simulator default (use_v1=False) → V2 dispense command class is sent.""" diff --git a/pylabrobot/hamilton/prep/driver/features/pipettes.py b/pylabrobot/hamilton/prep/driver/features/pipettes.py index 9274a2e4fdc..a23882936c4 100644 --- a/pylabrobot/hamilton/prep/driver/features/pipettes.py +++ b/pylabrobot/hamilton/prep/driver/features/pipettes.py @@ -73,7 +73,7 @@ ) from pylabrobot.resources.tip_rack import TipSpot, tip_origin from pylabrobot.resources.trash import Trash -from pylabrobot.resources.well import CrossSectionType, Well +from pylabrobot.resources.well import CrossSectionType, Well, WellBottomType from .. import prep_commands as PrepCmd from ..prep_commands import PIPETTOR_OBJECT_PATH @@ -314,56 +314,231 @@ def _effective_radius(resource) -> float: return float(resource.get_size_x() / 2) -def _build_container_segments(resource: object) -> list[PrepCmd.SegmentDescriptor]: - """Derive PrepCmd.SegmentDescriptor list from a Well's geometry for liquid-following. +# Containers already warned about following a V- or U-bottom as a cylinder, by name. +_warned_cylinder_fallback: set = set() - Each segment is a frustum. The firmware uses area_bottom/area_top to - interpolate cross-sectional area A(z) within the segment and computes the - Z-axis following speed as dz/dt = Q / A(z), where Q is volumetric flow rate. +# The most segments one channel has been sent and the device took; 150 were refused (0x0011). +MAX_CONTAINER_SEGMENTS = 9 - Returns [] when geometry cannot be determined; the firmware then falls back to - the tube_radius / cone model in PrepCmd.CommonParameters. + +def _merge_segments( + segments: Sequence[PrepCmd.SegmentDescriptor], limit: int = MAX_CONTAINER_SEGMENTS +) -> list[PrepCmd.SegmentDescriptor]: + """The segments with neighbours of one area joined, then the closest pairs until `limit` are left. + + A joined pair keeps its volume: its area is their volume over their height. + + Args: + segments: bottom up, each of constant area. + limit: how many segments may be left. + + Returns: + At most `limit` segments over the same height, holding the same volume. """ - if not isinstance(resource, Well): - return [] - well: Well = resource + merged = list(segments) + + def join(i: int) -> None: + a, b = merged[i], merged[i + 1] + height = a.height + b.height + area = (a.area_bottom * a.height + b.area_bottom * b.height) / height + merged[i : i + 2] = [PrepCmd.SegmentDescriptor(area_top=area, area_bottom=area, height=height)] + + i = 0 + while i < len(merged) - 1: + if math.isclose(merged[i].area_bottom, merged[i + 1].area_bottom, rel_tol=1e-9): + join(i) + else: + i += 1 + while len(merged) > limit: + differences = [ + abs(merged[i].area_bottom - merged[i + 1].area_bottom) for i in range(len(merged) - 1) + ] + join(differences.index(min(differences))) + return merged - size_z = well.get_size_z() - if well.cross_section_type == CrossSectionType.CIRCLE: - area = math.pi * (well.get_size_x() / 2) ** 2 - elif well.cross_section_type == CrossSectionType.RECTANGLE: - area = well.get_size_x() * well.get_size_y() - else: - return [] +def _get_profile_segments(resource: object) -> list[PrepCmd.SegmentDescriptor]: + """A container's cross-section as firmware segments, bottom up from the cavity bottom. - if well.supports_compute_height_volume_functions(): - # Non-linear geometry: approximate with N frustum segments by sampling dV/dh. - n_boundaries = 11 # 10 segments - heights = [size_z * i / (n_boundaries - 1) for i in range(n_boundaries)] - eps = size_z / (n_boundaries - 1) * 0.1 + One step per `height_volume_data` knot interval; else one per 0.5 mm of cavity depth from its + height-volume functions; else its footprint as one cylinder. Steps of one area are joined, and + the closest pairs, until `MAX_CONTAINER_SEGMENTS` are left. - def area_at(h: float) -> float: - h_lo = max(0.0, h - eps) - h_hi = min(size_z, h + eps) - dv = well.compute_volume_from_height(h_hi) - well.compute_volume_from_height(h_lo) - return float(dv / (h_hi - h_lo)) + Args: + resource: the container. - return [ - PrepCmd.SegmentDescriptor( - area_top=float(area_at(heights[i + 1])), - area_bottom=float(area_at(heights[i])), - height=float(heights[i + 1] - heights[i]), + Returns: + The segments, each of constant area (dV/dh, in mm2); [] for a shape it cannot tell. + + Raises: + ValueError: If the volume does not rise with height. + """ + if not isinstance(resource, Container): + return [] + try: + depth = resource.get_size_z() - resource.material_z_thickness + except NotImplementedError: + depth = resource.get_size_z() + if resource.height_volume_data: + knots = sorted(resource.height_volume_data.items()) + elif resource.supports_compute_height_volume_functions(): + steps = max(1, math.ceil(depth / 0.5)) + heights = [depth * i / steps for i in range(steps + 1)] + knots = [(h, resource.compute_volume_from_height(h)) for h in heights] + else: + cross_section = resource.cross_section_type if isinstance(resource, Well) else None + if cross_section == CrossSectionType.CIRCLE: + area = math.pi * (resource.get_size_x() / 2) ** 2 + elif cross_section == CrossSectionType.RECTANGLE: + area = resource.get_size_x() * resource.get_size_y() + else: + return [] + bottom = resource.bottom_type if isinstance(resource, Well) else None + if ( + bottom in (WellBottomType.V, WellBottomType.U) + and resource.name not in _warned_cylinder_fallback + ): + _warned_cylinder_fallback.add(resource.name) + logger.warning( + "%s has a %s-bottom but no height-volume data, so the tip follows it as a cylinder", + resource.name, + bottom.value, ) - for i in range(n_boundaries - 1) - ] + return [PrepCmd.SegmentDescriptor(area_top=area, area_bottom=area, height=depth)] + segments = [] + for (h0, v0), (h1, v1) in zip(knots, knots[1:]): + if h1 <= h0: + continue + if v1 <= v0: + raise ValueError(f"{resource.name}: the volume does not rise between {h0} and {h1} mm") + area = (v1 - v0) / (h1 - h0) + segments.append(PrepCmd.SegmentDescriptor(area_top=area, area_bottom=area, height=h1 - h0)) + return _merge_segments(segments) + + +def _get_profile_drop( + segments: Sequence[PrepCmd.SegmentDescriptor], liquid_height: float, volume: float +) -> float: + """How far a surface sinks through the segments when `volume` leaves, in mm. + + Args: + segments: bottom up from the cavity bottom; the top one extends upwards. + liquid_height: above the cavity bottom, in mm. + volume: drawn, in uL. - # Simple geometry: single segment with constant cross-section. + Returns: + The drop, down to the cavity bottom at most. + """ + bases = [0.0] + for segment in segments[:-1]: + bases.append(bases[-1] + segment.height) + height, left, drop = liquid_height, volume, 0.0 + for base, segment in reversed(list(zip(bases, segments))): + if height <= base: + continue + room = (height - base) * segment.area_bottom + if left <= room: + return drop + left / segment.area_bottom + left -= room + drop += height - base + height = base + return drop + + +def _scale_areas( + segments: Sequence[PrepCmd.SegmentDescriptor], scale: float +) -> list[PrepCmd.SegmentDescriptor]: + """The segments with every area multiplied by `scale`.""" return [ - PrepCmd.SegmentDescriptor(area_top=float(area), area_bottom=float(area), height=float(size_z)) + PrepCmd.SegmentDescriptor( + area_top=s.area_top * scale, area_bottom=s.area_bottom * scale, height=s.height + ) + for s in segments ] +def _cut_profile( + segments: Sequence[PrepCmd.SegmentDescriptor], start: float +) -> list[PrepCmd.SegmentDescriptor]: + """The profile from `start` up, as the firmware counts it: its segment 0 begins at `z_minimum`. + + Args: + segments: bottom up from the cavity bottom. + start: `z_minimum` above the cavity bottom, in mm; below it, the first segment is extended. + + Returns: + The segments from `start` up. + + Raises: + ValueError: If `start` lies at or above the top of the profile. + """ + if start <= 0: + first = segments[0] + extended = PrepCmd.SegmentDescriptor( + area_top=first.area_top, area_bottom=first.area_bottom, height=first.height - start + ) + return [extended, *segments[1:]] + base = 0.0 + for index, segment in enumerate(segments): + top = base + segment.height + if top > start: + shortened = PrepCmd.SegmentDescriptor( + area_top=segment.area_top, area_bottom=segment.area_bottom, height=top - start + ) + return [shortened, *segments[index + 1 :]] + base = top + raise ValueError(f"z_minimum {start} mm above the cavity bottom lies above the profile") + + +def _get_container_segments( + resource: object, + liquid_height: Optional[float] = None, + piston_volume: Optional[float] = None, + surface_following_distance: Optional[float] = None, + profile_start: float = 0.0, +) -> list[PrepCmd.SegmentDescriptor]: + """The segments the firmware follows the surface by, for one channel's container. + + Args: + resource: the container. + liquid_height: above the cavity bottom, in mm, where the draw starts. Needed with a distance. + piston_volume: what the piston moves, in uL. Needed with a distance. + surface_following_distance: how far the tip is to sink, in mm. None follows the profile as it + is; 0 returns no segments, for the caller to send with `tube_radius` 0. + profile_start: `z_minimum` above the cavity bottom, in mm, where the firmware's segment 0 begins. + + Returns: + The profile from `profile_start` up; with a distance, its areas scaled so the tip sinks that. + + Raises: + ValueError: If a distance is negative or given without the height and volume, or the profile + draws nothing. + """ + segments = _get_profile_segments(resource) + if segments: + segments = _cut_profile(segments, profile_start) + if surface_following_distance is None or not segments: + return segments + if surface_following_distance < 0: + raise ValueError( + f"surface_following_distance must be at least 0, is {surface_following_distance}" + ) + if surface_following_distance == 0: + return [] + if liquid_height is None or piston_volume is None: + raise ValueError("a surface following distance needs the liquid height and piston volume") + above_start = liquid_height - profile_start + if _get_profile_drop(segments, above_start, piston_volume) <= 0: + raise ValueError(f"nothing is drawn from {liquid_height} mm, so there is no drop to scale") + # Scaled areas change which heights the draw spans, so the scale is solved, not divided out. + low, high = 1e-4, 1e4 + for _ in range(80): + scale = math.sqrt(low * high) + drop = _get_profile_drop(_scale_areas(segments, scale), above_start, piston_volume) + low, high = (scale, high) if drop > surface_following_distance else (low, scale) + return _scale_areas(segments, math.sqrt(low * high)) + + class _WellGeometry(NamedTuple): """Absolute Z positions derived from well geometry.""" @@ -4639,7 +4814,13 @@ def _resolve_channel_context( if container_segments is not None and i < len(container_segments): ch_segments[ch] = container_segments[i] elif auto_container_geometry: - ch_segments[ch] = _build_container_segments(indexed_ops[ch].resource) + ch_segments[ch] = _get_container_segments( + indexed_ops[ch].resource, + profile_start=z_minimum[i] + - indexed_ops[ch] + .resource.get_location_wrt(self._require_deck(), "c", "c", "cavity_bottom") + .z, + ) else: ch_segments[ch] = [] diff --git a/pylabrobot/hamilton/prep/driver/features/pipettes_tests.py b/pylabrobot/hamilton/prep/driver/features/pipettes_tests.py index c062a88a0dc..eee35d0e6f4 100644 --- a/pylabrobot/hamilton/prep/driver/features/pipettes_tests.py +++ b/pylabrobot/hamilton/prep/driver/features/pipettes_tests.py @@ -5,13 +5,19 @@ import asyncio import functools from typing import Any, List -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, patch import pytest from pylabrobot.hamilton.prep import PrepSimulationDriver from pylabrobot.hamilton.prep.driver import prep_commands as PrepCmd -from pylabrobot.hamilton.prep.driver.features.pipettes import Pipettes +from pylabrobot.hamilton.prep.driver.features.pipettes import ( + MAX_CONTAINER_SEGMENTS, + Pipettes, + _get_container_segments, + _get_profile_drop, +) +from pylabrobot.hamilton.prep.driver.features.pipettes import logger as pipettes_logger from pylabrobot.hamilton.prep.driver.simulator import ( SIMULATED_X_AXIS_OFFSET, SIMULATED_X_SPEED, @@ -23,7 +29,7 @@ from pylabrobot.hamilton.transport.tcp.packets import Address from pylabrobot.hamilton.transport.tcp.wire_types import HcResultEntry from pylabrobot.lib.liquid_handling.pipette_batch_scheduling import ChannelBatch -from pylabrobot.resources import Coordinate, PetriDish, Resource +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 @@ -2368,6 +2374,109 @@ async def _t(): _run(_t()) +def test_container_segments_are_one_step_per_height_volume_knot(): + """Each knot interval of the data is one segment of its own dV/dh.""" + well = cor_96_wellplate_360uL_Fb(name="plate")["A1"][0] + assert well.height_volume_data is not None + knots = sorted(well.height_volume_data.items()) + segments = _get_container_segments(well) + assert len(segments) == len(knots) - 1 + for segment, (h0, v0), (h1, v1) in zip(segments, knots, knots[1:]): + assert segment.height == pytest.approx(h1 - h0) + assert segment.area_bottom == segment.area_top == pytest.approx((v1 - v0) / (h1 - h0)) + + +def test_container_segments_from_functions_join_steps_of_one_area(): + """Without data, steps of one area are one segment: a cylinder by functions is one cylinder.""" + container = Container( + name="c", + size_x=10, + size_y=10, + size_z=12, + material_z_thickness=2, + compute_volume_from_height=lambda h: 50.0 * h, + compute_height_from_volume=lambda v: v / 50.0, + ) + segments = _get_container_segments(container) + assert len(segments) == 1 + assert segments[0].height == pytest.approx(10.0) + assert segments[0].area_bottom == pytest.approx(50.0) + + +def test_container_segments_are_at_most_what_the_device_takes_and_keep_the_volume(): + """A cone's 0.5 mm steps become MAX_CONTAINER_SEGMENTS, holding what the functions say.""" + container = Container( + name="cone", + size_x=10, + size_y=10, + size_z=42, + material_z_thickness=2, + compute_volume_from_height=lambda h: 0.5 * h**2, + compute_height_from_volume=lambda v: (2 * v) ** 0.5, + ) + segments = _get_container_segments(container) + assert len(segments) == MAX_CONTAINER_SEGMENTS + assert sum(s.height for s in segments) == pytest.approx(40.0) + assert sum(s.area_bottom * s.height for s in segments) == pytest.approx(0.5 * 40.0**2) + + +def test_container_segments_fall_back_to_the_footprint_and_warn_for_a_v_bottom(): + """No height-volume model: one cylinder over the cavity depth, and a V-bottom says so once.""" + well = Well(name="v_well", size_x=6, size_y=6, size_z=10, material_z_thickness=1, bottom_type="V") + with patch.object(pipettes_logger, "warning") as warning: + segments = _get_container_segments(well) + _get_container_segments(well) + assert len(segments) == 1 + assert segments[0].area_bottom == pytest.approx(3.14159 * 9, rel=1e-4) + assert segments[0].height == pytest.approx(9.0) + assert warning.call_count == 1 + + +def test_profile_drop_matches_the_containers_own_height_from_volume(): + """The firmware's drop through the steps is the tracker's: height now less height after.""" + well = cor_96_wellplate_360uL_Fb(name="plate")["A1"][0] + start = well.compute_height_from_volume(200.0) + expected = start - well.compute_height_from_volume(175.0) + assert _get_profile_drop(_get_container_segments(well), start, 25.0) == pytest.approx(expected) + + +def test_a_surface_following_distance_scales_the_profile_to_that_drop(): + """The areas are scaled so the tip sinks exactly the distance asked; 0 is not yet known.""" + well = cor_96_wellplate_360uL_Fb(name="plate")["A1"][0] + start = well.compute_height_from_volume(200.0) + segments = _get_container_segments(well, start, 25.0, surface_following_distance=0.3) + assert _get_profile_drop(segments, start, 25.0) == pytest.approx(0.3, abs=1e-4) + assert _get_container_segments(well, start, 25.0, surface_following_distance=0.0) == [] + with pytest.raises(ValueError, match="liquid height and piston volume"): + _get_container_segments(well, surface_following_distance=0.3) + + +def test_container_segments_start_at_z_minimum_as_the_firmware_counts_them(): + """Segment 0 begins at z_minimum: cut inside an interval, on a knot, and extended below it.""" + well = cor_96_wellplate_360uL_Fb(name="plate")["A1"][0] + whole = _get_container_segments(well) + inside = _get_container_segments(well, profile_start=1.0) + assert inside[0].height == pytest.approx(1.69 - 1.0) + assert inside[0].area_bottom == pytest.approx(whole[1].area_bottom) + assert len(inside) == len(whole) - 1 + on_knot = _get_container_segments(well, profile_start=1.69) + assert on_knot[0].height == pytest.approx(whole[2].height) + below = _get_container_segments(well, profile_start=-0.5) + assert below[0].height == pytest.approx(whole[0].height + 0.5) + with pytest.raises(ValueError, match="above the profile"): + _get_container_segments(well, profile_start=20.0) + + +def test_a_surface_following_distance_counts_from_z_minimum(): + """Scaled from where the firmware starts the profile, the tip still sinks the distance asked.""" + well = cor_96_wellplate_360uL_Fb(name="plate")["A1"][0] + start = well.compute_height_from_volume(200.0) + segments = _get_container_segments( + well, start, 25.0, surface_following_distance=0.3, profile_start=2.0 + ) + assert _get_profile_drop(segments, start - 2.0, 25.0) == pytest.approx(0.3, abs=1e-4) + + def _probe_liquid_setup(): """A simulated Prep with a plate and a dish without height-volume functions.