open_pub/Uni-Lab-OS
1
0
Fork 0
mirror of https://github.com/dptech-corp/Uni-Lab-OS.git synced 2026-02-06 23:15:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

Update prcxi.py

Browse Source
This commit is contained in:
Guangxin Zhang
2025-07-16 18:22:02 +08:00
parent d81297d699
commit 71f6deda6b
2 changed files with 120 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

132
unilabos/devices/liquid_handling/liquid_handler_abstract.py
View File

@@ -48,43 +48,99 @@ class LiquidHandlerAbstract(LiquidHandler):
none_keys: List[str] = [], none_keys: List[str] = [],
): ):
"""A complete *remove* (aspirate → waste) operation.""" """A complete *remove* (aspirate → waste) operation."""
trash = self.deck.get_trash_area()
try: try:
if is_96_well: if is_96_well:
pass # This mode is not verified pass # This mode is not verified.
else: else:
if len(vols) != len(sources): # 首先应该对任务分组，然后每次1个/8个进行操作处理
raise ValueError("Length of `vols` must match `sources`.") if len(use_channels) == 1:
tip = []
for src, vol in zip(sources, vols): for _ in range(len(use_channels)):
await self.move_to(src, dis_to_top=top[0] if top else 0) tip.extend(next(self.current_tip))
tip = next(self.current_tip)
await self.pick_up_tips(tip) await self.pick_up_tips(tip)
await self.aspirate(
resources=[src], for _ in range(len(waste_liquid)):
vols=[vol], await self.aspirate(
use_channels=use_channels, # only aspirate96 used, default to None resources=sources,
flow_rates=[flow_rates[0]] if flow_rates else None, vols=[vols[_]],
offsets=[offsets[0]] if offsets else None, use_channels=use_channels,
liquid_height=[liquid_height[0]] if liquid_height else None, flow_rates=[flow_rates[0]] if flow_rates else None,
blow_out_air_volume=blow_out_air_volume[0] if blow_out_air_volume else None, offsets=[offsets[0]] if offsets else None,
spread=spread, liquid_height=[liquid_height[0]] if liquid_height else None,
) blow_out_air_volume=[blow_out_air_volume[0]] if blow_out_air_volume else None,
await self.custom_delay(seconds=delays[0] if delays else 0) spread=spread,
await self.dispense( )
resources=waste_liquid, if delays is not None:
vols=[vol], await self.custom_delay(seconds=delays[0])
use_channels=use_channels, await self.dispense(
flow_rates=[flow_rates[1]] if flow_rates else None, resources=[waste_liquid[_]],
offsets=[offsets[1]] if offsets else None, vols=[vols[_]],
liquid_height=[liquid_height[1]] if liquid_height else None, use_channels=use_channels,
blow_out_air_volume=blow_out_air_volume[1] if blow_out_air_volume else None, flow_rates=[flow_rates[1]] if flow_rates else None,
spread=spread, offsets=[offsets[1]] if offsets else None,
) blow_out_air_volume=[blow_out_air_volume[1]] if blow_out_air_volume else None,
await self.discard_tips() # For now, each of tips is discarded after use liquid_height=[liquid_height[1]] if liquid_height else None,
spread=spread,
)
await self.discard_tips()
elif len(use_channels) == 8:
tip = []
for _ in range(len(use_channels)):
tip.extend(next(self.current_tip))
await self.pick_up_tips(tip)
# 对于8个的情况，需要判断此时任务是不是能被8通道移液站来成功处理
if len(sources) % 8 != 0:
raise ValueError(f"Length of `sources` {len(sources)} must be a multiple of 8 for 8-channel mode.")
# 8个8个来取任务序列
for i in range(0, len(sources), 8):
# 取出8个任务
current_targets = waste_liquid[i:i + 8]
current_reagent_sources = sources[i:i + 8]
current_asp_vols = vols[i:i + 8]
current_dis_vols = vols[i:i + 8]
current_asp_flow_rates = flow_rates[i:i + 8] if flow_rates else [None] * 8
current_dis_flow_rates = flow_rates[-i*8-8:len(flow_rates)-i*8] if flow_rates else [None] * 8
current_asp_offset = offsets[i:i + 8] if offsets else [None] * 8
current_dis_offset = offsets[-i*8-8:len(offsets)-i*8] if offsets else [None] * 8
current_asp_liquid_height = liquid_height[i:i + 8] if liquid_height else [None] * 8
current_dis_liquid_height = liquid_height[-i*8-8:len(liquid_height)-i*8] if liquid_height else [None] * 8
current_asp_blow_out_air_volume = blow_out_air_volume[i:i + 8] if blow_out_air_volume else [None] * 8
current_dis_blow_out_air_volume = blow_out_air_volume[-i*8-8:len(blow_out_air_volume)-i*8] if blow_out_air_volume else [None] * 8
await self.aspirate(
resources=current_reagent_sources,
vols=current_asp_vols,
use_channels=use_channels,
flow_rates=current_asp_flow_rates,
offsets=current_asp_offset,
liquid_height=current_asp_liquid_height,
blow_out_air_volume=current_asp_blow_out_air_volume,
spread=spread,
)
if delays is not None:
await self.custom_delay(seconds=delays[0])
await self.dispense(
resources=current_targets,
vols=current_dis_vols,
use_channels=use_channels,
flow_rates=current_dis_flow_rates,
offsets=current_dis_offset,
liquid_height=current_dis_liquid_height,
blow_out_air_volume=current_dis_blow_out_air_volume,
spread=spread,
)
if delays is not None:
await self.custom_delay(seconds=delays[1])
#await self.touch_tip(current_targets)
except Exception as e: except Exception as e:
raise RuntimeError(f"Liquid removal failed: {e}") from e traceback.print_exc()
raise RuntimeError(f"Liquid addition failed: {e}") from e
# --------------------------------------------------------------- # ---------------------------------------------------------------
# ADD LIQUID ----------------------------------------------------- # ADD LIQUID -----------------------------------------------------
@@ -183,13 +239,13 @@ class LiquidHandlerAbstract(LiquidHandler):
current_asp_vols = asp_vols[i:i + 8] current_asp_vols = asp_vols[i:i + 8]
current_dis_vols = dis_vols[i:i + 8] current_dis_vols = dis_vols[i:i + 8]
current_asp_flow_rates = flow_rates[i:i + 8] if flow_rates else [None] * 8 current_asp_flow_rates = flow_rates[i:i + 8] if flow_rates else [None] * 8
current_dis_flow_rates = flow_rates[i+8 :i + 16] if flow_rates else [None] * 8 current_dis_flow_rates = flow_rates[-i*8-8:len(flow_rates)-i*8] if flow_rates else [None] * 8
current_asp_offset = offsets[i:i + 8] if offsets else [None] * 8 current_asp_offset = offsets[i:i + 8] if offsets else [None] * 8
current_dis_offset = offsets[i+8 :i + 16] if offsets else [None] * 8 current_dis_offset = offsets[-i*8-8:len(offsets)-i*8] if offsets else [None] * 8
current_asp_liquid_height = liquid_height[i:i + 8] if liquid_height else [None] * 8 current_asp_liquid_height = liquid_height[i:i + 8] if liquid_height else [None] * 8
current_dis_liquid_height = liquid_height[i+8 :i + 16] if liquid_height else [None] * 8 current_dis_liquid_height = liquid_height[-i*8-8:len(liquid_height)-i*8] if liquid_height else [None] * 8
current_asp_blow_out_air_volume = blow_out_air_volume[i:i + 8] if blow_out_air_volume else [None] * 8 current_asp_blow_out_air_volume = blow_out_air_volume[i:i + 8] if blow_out_air_volume else [None] * 8
current_dis_blow_out_air_volume = blow_out_air_volume[i+8 :i + 16] if blow_out_air_volume else [None] * 8 current_dis_blow_out_air_volume = blow_out_air_volume[-i*8-8:len(blow_out_air_volume)-i*8] if blow_out_air_volume else [None] * 8
await self.aspirate( await self.aspirate(
resources=current_reagent_sources, resources=current_reagent_sources,
@@ -347,11 +403,11 @@ class LiquidHandlerAbstract(LiquidHandler):
current_dis_vols = dis_vols[i:i + 8] current_dis_vols = dis_vols[i:i + 8]
current_asp_flow_rates = asp_flow_rates[i:i + 8] current_asp_flow_rates = asp_flow_rates[i:i + 8]
current_asp_offset = offsets[i:i + 8] if offsets else [None] * 8 current_asp_offset = offsets[i:i + 8] if offsets else [None] * 8
current_dis_offset = offsets[i+8 :i + 16] if offsets else [None] * 8 current_dis_offset = offsets[-i*8-8:len(offsets)-i*8] if offsets else [None] * 8
current_asp_liquid_height = liquid_height[i:i + 8] if liquid_height else [None] * 8 current_asp_liquid_height = liquid_height[i:i + 8] if liquid_height else [None] * 8
current_dis_liquid_height = liquid_height[i+8 :i + 16] if liquid_height else [None] * 8 current_dis_liquid_height = liquid_height[-i*8-8:len(liquid_height)-i*8] if liquid_height else [None] * 8
current_asp_blow_out_air_volume = blow_out_air_volume[i:i + 8] if blow_out_air_volume else [None] * 8 current_asp_blow_out_air_volume = blow_out_air_volume[i:i + 8] if blow_out_air_volume else [None] * 8
current_dis_blow_out_air_volume = blow_out_air_volume[i+8 :i + 16] if blow_out_air_volume else [None] * 8 current_dis_blow_out_air_volume = blow_out_air_volume[-i*8-8:len(blow_out_air_volume)-i*8] if blow_out_air_volume else [None] * 8
current_dis_flow_rates = dis_flow_rates[i:i + 8] if dis_flow_rates else [None] * 8 current_dis_flow_rates = dis_flow_rates[i:i + 8] if dis_flow_rates else [None] * 8
await self.aspirate( await self.aspirate(

37
unilabos/devices/liquid_handling/prcxi/prcxi.py
View File

@@ -1033,22 +1033,33 @@ if __name__ == "__main__":
# asyncio.run(handler.drop_tips(tip_rack.children[8:16],[0,1,2,3,4,5,6,7])) # asyncio.run(handler.drop_tips(tip_rack.children[8:16],[0,1,2,3,4,5,6,7]))
# asyncio.run(handler.mix(well_containers.children[:8], mix_time=3, mix_vol=50, height_to_bottom=0.5, offsets=Coordinate(0, 0, 0), mix_rate=100)) # asyncio.run(handler.mix(well_containers.children[:8], mix_time=3, mix_vol=50, height_to_bottom=0.5, offsets=Coordinate(0, 0, 0), mix_rate=100))
#print(json.dumps(handler._unilabos_backend.steps_todo_list, indent=2)) # Print matrix info #print(json.dumps(handler._unilabos_backend.steps_todo_list, indent=2)) # Print matrix info
asyncio.run(handler.add_liquid( # asyncio.run(handler.add_liquid(
asp_vols=[100]*16, # asp_vols=[100]*16,
dis_vols=[100]*16, # dis_vols=[100]*16,
reagent_sources=well_containers.children[-16:], # reagent_sources=well_containers.children[-16:],
targets=well_containers.children[:16], # targets=well_containers.children[:16],
# use_channels=[0, 1, 2, 3, 4, 5, 6, 7],
# flow_rates=[None] * 32,
# offsets=[Coordinate(0, 0, 0)] * 32,
# liquid_height=[None] * 16,
# blow_out_air_volume=[None] * 16,
# delays=None,
# mix_time=3,
# mix_vol=50,
# spread="wide",
# ))
asyncio.run(handler.remove_liquid(
vols=[100]*16,
sources=well_containers.children[-16:],
waste_liquid=well_containers.children[:16], # 这个有些奇怪，但是好像也只能这么写
use_channels=[0, 1, 2, 3, 4, 5, 6, 7], use_channels=[0, 1, 2, 3, 4, 5, 6, 7],
flow_rates=[None] * 16, flow_rates=[None] * 32,
offsets=[Coordinate(0, 0, 0)] * 16, offsets=[Coordinate(0, 0, 0)] * 32,
liquid_height=[None] * 16, liquid_height=[None] * 32,
blow_out_air_volume=[None] * 16, blow_out_air_volume=[None] * 32,
delays=None,
mix_time=3,
mix_vol=50,
spread="wide", spread="wide",
)) ))
# asyncio.run(handler.transfer_liquid( # asyncio.run(handler.transfer_liquid(
# asp_vols=[100]*16, # asp_vols=[100]*16,
# dis_vols=[100]*16, # dis_vols=[100]*16,