Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion pylabrobot/hamilton/prep/driver/features/head8.py
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,7 @@
queue_volume_transfers,
successes_from_failed_channels,
)
from pylabrobot.resources.tip_rack import TipSpot, tip_origin
from pylabrobot.resources.tip_rack import TipSpot, check_tip_racks_available, tip_origin
from pylabrobot.resources.utils import create_ordered_items_2d
from pylabrobot.resources.volume_tracker import does_volume_tracking

Expand Down Expand Up @@ -547,6 +547,7 @@ async def pick_up_tips(
self._require_all_channels(use_channels, "pick_up_tips")
if len(tip_spots) != NUM_PROBES:
raise ValueError(f"pick_up_tips requires {NUM_PROBES} tip spots, got {len(tip_spots)}")
check_tip_racks_available(tip_spots, "pick up tips from")
resolved_end = self._resolve_traverse_height(minimum_traverse_height_end)

for ch in use_channels:
Expand Down Expand Up @@ -643,6 +644,10 @@ async def drop_tips(
self._require_all_channels(use_channels, "drop_tips")
if len(destinations) != NUM_PROBES:
raise ValueError(f"drop_tips requires {NUM_PROBES} destinations, got {len(destinations)}")
check_tip_racks_available(
[destination for destination in destinations if isinstance(destination, TipSpot)],
"drop tips into",
)
tip = self._require_mounted_tip()
resolved_end = self._resolve_traverse_height(minimum_traverse_height_end)

Expand Down
29 changes: 29 additions & 0 deletions pylabrobot/hamilton/prep/driver/features/head8_tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
hamilton_96_tiprack_50uL_NTR,
hamilton_96_tiprack_1000uL,
)
from pylabrobot.resources.lid import Lid
from pylabrobot.resources.tip_tracker import does_tip_tracking, set_tip_tracking
from pylabrobot.resources.volume_tracker import does_volume_tracking, set_volume_tracking

Expand Down Expand Up @@ -166,6 +167,34 @@ async def _run() -> None:
asyncio.run(_run())


def test_head8_tip_commands_refuse_a_rack_no_head_can_reach():
"""A rack under a lid is not picked up from, nor dropped into, and nothing is sent."""

async def _run() -> None:
deck, tip_rack, _, _ = _make_deck()
p = PrepSimulationDriver(deck=deck, declared_configuration_json=RECORDING_PREP_HEAD8)
await p.setup()
assert p.head8 is not None
spots = tip_rack.column(0)

tip_rack.lid = Lid("lid", size_x=127.0, size_y=86.0, size_z=10.0, nesting_z_height=2.0)
sent, _ = _record_send(p)
with pytest.raises(ValueError, match="pick up tips from 'ntr': it is not available"):
await p.head8.pick_up_tips(spots)
assert sent == []

tip_rack.lid.unassign()
await p.head8.pick_up_tips(spots)
tip_rack.lid = Lid("lid", size_x=127.0, size_y=86.0, size_z=10.0, nesting_z_height=2.0)
sent.clear()
with pytest.raises(ValueError, match="drop tips into 'ntr': it is not available"):
await p.head8.drop_tips(spots)
assert sent == []
await p.stop()

asyncio.run(_run())


def test_head8_tips_move_between_spots_and_shafts():
"""With tip tracking on, each tip moves off its spot onto its probe's shaft, and back."""
from pylabrobot.resources.tip_tracking import set_tip_tracking
Expand Down
7 changes: 6 additions & 1 deletion pylabrobot/hamilton/prep/driver/features/pipettes.py
Original file line number Diff line number Diff line change
Expand Up @@ -86,7 +86,7 @@
queue_volume_transfers,
successes_from_failed_channels,
)
from pylabrobot.resources.tip_rack import TipSpot, tip_origin
from pylabrobot.resources.tip_rack import TipSpot, check_tip_racks_available, tip_origin
from pylabrobot.resources.trash import Trash
from pylabrobot.resources.well import CrossSectionType, Well, WellBottomType

