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

Update for discard tips

Browse Source
This commit is contained in:
Guangxin Zhang
2025-07-16 23:28:46 +08:00
parent 3584e92a1e
commit 46b7ce292b
2 changed files with 34 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -628,6 +628,10 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
for i in range(0, len(sources), 8): for i in range(0, len(sources), 8):
# 取出8个任务 # 取出8个任务
tip = []
for _ in range(len(use_channels)):
tip.extend(next(self.current_tip))
await self.pick_up_tips(tip)
current_targets = waste_liquid[i:i + 8] current_targets = waste_liquid[i:i + 8]
current_reagent_sources = sources[i:i + 8] current_reagent_sources = sources[i:i + 8]
current_asp_vols = vols[i:i + 8] current_asp_vols = vols[i:i + 8]
@@ -666,7 +670,7 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
if delays is not None: if delays is not None:
await self.custom_delay(seconds=delays[1]) await self.custom_delay(seconds=delays[1])
await self.touch_tip(current_targets) await self.touch_tip(current_targets)
await self.discard_tips() await self.discard_tips()
except Exception as e: except Exception as e:
traceback.print_exc() traceback.print_exc()
@@ -751,10 +755,6 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
await self.touch_tip(targets[_]) await self.touch_tip(targets[_])
elif len(use_channels) == 8: 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通道移液站来成功处理 # 对于8个的情况，需要判断此时任务是不是能被8通道移液站来成功处理
if len(targets) % 8 != 0: if len(targets) % 8 != 0:
@@ -763,7 +763,10 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
# 8个8个来取任务序列 # 8个8个来取任务序列
for i in range(0, len(targets), 8): for i in range(0, len(targets), 8):
# 取出8个任务 tip = []
for _ in range(len(use_channels)):
tip.extend(next(self.current_tip))
await self.pick_up_tips(tip)
current_targets = targets[i:i + 8] current_targets = targets[i:i + 8]
current_reagent_sources = reagent_sources[i:i + 8] current_reagent_sources = reagent_sources[i:i + 8]
current_asp_vols = asp_vols[i:i + 8] current_asp_vols = asp_vols[i:i + 8]
@@ -813,7 +816,8 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
if delays is not None: if delays is not None:
await self.custom_delay(seconds=delays[1]) await self.custom_delay(seconds=delays[1])
await self.touch_tip(current_targets) await self.touch_tip(current_targets)
await self.discard_tips() await self.discard_tips()
except Exception as e: except Exception as e:
traceback.print_exc() traceback.print_exc()
@@ -915,11 +919,6 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
await self.discard_tips() await self.discard_tips()
elif len(use_channels) == 8: 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通道移液站来成功处理 # 对于8个的情况，需要判断此时任务是不是能被8通道移液站来成功处理
if len(targets) % 8 != 0: if len(targets) % 8 != 0:
raise ValueError(f"Length of `targets` {len(targets)} must be a multiple of 8 for 8-channel mode.") raise ValueError(f"Length of `targets` {len(targets)} must be a multiple of 8 for 8-channel mode.")
@@ -928,6 +927,10 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
for i in range(0, len(targets), 8): for i in range(0, len(targets), 8):
# 取出8个任务 # 取出8个任务
tip = []
for _ in range(len(use_channels)):
tip.extend(next(self.current_tip))
await self.pick_up_tips(tip)
current_targets = targets[i:i + 8] current_targets = targets[i:i + 8]
current_reagent_sources = sources[i:i + 8] current_reagent_sources = sources[i:i + 8]
current_asp_vols = asp_vols[i:i + 8] current_asp_vols = asp_vols[i:i + 8]
@@ -978,7 +981,7 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
if delays is not None: if delays is not None:
await self.custom_delay(seconds=delays[1]) await self.custom_delay(seconds=delays[1])
await self.touch_tip(current_targets) await self.touch_tip(current_targets)
await self.discard_tips() await self.discard_tips()
except Exception as e: except Exception as e:
traceback.print_exc() traceback.print_exc()

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

@@ -1051,7 +1051,7 @@ if __name__ == "__main__":
"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "uuid": "57b1e4711e9e4a32b529f3132fc5931f",
} }
}) })
plate6 = PRCXI9300Trash(name="trash", size_x=50, size_y=50, size_z=10, category="trash") plate6 = PRCXI9300Trash(name="trash", size_x=50, size_y=500, size_z=10, category="trash")
plate6.load_state({ plate6.load_state({
"Material": { "Material": {
"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "uuid": "57b1e4711e9e4a32b529f3132fc5931f",
@@ -1079,27 +1079,28 @@ if __name__ == "__main__":
handler.set_tiprack([tip_rack]) # Set the tip rack for the handler handler.set_tiprack([tip_rack]) # Set the tip rack for the handler
asyncio.run(handler.setup()) # Initialize the handler and setup the connection asyncio.run(handler.setup()) # Initialize the handler and setup the connection
asyncio.run(handler.create_protocol(protocol_name="Test Protocol")) # Initialize the backend and setup the connection asyncio.run(handler.create_protocol(protocol_name="Test Protocol")) # Initialize the backend and setup the connection
#asyncio.run(handler.pick_up_tips(tip_rack.children[:8],[0,1,2,3,4,5,6,7])) asyncio.run(handler.pick_up_tips(tip_rack.children[:8],[0,1,2,3,4,5,6,7]))
# asyncio.run(handler.aspirate(well_containers.children[:8],[50]*8, [0,1,2,3,4,5,6,7])) # asyncio.run(handler.aspirate(well_containers.children[:8],[50]*8, [0,1,2,3,4,5,6,7]))
# asyncio.run(handler.dispense(well_containers.children[:8],[50]*8,[0,1,2,3,4,5,6,7])) # asyncio.run(handler.dispense(well_containers.children[:8],[50]*8,[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.drop_tips(tip_rack.children[8:16],[0,1,2,3,4,5,6,7]))
asyncio.run(handler.discard_tips())
# 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], # use_channels=[0, 1, 2, 3, 4, 5, 6, 7],
flow_rates=[None] * 32, # flow_rates=[None] * 32,
offsets=[Coordinate(0, 0, 0)] * 32, # offsets=[Coordinate(0, 0, 0)] * 32,
liquid_height=[None] * 16, # liquid_height=[None] * 16,
blow_out_air_volume=[None] * 16, # blow_out_air_volume=[None] * 16,
delays=None, # delays=None,
mix_time=3, # mix_time=3,
mix_vol=50, # mix_vol=50,
spread="wide", # spread="wide",
)) # ))
# asyncio.run(handler.remove_liquid( # asyncio.run(handler.remove_liquid(
# vols=[100]*16, # vols=[100]*16,