Skip to content
16 changes: 15 additions & 1 deletion pylabrobot/liquid_handling/liquid_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -1677,6 +1677,13 @@ async def aspirate96(
containers = resource.get_all_items() if resource.num_items > 1 else [resource.get_item(0)]
elif isinstance(resource, Container):
containers = [resource]
elif isinstance(resource, list) and all(isinstance(w, Well) for w in resource):
containers = resource
else:
raise TypeError(
f"Resource must be a Plate, Container, or list of Wells, got {type(resource)} "
f" for {resource}"
)

if len(containers) == 1: # single container
container = containers[0]
Expand Down Expand Up @@ -1814,10 +1821,17 @@ async def dispense96(
containers: Sequence[Container]
if isinstance(resource, Plate):
if resource.has_lid():
raise ValueError("Dispensing to plate with lid")
raise ValueError("Dispensing to plate with lid is not possible. Remove the lid first.")
containers = resource.get_all_items() if resource.num_items > 1 else [resource.get_item(0)]
elif isinstance(resource, Container):
containers = [resource]
elif isinstance(resource, list) and all(isinstance(w, Well) for w in resource):
containers = resource
else:
raise TypeError(
f"Resource must be a Plate, Container, or list of Wells, got {type(resource)} "
f"for {resource}"
)

# if we have enough liquid in the tip, remove it from the tip tracker for accounting.
# if we do not (for example because the plunger was up on tip pickup), and we
Expand Down
49 changes: 49 additions & 0 deletions pylabrobot/liquid_handling/liquid_handler_tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@
hamilton_96_tiprack_300uL_filter,
hamilton_96_tiprack_1000uL_filter,
)
from pylabrobot.resources.revvity.plates import Revvity_384_wellplate_28ul_Ub
from pylabrobot.resources.utils import create_ordered_items_2d
from pylabrobot.resources.volume_tracker import (
set_volume_tracking,
Expand Down Expand Up @@ -616,6 +617,37 @@ async def test_aspirate_dispense96(self):
)
)

async def test_dispense96_with_quadrant_well_list(self):
plate_384 = Revvity_384_wellplate_28ul_Ub(name="plate_384")
self.deck.assign_child_resource(plate_384, location=Coordinate(400, 100, 0))
quadrant_wells = plate_384.get_quadrant("tl")

await self.lh.pick_up_tips96(self.tip_rack)
await self.lh.aspirate96(self.plate, volume=10)

self.lh.backend.dispense96 = unittest.mock.AsyncMock() # type: ignore
await self.lh.dispense96(quadrant_wells, 10)
self.lh.backend.dispense96.assert_called_with( # type: ignore
dispense=MultiHeadDispensePlate(
wells=quadrant_wells,
offset=Coordinate.zero(),
tips=[self.lh.head96[i].get_tip() for i in range(96)],
volume=10,
flow_rate=None,
liquid_height=None,
blow_out_air_volume=None,
mix=None,
)
)

async def test_dispense96_well_list_mixed_parents(self):
plate2 = Cor_96_wellplate_360ul_Fb(name="plate2")
self.deck.assign_child_resource(plate2, location=Coordinate(400, 100, 0))
mixed = self.plate.get_all_items()[:48] + plate2.get_all_items()[:48]
await self.lh.pick_up_tips96(self.tip_rack)
with self.assertRaises(ValueError):
await self.lh.dispense96(mixed, 10)

async def test_transfer(self):
t = self.tip_rack.get_item("A1").get_tip()
self.lh.update_head_state({0: t})
Expand Down Expand Up @@ -1048,3 +1080,20 @@ async def test_96_head_volume_tracking_single_container(self):
assert well.tracker.get_used_volume() == 10 * 96

await self.lh.return_tips96()

async def test_96_head_volume_tracking_well_list(self):
plate_384 = Revvity_384_wellplate_28ul_Ub(name="plate_384")
self.deck.assign_child_resource(plate_384, location=Coordinate(600, 100, 0))
quadrant_wells = plate_384.get_quadrant("tl")
for well in quadrant_wells:
well.tracker.set_volume(10)

await self.lh.pick_up_tips96(self.tip_rack)
await self.lh.aspirate96(quadrant_wells, volume=10)
assert all(self.lh.head96[i].get_tip().tracker.get_used_volume() == 10 for i in range(96))
assert all(w.tracker.get_used_volume() == 0 for w in quadrant_wells)

await self.lh.dispense96(quadrant_wells, volume=10)
assert all(self.lh.head96[i].get_tip().tracker.get_used_volume() == 0 for i in range(96))
assert all(w.tracker.get_used_volume() == 10 for w in quadrant_wells)
await self.lh.return_tips96()