Expand Down Expand Up @@ -4649,6 +4649,7 @@ async def pick_up_tips(
offsets_list = list(offsets) if offsets is not None else [Coordinate.zero()] * len(tip_spots)
if len(offsets_list) != len(tip_spots):
raise ValueError("len(offsets) must equal len(tip_spots)")
check_tip_racks_available(tip_spots, "pick up tips from")

# The fewest groups of spots the channels can reach without moving the gantry between them
batches = plan_batches(
Expand Down Expand Up @@ -4919,6 +4920,10 @@ async def drop_tips(
offsets_list = list(offsets) if offsets is not None else [Coordinate.zero()] * len(destinations)
if len(offsets_list) != len(destinations):
raise ValueError("len(offsets) must equal len(destinations)")
check_tip_racks_available(
[destination for destination in destinations if isinstance(destination, TipSpot)],
"drop tips into",
)

if all(isinstance(destination, Trash) for destination in destinations):
await self._drop_tips_in_one_move(
Expand Down
34 changes: 34 additions & 0 deletions pylabrobot/hamilton/prep/driver/features/pipettes_tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,7 @@
hamilton_tip_300uL,
)
from pylabrobot.resources.hamilton.core_gripper_tools import hamilton_core_gripper_tool
from pylabrobot.resources.lid import Lid
from pylabrobot.resources.liquid import Liquid
from pylabrobot.resources.tip_tracking import set_tip_tracking
from pylabrobot.resources.volume_tracker import set_volume_tracking
Expand Down Expand Up @@ -168,6 +169,39 @@ async def _t():
_run(_t())


def test_tip_commands_refuse_a_rack_no_head_can_reach():
"""A rack under a lid is not picked up from, nor dropped into, and nothing is sent."""

async def _t():
set_tip_tracking(True)
try:
deck = PrepDeck()
tip_rack = deck[3] = hamilton_96_tiprack_50uL_NTR(name="ntr", with_tips=True)
p = PrepSimulationDriver(deck=deck)
await p.setup()
assert p.pipettes is not None
spots = [tip_rack.get_item("A1"), tip_rack.get_item("B1")]

tip_rack.lid = Lid("lid", size_x=127.0, size_y=86.0, size_z=10.0, nesting_z_height=2.0)
sent = _record(p)
with pytest.raises(ValueError, match="pick up tips from 'ntr': it is not available"):
await p.pipettes.pick_up_tips(spots, use_channels=[0, 1])
assert sent == []

tip_rack.lid.unassign()
await p.pipettes.pick_up_tips(spots, use_channels=[0, 1])
tip_rack.lid = Lid("lid", size_x=127.0, size_y=86.0, size_z=10.0, nesting_z_height=2.0)
sent.clear()
with pytest.raises(ValueError, match="drop tips into 'ntr': it is not available"):
await p.pipettes.drop_tips(spots, use_channels=[0, 1])
assert sent == []
await p.stop()
finally:
set_tip_tracking(False)

_run(_t())


def test_channels_tip_and_volume_trackers_follow_pick_up_aspirate_dispense_and_drop():
"""Tips move between their spots and the channels' shafts; wells and tips track the volume moved."""

Expand Down
4 changes: 3 additions & 1 deletion pylabrobot/hamilton/star/driver/features/head96.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@
from pylabrobot.resources.plate import Plate
from pylabrobot.resources.resource import Resource
from pylabrobot.resources.tip import Tip
from pylabrobot.resources.tip_rack import TipRack, tip_origin
from pylabrobot.resources.tip_rack import TipRack, check_tip_racks_available, tip_origin
from pylabrobot.resources.volume_tracker import VolumeTracker, does_volume_tracking
from pylabrobot.resources.well import Well

Expand Down Expand Up @@ -698,6 +698,7 @@ async def pick_up_tips(
raise RuntimeError("the head already carries tips; drop them before picking up more")
if tip_rack.num_items != 96:
raise ValueError("Tip rack must have 96 tips")
check_tip_racks_available(tip_rack.get_all_items()[:1], "pick up tips from")
tips = [
spot.tip_for_pickup() if not spot.tracks_tips or spot.tip is not None else None
for spot in tip_rack.get_all_items()
Expand Down Expand Up @@ -783,6 +784,7 @@ async def drop_tips(
if isinstance(resource, TipRack):
if resource.num_items != 96:
raise ValueError("Tip rack must have 96 tips")
check_tip_racks_available(resource.get_all_items()[:1], "drop tips into")
location = resource.get_item("A1").get_location_wrt(deck, x="c", y="c", z="b")
else:
location = self._position_centred_in(resource)
Expand Down
16 changes: 16 additions & 0 deletions pylabrobot/hamilton/star/driver/features/head_tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -182,6 +182,22 @@ def recorded(written: str, read: Optional[str]) -> None:

self.driver._log_exchange = recorded # type: ignore[method-assign]

async def test_a_covered_rack_is_not_picked_up_from_nor_dropped_into(self):
from pylabrobot.resources.lid import Lid

self.tip_rack.lid = Lid("lid", size_x=127.0, size_y=86.0, size_z=10.0, nesting_z_height=2.0)
with self.assertRaisesRegex(ValueError, "pick up tips from 'tip_rack_01': it is not available"):
await self.head.pick_up_tips(self.tip_rack)
self.assertEqual(self.sent, [])

self.tip_rack.lid.unassign()
await self.head.pick_up_tips(self.tip_rack)
self.tip_rack.lid = Lid("lid", size_x=127.0, size_y=86.0, size_z=10.0, nesting_z_height=2.0)
self.sent.clear()
with self.assertRaisesRegex(ValueError, "drop tips into 'tip_rack_01': it is not available"):
await self.head.drop_tips(self.tip_rack)
self.assertEqual(self.sent, [])

async def test_pick_up_and_drop_send_what_legacy_sends(self):
await self.head.pick_up_tips(self.tip_rack)
await self.head.drop_tips(self.tip_rack)
Expand Down
9 changes: 8 additions & 1 deletion pylabrobot/hamilton/star/driver/features/pipettes.py
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,12 @@
from pylabrobot.resources.n_channel_pipettes import NChannelPipette, TipMountingShaft
from pylabrobot.resources.resource import Resource
from pylabrobot.resources.tip import Tip
from pylabrobot.resources.tip_rack import TipRack, TipSpot, tip_origin
from pylabrobot.resources.tip_rack import (
TipRack,
TipSpot,
check_tip_racks_available,
tip_origin,
)
from pylabrobot.resources.volume_tracker import VolumeTracker, does_volume_tracking
from pylabrobot.resources.well import Well

Expand Down Expand Up @@ -4270,6 +4275,7 @@ async def pick_up_tips(

if not tip_spots:
return
check_tip_racks_available(tip_spots, "pick up tips from")
tips = [spot.tip_for_pickup() for spot in tip_spots]
if not all(isinstance(tip, HamiltonTip) for tip in tips):
raise TypeError("the STAR picks up Hamilton tips")
Expand Down Expand Up @@ -4540,6 +4546,7 @@ async def drop_tips(
offsets = [Coordinate.zero()] * len(destinations) if offsets is None else list(offsets)

spots = [place for place in destinations if isinstance(place, TipSpot)]
check_tip_racks_available(spots, "drop tips into")
if len({id(spot) for spot in spots}) != len(spots):
raise ValueError("each tip must go into a spot of its own")
for spot in spots:
Expand Down
30 changes: 30 additions & 0 deletions pylabrobot/hamilton/star/driver/features/pipettes_tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -2299,6 +2299,36 @@ async def answered(module: str, command: str, **kwargs: Any):
driver._answer = answered


class TestTipCommandsRefuseARackNoHeadCanReach(unittest.IsolatedAsyncioTestCase):
"""A rack under a lid is not picked up from, nor dropped into, and nothing is sent."""

def setUp(self):
from pylabrobot.resources import set_tip_tracking

set_tip_tracking(True)
self.addCleanup(set_tip_tracking, False)

async def test_a_pickup_from_a_covered_rack_is_refused(self):
from pylabrobot.resources.lid import Lid

pipettes, rack, sent = await channels_over_a_rack()
rack.lid = Lid("lid", size_x=127.0, size_y=86.0, size_z=10.0, nesting_z_height=2.0)
with self.assertRaisesRegex(ValueError, "pick up tips from 'rack': it is not available"):
await pipettes.pick_up_tips(rack["A1:H1"])
self.assertEqual(sent, [])

async def test_a_drop_into_a_covered_rack_is_refused(self):
from pylabrobot.resources.lid import Lid

pipettes, rack, sent = await channels_over_a_rack()
await pipettes.pick_up_tips(rack["A1:H1"])
rack.lid = Lid("lid", size_x=127.0, size_y=86.0, size_z=10.0, nesting_z_height=2.0)
sent.clear()
with self.assertRaisesRegex(ValueError, "drop tips into 'rack': it is not available"):
await pipettes.drop_tips(rack["A1:H1"])
self.assertEqual(sent, [])


class TestProbeTipPresenceViaPickup(unittest.IsolatedAsyncioTestCase):
"""Spots probed by picking their tips up and putting them back; empty ones answer as a device."""

Expand Down
8 changes: 4 additions & 4 deletions pylabrobot/legacy/liquid_handling/liquid_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -330,12 +330,12 @@ def _check_no_lid(resource: Resource, action: str) -> None:


def _check_tip_racks_available(resources: Sequence[TipSpot], action: str) -> None:
"""Raise if a lid or another tip rack sits on a rack behind ``resources``, checking each rack once.
``action`` is a verb phrase for the error."""
"""Raise if a rack behind ``resources`` is not available for tip handling, checking each rack
once. ``action`` is a verb phrase for the error."""
racks = {id(r.parent): r.parent for r in resources if isinstance(r.parent, TipRack)}
for rack in racks.values():
if not rack._available_for_tip_handling:
raise ValueError(f"Cannot {action} {rack.name!r}: something is stacked on top of it.")
if not rack.available_for_tip_handling:
raise ValueError(f"Cannot {action} {rack.name!r}: it is not available for tip handling.")


class LiquidHandler(Resource, Machine):
Expand Down
8 changes: 4 additions & 4 deletions pylabrobot/legacy/liquid_handling/liquid_handler_tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -1403,22 +1403,22 @@ async def asyncSetUp(self):
await self.lh.setup()

async def test_only_the_top_rack_of_a_stack_can_be_picked_up_from(self):
with self.assertRaisesRegex(ValueError, "'bottom': something is stacked on top of it"):
with self.assertRaisesRegex(ValueError, "'bottom': it is not available for tip handling"):
await self.lh.pick_up_tips([self.bottom.get_item("A1")])
self.backend.pick_up_tips.assert_not_called()
await self.lh.pick_up_tips([self.top.get_item("A1")])
self.backend.pick_up_tips.assert_called_once()

async def test_tips_are_not_dropped_into_a_covered_rack(self):
with self.assertRaisesRegex(ValueError, "'bottom': something is stacked on top of it"):
with self.assertRaisesRegex(ValueError, "'bottom': it is not available for tip handling"):
await self.lh.drop_tips([self.bottom.get_item("A1")])
self.backend.drop_tips.assert_not_called()

async def test_a_lid_covers_a_rack_for_the_96_head(self):
self.top.lid = Lid("lid", size_x=127.76, size_y=85.48, size_z=10.0, nesting_z_height=2.0)
with self.assertRaisesRegex(ValueError, "'top': something is stacked on top of it"):
with self.assertRaisesRegex(ValueError, "'top': it is not available for tip handling"):
await self.lh.pick_up_tips96(self.top)
with self.assertRaisesRegex(ValueError, "'top': something is stacked on top of it"):
with self.assertRaisesRegex(ValueError, "'top': it is not available for tip handling"):
await self.lh.drop_tips96(self.top)
self.backend.pick_up_tips96.assert_not_called()
self.backend.drop_tips96.assert_not_called()
Expand Down
11 changes: 10 additions & 1 deletion pylabrobot/lib/liquid_handling/tip_spot_finding.py
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,7 @@ def find_tip_spots(
has_filter: Optional[bool] = None,
count: Optional[int] = None,
x_aligned: bool = False,
available_for_tip_handling: Optional[bool] = None,
) -> List[TipSpot]:
"""Find tip spots in consumption order.

Expand All @@ -126,6 +127,9 @@ def find_tip_spots(
x_aligned: Return spots from a single rack column: the first column in consumption order
holding at least count spots. A shorter column is skipped. Falls back to the unaligned
batch when no column can serve count. Without count, returns the first column.
available_for_tip_handling: True for spots in racks a head can reach, False for spots in
racks it cannot, e.g. under a lid or under another rack in a stack, None for both. See
`TipRack.available_for_tip_handling`.

Returns:
Matching tip spots.
Expand All @@ -137,7 +141,12 @@ def find_tip_spots(
if count is not None and count <= 0:
raise ValueError(f"count must be positive, got {count}")

racks = _get_tip_racks(root)
racks = [
(rack, frame)
for rack, frame in _get_tip_racks(root)
if available_for_tip_handling is None
or rack.available_for_tip_handling == available_for_tip_handling
]

rack_order = sorted(range(len(racks)), key=lambda index: _get_rack_rank(*racks[index], index))

Expand Down
31 changes: 30 additions & 1 deletion pylabrobot/lib/liquid_handling/tip_spot_finding_tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,10 +7,18 @@
import pytest

from pylabrobot.lib.liquid_handling.tip_spot_finding import find_tip_spots
from pylabrobot.resources import TIP_CAR_480_A00, Coordinate, Resource, TipRack, TipSpot
from pylabrobot.resources import (
TIP_CAR_480_A00,
Coordinate,
Resource,
ResourceStack,
TipRack,
TipSpot,
)
from pylabrobot.resources.hamilton import (
hamilton_96_tiprack_50uL,
hamilton_96_tiprack_300uL_filter,
hamilton_96_tiprack_300uL_NTR,
hamilton_96_tiprack_1000uL_filter,
)

Expand Down Expand Up @@ -193,6 +201,27 @@ def test_matches_reference_on_random_decks(seed: int) -> None:
assert find_tip_spots(root, *query) == _reference_find_tip_spots(racks, *query), query


def test_available_for_tip_handling_selects_by_whether_a_head_can_reach_the_rack() -> None:
"""Asked for, a rack with another standing on it gives no spots until that one is taken off."""
root = Resource("root", size_x=1000, size_y=1000, size_z=100)
stack = ResourceStack("stack", "z")
root.assign_child_resource(stack, location=Coordinate(0, 0, 0))
lower, upper = hamilton_96_tiprack_300uL_NTR("lower"), hamilton_96_tiprack_300uL_NTR("upper")
stack.assign_child_resource(lower)
stack.assign_child_resource(upper)

either = find_tip_spots(root, has_tip=True)
reachable = find_tip_spots(root, has_tip=True, available_for_tip_handling=True)
covered = find_tip_spots(root, has_tip=True, available_for_tip_handling=False)
assert {spot.parent for spot in either} == {lower, upper}
assert {spot.parent for spot in reachable} == {upper}
assert {spot.parent for spot in covered} == {lower}

stack.unassign_child_resource(upper)
reachable = find_tip_spots(root, has_tip=True, available_for_tip_handling=True)
assert {spot.parent for spot in reachable} == {lower}


def test_root_can_be_any_level_of_the_tree() -> None:
"""A single rack, a carrier or a facility above several benches all work as the root."""
facility = Resource("facility", size_x=5000, size_y=5000, size_z=100)
Expand Down
Loading
Loading