mirror of
https://github.com/dptech-corp/Uni-Lab-OS.git
synced 2025-12-17 13:01:12 +00:00
1421 lines
55 KiB
Python
1421 lines
55 KiB
Python
import asyncio
|
||
import collections
|
||
import contextlib
|
||
import json
|
||
import socket
|
||
import time
|
||
from typing import Any, List, Dict, Optional, TypedDict, Union, Sequence, Iterator, Literal
|
||
|
||
from pylabrobot.liquid_handling import (
|
||
LiquidHandlerBackend,
|
||
Pickup,
|
||
SingleChannelAspiration,
|
||
Drop,
|
||
SingleChannelDispense,
|
||
PickupTipRack,
|
||
DropTipRack,
|
||
MultiHeadAspirationPlate, ChatterBoxBackend, LiquidHandlerChatterboxBackend,
|
||
)
|
||
from pylabrobot.liquid_handling.standard import (
|
||
MultiHeadAspirationContainer,
|
||
MultiHeadDispenseContainer,
|
||
MultiHeadDispensePlate,
|
||
ResourcePickup,
|
||
ResourceMove,
|
||
ResourceDrop,
|
||
)
|
||
from pylabrobot.resources import Tip, Deck, Plate, Well, TipRack, Resource, Container, Coordinate, TipSpot, Trash
|
||
|
||
from unilabos.devices.liquid_handling.liquid_handler_abstract import LiquidHandlerAbstract
|
||
|
||
|
||
class PRCXIError(RuntimeError):
|
||
"""Lilith 返回 Success=false 时抛出的业务异常"""
|
||
|
||
|
||
class Material(TypedDict): # 和Plate同关系
|
||
uuid: str
|
||
Code: Optional[str]
|
||
Name: Optional[str]
|
||
SummaryName: Optional[str]
|
||
PipetteHeight: Optional[int]
|
||
materialEnum: Optional[int]
|
||
|
||
|
||
class WorkTablets(TypedDict):
|
||
Number: int
|
||
Code: str
|
||
Material: Dict[str, Any]
|
||
|
||
|
||
class MatrixInfo(TypedDict):
|
||
MatrixId: str
|
||
MatrixName: str
|
||
MatrixCount: int
|
||
WorkTablets: list[WorkTablets]
|
||
|
||
|
||
class PRCXI9300Deck(Deck):
|
||
"""PRCXI 9300 的专用 Deck 类，继承自 Deck。
|
||
|
||
该类定义了 PRCXI 9300 的工作台布局和槽位信息。
|
||
"""
|
||
|
||
def __init__(self, name: str, size_x: float, size_y: float, size_z: float):
|
||
super().__init__(name, size_x, size_y, size_z)
|
||
self.slots = [None] * 6 # PRCXI 9300 有 6 个槽位
|
||
|
||
|
||
class PRCXI9300Container(Plate, TipRack):
|
||
"""PRCXI 9300 的专用 Container 类，继承自 Plate和TipRack。
|
||
|
||
该类定义了 PRCXI 9300 的工作台布局和槽位信息。
|
||
"""
|
||
|
||
def __init__(self, name: str, size_x: float, size_y: float, size_z: float, category: str, ordering: collections.OrderedDict, model: Optional[str] = None,):
|
||
super().__init__(name, size_x, size_y, size_z, category=category, ordering=ordering, model=model)
|
||
self._unilabos_state = {}
|
||
|
||
def load_state(self, state: Dict[str, Any]) -> None:
|
||
"""从给定的状态加载工作台信息。"""
|
||
super().load_state(state)
|
||
self._unilabos_state = state
|
||
|
||
def serialize_state(self) -> Dict[str, Dict[str, Any]]:
|
||
data = super().serialize_state()
|
||
data.update(self._unilabos_state)
|
||
return data
|
||
|
||
|
||
class PRCXI9300Trash(Trash):
|
||
"""PRCXI 9300 的专用 Trash 类，继承自 Trash。
|
||
|
||
该类定义了 PRCXI 9300 的工作台布局和槽位信息。
|
||
"""
|
||
|
||
def __init__(self, name: str, size_x: float, size_y: float, size_z: float, category: str, **kwargs):
|
||
if name != "trash":
|
||
name = "trash"
|
||
print("PRCXI9300Trash name must be 'trash', using 'trash' instead.")
|
||
super().__init__(name, size_x, size_y, size_z, category=category, **kwargs)
|
||
self._unilabos_state = {}
|
||
|
||
def load_state(self, state: Dict[str, Any]) -> None:
|
||
"""从给定的状态加载工作台信息。"""
|
||
#super().load_state(state)
|
||
self._unilabos_state = state
|
||
|
||
def serialize_state(self) -> Dict[str, Dict[str, Any]]:
|
||
data = super().serialize_state()
|
||
data.update(self._unilabos_state)
|
||
return data
|
||
|
||
class PRCXI9300Handler(LiquidHandlerAbstract):
|
||
support_touch_tip = False
|
||
|
||
@property
|
||
def reset_ok(self) -> bool:
|
||
"""检查设备是否已重置成功。"""
|
||
if self._unilabos_backend.debug:
|
||
return True
|
||
return self._unilabos_backend.is_reset_ok
|
||
|
||
def __init__(self, deck: Deck, host: str, port: int, timeout: float, channel_num=8, axis="Left", setup=True, debug=False, simulator=False, matrix_id=""):
|
||
tablets_info = []
|
||
count = 0
|
||
for child in deck.children:
|
||
if "Material" in child._unilabos_state:
|
||
count += 1
|
||
tablets_info.append(
|
||
WorkTablets(Number=count, Code=f"T{count}", Material=child._unilabos_state["Material"])
|
||
)
|
||
self._unilabos_backend = PRCXI9300Backend(tablets_info, host, port, timeout, channel_num, axis, setup, debug, matrix_id)
|
||
super().__init__(backend=self._unilabos_backend, deck=deck, simulator=simulator, channel_num=channel_num)
|
||
|
||
def set_liquid(self, wells: list[Well], liquid_names: list[str], volumes: list[float]):
|
||
return super().set_liquid(wells, liquid_names, volumes)
|
||
|
||
def set_group(self, group_name: str, wells: List[Well], volumes: List[float]):
|
||
return super().set_group(group_name, wells, volumes)
|
||
|
||
async def transfer_group(self, source_group_name: str, target_group_name: str, unit_volume: float):
|
||
return await super().transfer_group(source_group_name, target_group_name, unit_volume)
|
||
|
||
async def create_protocol(
|
||
self,
|
||
protocol_name: str = "",
|
||
protocol_description: str = "",
|
||
protocol_version: str = "",
|
||
protocol_author: str = "",
|
||
protocol_date: str = "",
|
||
protocol_type: str = "",
|
||
none_keys: List[str] = [],
|
||
):
|
||
self._unilabos_backend.create_protocol(protocol_name)
|
||
|
||
async def run_protocol(self):
|
||
return self._unilabos_backend.run_protocol()
|
||
|
||
async def remove_liquid(
|
||
self,
|
||
vols: List[float],
|
||
sources: Sequence[Container],
|
||
waste_liquid: Optional[Container] = None,
|
||
*,
|
||
use_channels: Optional[List[int]] = None,
|
||
flow_rates: Optional[List[Optional[float]]] = None,
|
||
offsets: Optional[List[Coordinate]] = None,
|
||
liquid_height: Optional[List[Optional[float]]] = None,
|
||
blow_out_air_volume: Optional[List[Optional[float]]] = None,
|
||
spread: Optional[Literal["wide", "tight", "custom"]] = "wide",
|
||
delays: Optional[List[int]] = None,
|
||
is_96_well: Optional[bool] = False,
|
||
top: Optional[List[float]] = None,
|
||
none_keys: List[str] = [],
|
||
):
|
||
return await super().remove_liquid(
|
||
vols,
|
||
sources,
|
||
waste_liquid,
|
||
use_channels=use_channels,
|
||
flow_rates=flow_rates,
|
||
offsets=offsets,
|
||
liquid_height=liquid_height,
|
||
blow_out_air_volume=blow_out_air_volume,
|
||
spread=spread,
|
||
delays=delays,
|
||
is_96_well=is_96_well,
|
||
top=top,
|
||
none_keys=none_keys,
|
||
)
|
||
|
||
async def add_liquid(
|
||
self,
|
||
asp_vols: Union[List[float], float],
|
||
dis_vols: Union[List[float], float],
|
||
reagent_sources: Sequence[Container],
|
||
targets: Sequence[Container],
|
||
*,
|
||
use_channels: Optional[List[int]] = None,
|
||
flow_rates: Optional[List[Optional[float]]] = None,
|
||
offsets: Optional[List[Coordinate]] = None,
|
||
liquid_height: Optional[List[Optional[float]]] = None,
|
||
blow_out_air_volume: Optional[List[Optional[float]]] = None,
|
||
spread: Optional[Literal["wide", "tight", "custom"]] = "wide",
|
||
is_96_well: bool = False,
|
||
delays: Optional[List[int]] = None,
|
||
mix_time: Optional[int] = None,
|
||
mix_vol: Optional[int] = None,
|
||
mix_rate: Optional[int] = None,
|
||
mix_liquid_height: Optional[float] = None,
|
||
none_keys: List[str] = [],
|
||
):
|
||
return await super().add_liquid(
|
||
asp_vols,
|
||
dis_vols,
|
||
reagent_sources,
|
||
targets,
|
||
use_channels=use_channels,
|
||
flow_rates=flow_rates,
|
||
offsets=offsets,
|
||
liquid_height=liquid_height,
|
||
blow_out_air_volume=blow_out_air_volume,
|
||
spread=spread,
|
||
is_96_well=is_96_well,
|
||
delays=delays,
|
||
mix_time=mix_time,
|
||
mix_vol=mix_vol,
|
||
mix_rate=mix_rate,
|
||
mix_liquid_height=mix_liquid_height,
|
||
none_keys=none_keys,
|
||
)
|
||
|
||
async def transfer_liquid(
|
||
self,
|
||
sources: Sequence[Container],
|
||
targets: Sequence[Container],
|
||
tip_racks: Sequence[TipRack],
|
||
*,
|
||
use_channels: Optional[List[int]] = None,
|
||
asp_vols: Union[List[float], float],
|
||
dis_vols: Union[List[float], float],
|
||
asp_flow_rates: Optional[List[Optional[float]]] = None,
|
||
dis_flow_rates: Optional[List[Optional[float]]] = None,
|
||
offsets: Optional[List[Coordinate]] = None,
|
||
touch_tip: bool = False,
|
||
liquid_height: Optional[List[Optional[float]]] = None,
|
||
blow_out_air_volume: Optional[List[Optional[float]]] = None,
|
||
spread: Literal["wide", "tight", "custom"] = "wide",
|
||
is_96_well: bool = False,
|
||
mix_stage: Optional[Literal["none", "before", "after", "both"]] = "none",
|
||
mix_times: Optional[List[int]] = None,
|
||
mix_vol: Optional[int] = None,
|
||
mix_rate: Optional[int] = None,
|
||
mix_liquid_height: Optional[float] = None,
|
||
delays: Optional[List[int]] = None,
|
||
none_keys: List[str] = [],
|
||
):
|
||
return await super().transfer_liquid(
|
||
sources,
|
||
targets,
|
||
tip_racks,
|
||
use_channels=use_channels,
|
||
asp_vols=asp_vols,
|
||
dis_vols=dis_vols,
|
||
asp_flow_rates=asp_flow_rates,
|
||
dis_flow_rates=dis_flow_rates,
|
||
offsets=offsets,
|
||
touch_tip=touch_tip,
|
||
liquid_height=liquid_height,
|
||
blow_out_air_volume=blow_out_air_volume,
|
||
spread=spread,
|
||
is_96_well=is_96_well,
|
||
mix_stage=mix_stage,
|
||
mix_times=mix_times,
|
||
mix_vol=mix_vol,
|
||
mix_rate=mix_rate,
|
||
mix_liquid_height=mix_liquid_height,
|
||
delays=delays,
|
||
none_keys=none_keys,
|
||
)
|
||
|
||
async def custom_delay(self, seconds=0, msg=None):
|
||
return await super().custom_delay(seconds, msg)
|
||
|
||
async def touch_tip(self, targets: Sequence[Container]):
|
||
return await super().touch_tip(targets)
|
||
|
||
async def mix(
|
||
self,
|
||
targets: Sequence[Container],
|
||
mix_time: int = None,
|
||
mix_vol: Optional[int] = None,
|
||
height_to_bottom: Optional[float] = None,
|
||
offsets: Optional[Coordinate] = None,
|
||
mix_rate: Optional[float] = None,
|
||
none_keys: List[str] = [],
|
||
):
|
||
return await self._unilabos_backend.mix(targets, mix_time, mix_vol, height_to_bottom, offsets, mix_rate, none_keys)
|
||
|
||
def iter_tips(self, tip_racks: Sequence[TipRack]) -> Iterator[Resource]:
|
||
return super().iter_tips(tip_racks)
|
||
|
||
async def pick_up_tips(self, tip_spots: List[TipSpot], use_channels: Optional[List[int]] = None,
|
||
offsets: Optional[List[Coordinate]] = None, **backend_kwargs):
|
||
return await super().pick_up_tips(tip_spots, use_channels, offsets, **backend_kwargs)
|
||
|
||
async def aspirate(self, resources: Sequence[Container], vols: List[float],
|
||
use_channels: Optional[List[int]] = None, flow_rates: Optional[List[Optional[float]]] = None,
|
||
offsets: Optional[List[Coordinate]] = None,
|
||
liquid_height: Optional[List[Optional[float]]] = None,
|
||
blow_out_air_volume: Optional[List[Optional[float]]] = None,
|
||
spread: Literal["wide", "tight", "custom"] = "wide", **backend_kwargs):
|
||
|
||
return await super().aspirate(resources, vols, use_channels, flow_rates, offsets, liquid_height,
|
||
blow_out_air_volume, spread, **backend_kwargs)
|
||
|
||
async def drop_tips(self, tip_spots: Sequence[Union[TipSpot, Trash]], use_channels: Optional[List[int]] = None,
|
||
offsets: Optional[List[Coordinate]] = None, allow_nonzero_volume: bool = False,
|
||
**backend_kwargs):
|
||
return await super().drop_tips(tip_spots, use_channels, offsets, allow_nonzero_volume, **backend_kwargs)
|
||
|
||
async def dispense(self, resources: Sequence[Container], vols: List[float],
|
||
use_channels: Optional[List[int]] = None, flow_rates: Optional[List[Optional[float]]] = None,
|
||
offsets: Optional[List[Coordinate]] = None,
|
||
liquid_height: Optional[List[Optional[float]]] = None,
|
||
blow_out_air_volume: Optional[List[Optional[float]]] = None,
|
||
spread: Literal["wide", "tight", "custom"] = "wide", **backend_kwargs):
|
||
return await super().dispense(resources, vols, use_channels, flow_rates, offsets, liquid_height,
|
||
blow_out_air_volume, spread, **backend_kwargs)
|
||
|
||
async def discard_tips(self, use_channels: Optional[List[int]] = None, allow_nonzero_volume: bool = True,
|
||
offsets: Optional[List[Coordinate]] = None, **backend_kwargs):
|
||
return await super().discard_tips(use_channels, allow_nonzero_volume, offsets, **backend_kwargs)
|
||
|
||
def set_tiprack(self, tip_racks: Sequence[TipRack]):
|
||
super().set_tiprack(tip_racks)
|
||
|
||
async def move_to(self, well: Well, dis_to_top: float = 0, channel: int = 0):
|
||
return await super().move_to(well, dis_to_top, channel)
|
||
|
||
|
||
class PRCXI9300Backend(LiquidHandlerBackend):
|
||
"""PRCXI 9300 的后端实现，继承自 LiquidHandlerBackend。
|
||
|
||
该类提供了与 PRCXI 9300 设备进行通信的基本方法，包括方案管理、自动化控制、运行状态查询等。
|
||
"""
|
||
|
||
_num_channels = 8 # 默认通道数为 8
|
||
_is_reset_ok = False
|
||
|
||
@property
|
||
def is_reset_ok(self) -> bool:
|
||
self._is_reset_ok = self.api_client.get_reset_status()
|
||
return self._is_reset_ok
|
||
|
||
matrix_info: MatrixInfo
|
||
protocol_name: str
|
||
steps_todo_list = []
|
||
|
||
def __init__(
|
||
self,
|
||
tablets_info: list[WorkTablets],
|
||
host: str = "127.0.0.1",
|
||
port: int = 9999,
|
||
timeout: float = 10.0,
|
||
channel_num: int=8,
|
||
axis: str="Left",
|
||
setup=True,
|
||
debug=False,
|
||
matrix_id="",
|
||
) -> None:
|
||
super().__init__()
|
||
self.tablets_info = tablets_info
|
||
self.matrix_id = matrix_id
|
||
self.api_client = PRCXI9300Api(host, port, timeout, axis, debug)
|
||
self.host, self.port, self.timeout = host, port, timeout
|
||
self._num_channels = channel_num
|
||
self._execute_setup = setup
|
||
self.debug = debug
|
||
|
||
def create_protocol(self, protocol_name):
|
||
self.protocol_name = protocol_name
|
||
self.steps_todo_list = []
|
||
|
||
def run_protocol(self):
|
||
assert self.is_reset_ok, "PRCXI9300Backend is not reset successfully. Please call setup() first."
|
||
run_time = time.time()
|
||
self.matrix_info = MatrixInfo(
|
||
MatrixId=f"{int(run_time)}",
|
||
MatrixName=f"protocol_{run_time}",
|
||
MatrixCount=len(self.tablets_info),
|
||
WorkTablets=self.tablets_info,
|
||
)
|
||
#print(json.dumps(self.matrix_info, indent=2))
|
||
if not len(self.matrix_id):
|
||
res = self.api_client.add_WorkTablet_Matrix(self.matrix_info)
|
||
assert res["Success"], f"Failed to create matrix: {res.get('Message', 'Unknown error')}"
|
||
print(f"PRCXI9300Backend created matrix with ID: {self.matrix_info['MatrixId']}, result: {res}")
|
||
solution_id = self.api_client.add_solution(
|
||
f"protocol_{run_time}", self.matrix_info["MatrixId"], self.steps_todo_list
|
||
)
|
||
else:
|
||
print(f"PRCXI9300Backend using predefined worktable {self.matrix_id}, skipping matrix creation.")
|
||
solution_id = self.api_client.add_solution(
|
||
f"protocol_{run_time}", self.matrix_id, self.steps_todo_list
|
||
)
|
||
print(f"PRCXI9300Backend created solution with ID: {solution_id}")
|
||
self.api_client.load_solution(solution_id)
|
||
print(json.dumps(self.steps_todo_list, indent=2))
|
||
return self.api_client.start()
|
||
|
||
@classmethod
|
||
def check_channels(cls, use_channels: List[int]) -> List[int]:
|
||
"""检查通道是否符合要求，PRCXI9300Backend 只支持所有 8 个通道。"""
|
||
if use_channels != [0, 1, 2, 3, 4, 5, 6, 7]:
|
||
print("PRCXI9300Backend only supports all 8 channels, using default [0, 1, 2, 3, 4, 5, 6, 7].")
|
||
return [0, 1, 2, 3, 4, 5, 6, 7]
|
||
return use_channels
|
||
|
||
async def setup(self):
|
||
await super().setup()
|
||
try:
|
||
if self._execute_setup:
|
||
self.api_client.call("IAutomation", "Reset")
|
||
while not self.is_reset_ok:
|
||
print("Waiting for PRCXI9300 to reset...")
|
||
await asyncio.sleep(1)
|
||
print("PRCXI9300 reset successfully.")
|
||
except ConnectionRefusedError as e:
|
||
raise RuntimeError(
|
||
f"Failed to connect to PRCXI9300 API at {self.host}:{self.port}. "
|
||
"Please ensure the PRCXI9300 service is running."
|
||
) from e
|
||
|
||
async def stop(self):
|
||
self.api_client.call("IAutomation", "Stop")
|
||
|
||
async def pick_up_tips(self, ops: List[Pickup], use_channels: List[int] = None):
|
||
"""Pick up tips from the specified resource."""
|
||
|
||
plate_indexes = []
|
||
for op in ops:
|
||
plate = op.resource.parent
|
||
deck = plate.parent
|
||
plate_index = deck.children.index(plate)
|
||
# print(f"Plate index: {plate_index}, Plate name: {plate.name}")
|
||
# print(f"Number of children in deck: {len(deck.children)}")
|
||
|
||
plate_indexes.append(plate_index)
|
||
|
||
if len(set(plate_indexes)) != 1:
|
||
raise ValueError("All pickups must be from the same plate. Found different plates: " + str(plate_indexes))
|
||
|
||
tip_columns = []
|
||
for op in ops:
|
||
tipspot = op.resource
|
||
tipspot_index = tipspot.parent.children.index(tipspot)
|
||
tip_columns.append(tipspot_index // 8)
|
||
if len(set(tip_columns)) != 1:
|
||
raise ValueError("All pickups must be from the same tip column. Found different columns: " + str(tip_columns))
|
||
PlateNo = plate_indexes[0] + 1
|
||
hole_col = tip_columns[0] + 1
|
||
hole_row = 1
|
||
if self._num_channels == 1:
|
||
hole_row = tipspot_index % 8 + 1
|
||
|
||
step = self.api_client.Load(dosage=0, plate_no=PlateNo, is_whole_plate=False, hole_row=hole_row, hole_col=hole_col,
|
||
blending_times=0, balance_height=0, plate_or_hole=f"H{hole_col}-8,T{PlateNo}",
|
||
hole_numbers="1,2,3,4,5,6,7,8")
|
||
self.steps_todo_list.append(step)
|
||
|
||
async def drop_tips(self, ops: List[Drop], use_channels: List[int] = None):
|
||
"""Pick up tips from the specified resource."""
|
||
|
||
# 检查trash #
|
||
if ops[0].resource.name == "trash":
|
||
|
||
PlateNo = ops[0].resource.parent.children.index(ops[0].resource) + 1
|
||
|
||
step = self.api_client.UnLoad(
|
||
dosage=0,
|
||
plate_no=PlateNo,
|
||
is_whole_plate=False,
|
||
hole_row=1,
|
||
hole_col=3,
|
||
blending_times=0,
|
||
balance_height=0,
|
||
plate_or_hole=f"H{1}-8,T{PlateNo}",
|
||
hole_numbers="1,2,3,4,5,6,7,8",
|
||
)
|
||
self.steps_todo_list.append(step)
|
||
return
|
||
#print(ops[0].resource.parent.children.index(ops[0].resource))
|
||
|
||
|
||
plate_indexes = []
|
||
for op in ops:
|
||
plate = op.resource.parent
|
||
deck = plate.parent
|
||
plate_index = deck.children.index(plate)
|
||
plate_indexes.append(plate_index)
|
||
if len(set(plate_indexes)) != 1:
|
||
raise ValueError("All drop_tips must be from the same plate. Found different plates: " + str(plate_indexes))
|
||
|
||
tip_columns = []
|
||
for op in ops:
|
||
tipspot = op.resource
|
||
tipspot_index = tipspot.parent.children.index(tipspot)
|
||
tip_columns.append(tipspot_index // 8)
|
||
if len(set(tip_columns)) != 1:
|
||
raise ValueError("All drop_tips must be from the same tip column. Found different columns: " + str(tip_columns))
|
||
|
||
PlateNo = plate_indexes[0] + 1
|
||
hole_col = tip_columns[0] + 1
|
||
|
||
if self.channel_num == 1:
|
||
hole_row = tipspot_index % 8 + 1
|
||
|
||
step = self.api_client.UnLoad(
|
||
dosage=0,
|
||
plate_no=PlateNo,
|
||
is_whole_plate=False,
|
||
hole_row=hole_row,
|
||
hole_col=hole_col,
|
||
blending_times=0,
|
||
balance_height=0,
|
||
plate_or_hole=f"H{hole_col}-8,T{PlateNo}",
|
||
hole_numbers="1,2,3,4,5,6,7,8",
|
||
)
|
||
self.steps_todo_list.append(step)
|
||
|
||
async def mix(
|
||
self,
|
||
targets: Sequence[Container],
|
||
mix_time: int = None,
|
||
mix_vol: Optional[int] = None,
|
||
height_to_bottom: Optional[float] = None,
|
||
offsets: Optional[Coordinate] = None,
|
||
mix_rate: Optional[float] = None,
|
||
none_keys: List[str] = [],
|
||
):
|
||
|
||
"""Mix liquid in the specified resources."""
|
||
|
||
plate_indexes = []
|
||
for op in targets:
|
||
deck = op.parent.parent
|
||
plate = op.parent
|
||
plate_index = deck.children.index(plate)
|
||
plate_indexes.append(plate_index)
|
||
|
||
if len(set(plate_indexes)) != 1:
|
||
raise ValueError("All pickups must be from the same plate. Found different plates: " + str(plate_indexes))
|
||
|
||
tip_columns = []
|
||
for op in targets:
|
||
tipspot_index = op.parent.children.index(op)
|
||
tip_columns.append(tipspot_index // 8)
|
||
|
||
if len(set(tip_columns)) != 1:
|
||
raise ValueError("All pickups must be from the same tip column. Found different columns: " + str(tip_columns))
|
||
|
||
PlateNo = plate_indexes[0] + 1
|
||
hole_col = tip_columns[0] + 1
|
||
hole_row = 1
|
||
if self.num_channels == 1:
|
||
hole_row = tipspot_index % 8 + 1
|
||
|
||
assert mix_time > 0
|
||
step = self.api_client.Blending(
|
||
dosage=mix_vol,
|
||
plate_no=PlateNo,
|
||
is_whole_plate=False,
|
||
hole_row=hole_row,
|
||
hole_col=hole_col,
|
||
blending_times=mix_time,
|
||
balance_height=0,
|
||
plate_or_hole=f"H{hole_col}-8,T{PlateNo}",
|
||
hole_numbers="1,2,3,4,5,6,7,8",
|
||
)
|
||
self.steps_todo_list.append(step)
|
||
|
||
async def aspirate(self, ops: List[SingleChannelAspiration], use_channels: List[int] = None):
|
||
|
||
"""Aspirate liquid from the specified resources."""
|
||
|
||
plate_indexes = []
|
||
for op in ops:
|
||
plate = op.resource.parent
|
||
deck = plate.parent
|
||
plate_index = deck.children.index(plate)
|
||
plate_indexes.append(plate_index)
|
||
|
||
if len(set(plate_indexes)) != 1:
|
||
raise ValueError("All pickups must be from the same plate. Found different plates: " + str(plate_indexes))
|
||
|
||
tip_columns = []
|
||
for op in ops:
|
||
tipspot = op.resource
|
||
tipspot_index = tipspot.parent.children.index(tipspot)
|
||
tip_columns.append(tipspot_index // 8)
|
||
|
||
if len(set(tip_columns)) != 1:
|
||
raise ValueError("All pickups must be from the same tip column. Found different columns: " + str(tip_columns))
|
||
|
||
volumes = [op.volume for op in ops]
|
||
if len(set(volumes)) != 1:
|
||
raise ValueError("All aspirate volumes must be the same. Found different volumes: " + str(volumes))
|
||
|
||
PlateNo = plate_indexes[0] + 1
|
||
hole_col = tip_columns[0] + 1
|
||
hole_row = 1
|
||
if self.num_channels == 1:
|
||
hole_row = tipspot_index % 8 + 1
|
||
|
||
step = self.api_client.Imbibing(dosage=int(volumes[0]), plate_no=PlateNo, is_whole_plate=False, hole_row=hole_row,
|
||
hole_col=hole_col, blending_times=0, balance_height=0,
|
||
plate_or_hole=f"H{hole_col}-8,T{PlateNo}", hole_numbers="1,2,3,4,5,6,7,8")
|
||
self.steps_todo_list.append(step)
|
||
|
||
|
||
|
||
async def dispense(self, ops: List[SingleChannelDispense], use_channels: List[int] = None):
|
||
|
||
"""Dispense liquid into the specified resources."""
|
||
|
||
plate_indexes = []
|
||
for op in ops:
|
||
plate = op.resource.parent
|
||
deck = plate.parent
|
||
plate_index = deck.children.index(plate)
|
||
plate_indexes.append(plate_index)
|
||
|
||
if len(set(plate_indexes)) != 1:
|
||
raise ValueError("All dispense must be from the same plate. Found different plates: " + str(plate_indexes))
|
||
|
||
tip_columns = []
|
||
for op in ops:
|
||
tipspot = op.resource
|
||
tipspot_index = tipspot.parent.children.index(tipspot)
|
||
tip_columns.append(tipspot_index // 8)
|
||
|
||
if len(set(tip_columns)) != 1:
|
||
raise ValueError("All dispense must be from the same tip column. Found different columns: " + str(tip_columns))
|
||
|
||
volumes = [op.volume for op in ops]
|
||
if len(set(volumes)) != 1:
|
||
raise ValueError("All dispense volumes must be the same. Found different volumes: " + str(volumes))
|
||
|
||
PlateNo = plate_indexes[0] + 1
|
||
hole_col = tip_columns[0] + 1
|
||
|
||
hole_row = 1
|
||
if self.num_channels == 1:
|
||
hole_row = tipspot_index % 8 + 1
|
||
|
||
step = self.api_client.Tapping(
|
||
dosage=int(volumes[0]),
|
||
plate_no=PlateNo,
|
||
is_whole_plate=False,
|
||
hole_row=hole_row,
|
||
hole_col=hole_col,
|
||
blending_times=0,
|
||
balance_height=0,
|
||
plate_or_hole=f"H{hole_col}-8,T{PlateNo}",
|
||
hole_numbers="1,2,3,4,5,6,7,8",
|
||
)
|
||
self.steps_todo_list.append(step)
|
||
|
||
async def pick_up_tips96(self, pickup: PickupTipRack):
|
||
raise NotImplementedError("The PRCXI backend does not support the 96 head.")
|
||
|
||
async def drop_tips96(self, drop: DropTipRack):
|
||
raise NotImplementedError("The PRCXI backend does not support the 96 head.")
|
||
|
||
async def aspirate96(self, aspiration: Union[MultiHeadAspirationPlate, MultiHeadAspirationContainer]):
|
||
raise NotImplementedError("The Opentrons backend does not support the 96 head.")
|
||
|
||
async def dispense96(self, dispense: Union[MultiHeadDispensePlate, MultiHeadDispenseContainer]):
|
||
raise NotImplementedError("The Opentrons backend does not support the 96 head.")
|
||
|
||
async def pick_up_resource(self, pickup: ResourcePickup):
|
||
raise NotImplementedError("The Opentrons backend does not support the robotic arm.")
|
||
|
||
async def move_picked_up_resource(self, move: ResourceMove):
|
||
raise NotImplementedError("The Opentrons backend does not support the robotic arm.")
|
||
|
||
async def drop_resource(self, drop: ResourceDrop):
|
||
raise NotImplementedError("The Opentrons backend does not support the robotic arm.")
|
||
|
||
def can_pick_up_tip(self, channel_idx: int, tip: Tip) -> bool:
|
||
return True # PRCXI9300Backend does not have tip compatibility issues
|
||
|
||
def serialize(self) -> dict:
|
||
raise NotImplementedError()
|
||
|
||
@property
|
||
def num_channels(self) -> int:
|
||
return self._num_channels
|
||
|
||
|
||
class PRCXI9300Api:
|
||
def __init__(self, host: str = "127.0.0.1", port: int = 9999, timeout: float = 10.0, axis="Left", debug: bool = False) -> None:
|
||
self.host, self.port, self.timeout = host, port, timeout
|
||
self.debug = debug
|
||
self.axis = axis
|
||
|
||
@staticmethod
|
||
def _len_prefix(n: int) -> bytes:
|
||
return bytes.fromhex(format(n, "016x"))
|
||
|
||
def _raw_request(self, payload: str) -> str:
|
||
if self.debug:
|
||
return " "
|
||
with contextlib.closing(socket.socket()) as sock:
|
||
sock.settimeout(self.timeout)
|
||
sock.connect((self.host, self.port))
|
||
data = payload.encode()
|
||
sock.sendall(self._len_prefix(len(data)) + data)
|
||
|
||
chunks, first = [], True
|
||
while True:
|
||
chunk = sock.recv(4096)
|
||
if not chunk:
|
||
break
|
||
if first:
|
||
chunk, first = chunk[8:], False
|
||
chunks.append(chunk)
|
||
return b"".join(chunks).decode()
|
||
|
||
# ---------------------------------------------------- 方案相关（ISolution）
|
||
def list_solutions(self) -> List[Dict[str, Any]]:
|
||
"""GetSolutionList"""
|
||
return self.call("ISolution", "GetSolutionList")
|
||
|
||
def load_solution(self, solution_id: str) -> bool:
|
||
"""LoadSolution"""
|
||
return self.call("ISolution", "LoadSolution", [solution_id])
|
||
|
||
def add_solution(self, name: str, matrix_id: str, steps: List[Dict[str, Any]]) -> str:
|
||
"""AddSolution → 返回新方案 GUID"""
|
||
return self.call("ISolution", "AddSolution", [name, matrix_id, steps])
|
||
|
||
# ---------------------------------------------------- 自动化控制（IAutomation）
|
||
def start(self) -> bool:
|
||
return self.call("IAutomation", "Start")
|
||
|
||
def call(self, service: str, method: str, params: Optional[list] = None) -> Any:
|
||
payload = json.dumps(
|
||
{"ServiceName": service, "MethodName": method, "Paramters": params or []}, separators=(",", ":")
|
||
)
|
||
resp = json.loads(self._raw_request(payload))
|
||
if not resp.get("Success", False):
|
||
raise PRCXIError(resp.get("Msg", "Unknown error"))
|
||
data = resp.get("Data")
|
||
try:
|
||
return json.loads(data)
|
||
except (TypeError, json.JSONDecodeError):
|
||
return data
|
||
|
||
def pause(self) -> bool:
|
||
"""Pause"""
|
||
return self.call("IAutomation", "Pause")
|
||
|
||
def resume(self) -> bool:
|
||
"""Resume"""
|
||
return self.call("IAutomation", "Resume")
|
||
|
||
def get_error_code(self) -> Optional[str]:
|
||
"""GetErrorCode"""
|
||
return self.call("IAutomation", "GetErrorCode")
|
||
|
||
def get_reset_status(self) -> bool:
|
||
"""GetErrorCode"""
|
||
if self.debug:
|
||
return True
|
||
res = self.call("IAutomation", "GetResetStatus")
|
||
return not res
|
||
|
||
def clear_error_code(self) -> bool:
|
||
"""RemoveErrorCodet"""
|
||
return self.call("IAutomation", "RemoveErrorCodet")
|
||
|
||
# ---------------------------------------------------- 运行状态（IMachineState）
|
||
def step_state_list(self) -> List[Dict[str, Any]]:
|
||
"""GetStepStateList"""
|
||
return self.call("IMachineState", "GetStepStateList")
|
||
|
||
def step_status(self, seq_num: int) -> Dict[str, Any]:
|
||
"""GetStepStatus"""
|
||
return self.call("IMachineState", "GetStepStatus", [seq_num])
|
||
|
||
def step_state(self, seq_num: int) -> Dict[str, Any]:
|
||
"""GetStepState"""
|
||
return self.call("IMachineState", "GetStepState", [seq_num])
|
||
|
||
def axis_location(self, axis_num: int = 1) -> Dict[str, Any]:
|
||
"""GetLocation"""
|
||
return self.call("IMachineState", "GetLocation", [axis_num])
|
||
|
||
# ---------------------------------------------------- 版位矩阵（IMatrix）
|
||
def list_matrices(self) -> List[Dict[str, Any]]:
|
||
"""GetWorkTabletMatrices"""
|
||
return self.call("IMatrix", "GetWorkTabletMatrices")
|
||
|
||
def matrix_by_id(self, matrix_id: str) -> Dict[str, Any]:
|
||
"""GetWorkTabletMatrixById"""
|
||
return self.call("IMatrix", "GetWorkTabletMatrixById", [matrix_id])
|
||
|
||
def add_WorkTablet_Matrix(self, matrix: MatrixInfo):
|
||
return self.call("IMatrix", "AddWorkTabletMatrix", [matrix])
|
||
|
||
def Load(self, dosage: int, plate_no: int, is_whole_plate: bool, hole_row: int, hole_col: int, blending_times: int,
|
||
balance_height: int, plate_or_hole: str, hole_numbers: str, assist_fun1: str = "", assist_fun2: str = "",
|
||
assist_fun3: str = "", assist_fun4: str = "", assist_fun5: str = "",
|
||
liquid_method: str = "NormalDispense") -> Dict[str, Any]:
|
||
return {
|
||
"StepAxis": self.axis,
|
||
"Function": "Load",
|
||
"DosageNum": dosage,
|
||
"PlateNo": plate_no,
|
||
"IsWholePlate": is_whole_plate,
|
||
"HoleRow": hole_row,
|
||
"HoleCol": hole_col,
|
||
"BlendingTimes": blending_times,
|
||
"BalanceHeight": balance_height,
|
||
"PlateOrHoleNum": plate_or_hole,
|
||
"AssistFun1": assist_fun1,
|
||
"AssistFun2": assist_fun2,
|
||
"AssistFun3": assist_fun3,
|
||
"AssistFun4": assist_fun4,
|
||
"AssistFun5": assist_fun5,
|
||
"HoleNumbers": hole_numbers,
|
||
"LiquidDispensingMethod": liquid_method,
|
||
}
|
||
|
||
def Imbibing(self, dosage: int, plate_no: int, is_whole_plate: bool, hole_row: int, hole_col: int,
|
||
blending_times: int, balance_height: int, plate_or_hole: str, hole_numbers: str, assist_fun1: str = "",
|
||
assist_fun2: str = "", assist_fun3: str = "", assist_fun4: str = "", assist_fun5: str = "",
|
||
liquid_method: str = "NormalDispense") -> Dict[str, Any]:
|
||
return {
|
||
"StepAxis": self.axis,
|
||
"Function": "Imbibing",
|
||
"DosageNum": dosage,
|
||
"PlateNo": plate_no,
|
||
"IsWholePlate": is_whole_plate,
|
||
"HoleRow": hole_row,
|
||
"HoleCol": hole_col,
|
||
"BlendingTimes": blending_times,
|
||
"BalanceHeight": balance_height,
|
||
"PlateOrHoleNum": plate_or_hole,
|
||
"AssistFun1": assist_fun1,
|
||
"AssistFun2": assist_fun2,
|
||
"AssistFun3": assist_fun3,
|
||
"AssistFun4": assist_fun4,
|
||
"AssistFun5": assist_fun5,
|
||
"HoleNumbers": hole_numbers,
|
||
"LiquidDispensingMethod": liquid_method,
|
||
}
|
||
|
||
def Tapping(
|
||
self,
|
||
dosage: int,
|
||
plate_no: int,
|
||
is_whole_plate: bool,
|
||
hole_row: int,
|
||
hole_col: int,
|
||
blending_times: int,
|
||
balance_height: int,
|
||
plate_or_hole: str,
|
||
hole_numbers: str,
|
||
assist_fun1: str = "",
|
||
assist_fun2: str = "",
|
||
assist_fun3: str = "",
|
||
assist_fun4: str = "",
|
||
assist_fun5: str = "",
|
||
liquid_method: str = "NormalDispense",
|
||
) -> Dict[str, Any]:
|
||
return {
|
||
"StepAxis": self.axis,
|
||
"Function": "Tapping",
|
||
"DosageNum": dosage,
|
||
"PlateNo": plate_no,
|
||
"IsWholePlate": is_whole_plate,
|
||
"HoleRow": hole_row,
|
||
"HoleCol": hole_col,
|
||
"BlendingTimes": blending_times,
|
||
"BalanceHeight": balance_height,
|
||
"PlateOrHoleNum": plate_or_hole,
|
||
"AssistFun1": assist_fun1,
|
||
"AssistFun2": assist_fun2,
|
||
"AssistFun3": assist_fun3,
|
||
"AssistFun4": assist_fun4,
|
||
"AssistFun5": assist_fun5,
|
||
"HoleNumbers": hole_numbers,
|
||
"LiquidDispensingMethod": liquid_method,
|
||
}
|
||
|
||
def Blending(
|
||
self,
|
||
dosage: int,
|
||
plate_no: int,
|
||
is_whole_plate: bool,
|
||
hole_row: int,
|
||
hole_col: int,
|
||
blending_times: int,
|
||
balance_height: int,
|
||
plate_or_hole: str,
|
||
hole_numbers: str,
|
||
assist_fun1: str = "",
|
||
assist_fun2: str = "",
|
||
assist_fun3: str = "",
|
||
assist_fun4: str = "",
|
||
assist_fun5: str = "",
|
||
liquid_method: str = "NormalDispense",
|
||
) -> Dict[str, Any]:
|
||
return {
|
||
"StepAxis": self.axis,
|
||
"Function": "Blending",
|
||
"DosageNum": dosage,
|
||
"PlateNo": plate_no,
|
||
"IsWholePlate": is_whole_plate,
|
||
"HoleRow": hole_row,
|
||
"HoleCol": hole_col,
|
||
"BlendingTimes": blending_times,
|
||
"BalanceHeight": balance_height,
|
||
"PlateOrHoleNum": plate_or_hole,
|
||
"AssistFun1": assist_fun1,
|
||
"AssistFun2": assist_fun2,
|
||
"AssistFun3": assist_fun3,
|
||
"AssistFun4": assist_fun4,
|
||
"AssistFun5": assist_fun5,
|
||
"HoleNumbers": hole_numbers,
|
||
"LiquidDispensingMethod": liquid_method,
|
||
}
|
||
|
||
def UnLoad(
|
||
self,
|
||
dosage: int,
|
||
plate_no: int,
|
||
is_whole_plate: bool,
|
||
hole_row: int,
|
||
hole_col: int,
|
||
blending_times: int,
|
||
balance_height: int,
|
||
plate_or_hole: str,
|
||
hole_numbers: str,
|
||
assist_fun1: str = "",
|
||
assist_fun2: str = "",
|
||
assist_fun3: str = "",
|
||
assist_fun4: str = "",
|
||
assist_fun5: str = "",
|
||
liquid_method: str = "NormalDispense",
|
||
) -> Dict[str, Any]:
|
||
return {
|
||
"StepAxis": self.axis,
|
||
"Function": "UnLoad",
|
||
"DosageNum": dosage,
|
||
"PlateNo": plate_no,
|
||
"IsWholePlate": is_whole_plate,
|
||
"HoleRow": hole_row,
|
||
"HoleCol": hole_col,
|
||
"BlendingTimes": blending_times,
|
||
"BalanceHeight": balance_height,
|
||
"PlateOrHoleNum": plate_or_hole,
|
||
"AssistFun1": assist_fun1,
|
||
"AssistFun2": assist_fun2,
|
||
"AssistFun3": assist_fun3,
|
||
"AssistFun4": assist_fun4,
|
||
"AssistFun5": assist_fun5,
|
||
"HoleNumbers": hole_numbers,
|
||
"LiquidDispensingMethod": liquid_method,
|
||
}
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# Example usage
|
||
# 1. 用导出的json，给每个T1 T2板子设定相应的物料，如果是孔板和枪头盒，要对应区分
|
||
# 2. backend需要支持num channel为1的情况
|
||
# 3. 设计一个单点动作流程，可以跑
|
||
# 4.
|
||
|
||
|
||
deck = PRCXI9300Deck(name="PRCXI_Deck_9300", size_x=100, size_y=100, size_z=100)
|
||
|
||
from pylabrobot.resources.opentrons.tip_racks import opentrons_96_tiprack_300ul,opentrons_96_tiprack_10ul
|
||
from pylabrobot.resources.opentrons.plates import corning_96_wellplate_360ul_flat, nest_96_wellplate_2ml_deep
|
||
|
||
def get_well_container(name: str) -> PRCXI9300Container:
|
||
well_containers = corning_96_wellplate_360ul_flat(name).serialize()
|
||
plate = PRCXI9300Container(name=name, size_x=50, size_y=50, size_z=10, category="plate",
|
||
ordering=well_containers["ordering"])
|
||
plate_serialized = plate.serialize()
|
||
plate_serialized["parent_name"] = deck.name
|
||
well_containers.update({k: v for k, v in plate_serialized.items() if k not in ["children"]})
|
||
new_plate: PRCXI9300Container = PRCXI9300Container.deserialize(well_containers)
|
||
return new_plate
|
||
|
||
def get_tip_rack(name: str) -> PRCXI9300Container:
|
||
tip_racks = opentrons_96_tiprack_300ul("name").serialize()
|
||
tip_rack = PRCXI9300Container(name=name, size_x=50, size_y=50, size_z=10, category="tip_rack",
|
||
ordering=tip_racks["ordering"])
|
||
tip_rack_serialized = tip_rack.serialize()
|
||
tip_rack_serialized["parent_name"] = deck.name
|
||
tip_racks.update({k: v for k, v in tip_rack_serialized.items() if k not in ["children"]})
|
||
new_tip_rack: PRCXI9300Container = PRCXI9300Container.deserialize(tip_racks)
|
||
return new_tip_rack
|
||
|
||
plate1 = get_tip_rack("RackT1")
|
||
plate1.load_state({
|
||
"Material": {
|
||
"uuid": "076250742950465b9d6ea29a225dfb00",
|
||
"Code": "ZX-001-300",
|
||
"Name": "300μL Tip头"
|
||
}
|
||
})
|
||
|
||
plate2 = get_well_container("PlateT2")
|
||
plate2.load_state({
|
||
"Material": {
|
||
"uuid": "57b1e4711e9e4a32b529f3132fc5931f",
|
||
"Code": "ZX-019-2.2",
|
||
"Name": "96深孔板"
|
||
}
|
||
})
|
||
|
||
|
||
plate3 = PRCXI9300Trash("trash", size_x=50, size_y=100, size_z=10, category="trash")
|
||
plate3.load_state({
|
||
"Material": {
|
||
"uuid": "730067cf07ae43849ddf4034299030e9"
|
||
}
|
||
})
|
||
|
||
plate4 = get_well_container("PlateT4")
|
||
plate4.load_state({
|
||
"Material": {
|
||
"uuid": "57b1e4711e9e4a32b529f3132fc5931f",
|
||
"Code": "ZX-019-2.2",
|
||
"Name": "96深孔板"
|
||
}
|
||
})
|
||
|
||
plate5 = get_well_container("PlateT5")
|
||
plate5.load_state({
|
||
"Material": {
|
||
"uuid": "57b1e4711e9e4a32b529f3132fc5931f",
|
||
"Code": "ZX-019-2.2",
|
||
"Name": "96深孔板"
|
||
}
|
||
})
|
||
plate6 = get_well_container("PlateT6")
|
||
|
||
plate6.load_state({
|
||
"Material": {
|
||
"uuid": "57b1e4711e9e4a32b529f3132fc5931f",
|
||
"Code": "ZX-019-2.2",
|
||
"Name": "96深孔板"
|
||
}
|
||
})
|
||
|
||
deck.assign_child_resource(plate1, location=Coordinate(0, 0, 0))
|
||
deck.assign_child_resource(plate2, location=Coordinate(0, 0, 0))
|
||
deck.assign_child_resource(plate3, location=Coordinate(0, 0, 0))
|
||
deck.assign_child_resource(plate4, location=Coordinate(0, 0, 0))
|
||
deck.assign_child_resource(plate5, location=Coordinate(0, 0, 0))
|
||
deck.assign_child_resource(plate6, location=Coordinate(0, 0, 0))
|
||
|
||
# plate_2_liquids = [[('water', 500)]]*96
|
||
|
||
# plate2.set_well_liquids(plate_2_liquids)
|
||
|
||
|
||
|
||
|
||
handler = PRCXI9300Handler(deck=deck, host="10.181.214.132", port=9999,
|
||
timeout=10.0, setup=False, debug=False,
|
||
simulator=True,
|
||
matrix_id="71593",
|
||
channel_num=8, axis="Left") # Initialize the handler with the deck and host settings
|
||
|
||
|
||
|
||
|
||
plate_2_liquids = handler.set_group("water", plate2.children[:8], [200]*8)
|
||
plate5_liquids = handler.set_group("master_mix", plate5.children[:8], [100]*8)
|
||
|
||
handler.set_tiprack([plate1])
|
||
asyncio.run(handler.setup()) # Initialize the handler and setup the connection
|
||
from pylabrobot.resources import set_volume_tracking
|
||
# from pylabrobot.resources import set_tip_tracking
|
||
set_volume_tracking(enabled=True)
|
||
# from unilabos.resources.graphio import *
|
||
# # A = tree_to_list([resource_plr_to_ulab(deck)])
|
||
# # with open("deck_9300_new.json", "w", encoding="utf-8") as f:
|
||
# # json.dump(A, f, indent=4, ensure_ascii=False)
|
||
asyncio.run(handler.create_protocol(protocol_name="Test Protocol")) # Initialize the backend and setup the connection
|
||
asyncio.run(handler.transfer_group("water", "master_mix", 100)) # Reset tip tracking
|
||
|
||
# asyncio.run(handler.pick_up_tips(plate1.children[:8],[0,1,2,3,4,5,6,7]))
|
||
# print(plate1.children[:8])
|
||
# asyncio.run(handler.aspirate(plate2.children[:8],[50]*8, [0,1,2,3,4,5,6,7]))
|
||
# print(plate2.children[:8])
|
||
# asyncio.run(handler.dispense(plate5.children[:8],[50]*8,[0,1,2,3,4,5,6,7]))
|
||
# print(plate5.children[:8])
|
||
|
||
# #asyncio.run(handler.drop_tips(tip_rack.children[8:16],[0,1,2,3,4,5,6,7]))
|
||
# asyncio.run(handler.discard_tips([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))
|
||
# #print(json.dumps(handler._unilabos_backend.steps_todo_list, indent=2)) # Print matrix info
|
||
# asyncio.run(handler.add_liquid(
|
||
# asp_vols=[100]*16,
|
||
# dis_vols=[100]*16,
|
||
# reagent_sources=plate2.children[:16],
|
||
# targets=plate5.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.run_protocol()) # Run the protocol
|
||
# asyncio.run(handler.remove_liquid(
|
||
# vols=[100]*16,
|
||
# sources=plate2.children[-16:],
|
||
# waste_liquid=plate5.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] * 32,
|
||
# blow_out_air_volume=[None] * 32,
|
||
# spread="wide",
|
||
# ))
|
||
|
||
# acid = [20]*8+[40]*8+[60]*8+[80]*8+[100]*8+[120]*8+[140]*8+[160]*8+[180]*8+[200]*8+[220]*8+[240]*8
|
||
# alkaline = acid[::-1] # Reverse the acid list for alkaline
|
||
# asyncio.run(handler.transfer_liquid(
|
||
# asp_vols=acid,
|
||
# dis_vols=acid,
|
||
# tip_racks=[plate1],
|
||
# sources=plate2.children[:],
|
||
# targets=plate5.children[:],
|
||
# use_channels=[0, 1, 2, 3, 4, 5, 6, 7],
|
||
# offsets=[Coordinate(0, 0, 0)] * 32,
|
||
# asp_flow_rates=[None] * 16,
|
||
# dis_flow_rates=[None] * 16,
|
||
# liquid_height=[None] * 32,
|
||
# blow_out_air_volume=[None] * 32,
|
||
# mix_times=3,
|
||
# mix_vol=50,
|
||
# spread="wide",
|
||
# ))
|
||
# asyncio.run(handler.run_protocol()) # Run the protocol
|
||
# # input("Running protocol...")
|
||
# # input("Press Enter to continue...") # Wait for user input before proceeding
|
||
# # print("PRCXI9300Handler initialized with deck and host settings.")
|
||
|
||
|
||
|
||
### 9320 ###
|
||
|
||
|
||
# deck = PRCXI9300Deck(name="PRCXI_Deck", size_x=100, size_y=100, size_z=100)
|
||
|
||
# from pylabrobot.resources.opentrons.tip_racks import tipone_96_tiprack_200ul,opentrons_96_tiprack_10ul
|
||
# from pylabrobot.resources.opentrons.plates import corning_96_wellplate_360ul_flat, nest_96_wellplate_2ml_deep
|
||
|
||
# def get_well_container(name: str) -> PRCXI9300Container:
|
||
# well_containers = corning_96_wellplate_360ul_flat(name).serialize()
|
||
# plate = PRCXI9300Container(name=name, size_x=50, size_y=50, size_z=10, category="plate",
|
||
# ordering=collections.OrderedDict())
|
||
# plate_serialized = plate.serialize()
|
||
# plate_serialized["parent_name"] = deck.name
|
||
# well_containers.update({k: v for k, v in plate_serialized.items() if k not in ["children"]})
|
||
# new_plate: PRCXI9300Container = PRCXI9300Container.deserialize(well_containers)
|
||
# return new_plate
|
||
|
||
# def get_tip_rack(name: str) -> PRCXI9300Container:
|
||
# tip_racks = opentrons_96_tiprack_10ul("name").serialize()
|
||
# tip_rack = PRCXI9300Container(name=name, size_x=50, size_y=50, size_z=10, category="tip_rack",
|
||
# ordering=collections.OrderedDict())
|
||
# tip_rack_serialized = tip_rack.serialize()
|
||
# tip_rack_serialized["parent_name"] = deck.name
|
||
# tip_racks.update({k: v for k, v in tip_rack_serialized.items() if k not in ["children"]})
|
||
# new_tip_rack: PRCXI9300Container = PRCXI9300Container.deserialize(tip_racks)
|
||
# return new_tip_rack
|
||
|
||
# plate1 = get_well_container("HPLCPlateT1")
|
||
# plate1.load_state({
|
||
# "Material": {
|
||
# "uuid": "548bbc3df0d4447586f2c19d2c0c0c55",
|
||
# "Code": "HPLC01",
|
||
# "Name": "HPLC料盘"
|
||
# }
|
||
# })
|
||
# plate2 = get_well_container("PlateT2")
|
||
# plate2.load_state({
|
||
# "Material": {
|
||
# "uuid": "04211a2dc93547fe9bf6121eac533650",
|
||
# }
|
||
# })
|
||
# plate3 = get_well_container("PlateT3")
|
||
# plate3.load_state({
|
||
# "Material": {
|
||
# "uuid": "04211a2dc93547fe9bf6121eac533650",
|
||
# }
|
||
# })
|
||
# trash = PRCXI9300Trash(name="trash", size_x=50, size_y=50, size_z=10, category="trash")
|
||
# trash.load_state({
|
||
# "Material": {
|
||
# "uuid": "730067cf07ae43849ddf4034299030e9"
|
||
# }
|
||
# })
|
||
# plate5 = get_well_container("PlateT5")
|
||
# plate5.load_state({
|
||
# "Material": {
|
||
# "uuid": "04211a2dc93547fe9bf6121eac533650",
|
||
# }
|
||
# })
|
||
# plate6 = get_well_container("PlateT6")
|
||
# plate6.load_state({
|
||
# "Material": {
|
||
# "uuid": "04211a2dc93547fe9bf6121eac533650"
|
||
# }
|
||
# })
|
||
# plate7 = PRCXI9300Container(name="plateT7", size_x=50, size_y=50, size_z=10, category="plate", ordering=collections.OrderedDict())
|
||
# plate7.load_state({
|
||
# "Material": {
|
||
# "uuid": "04211a2dc93547fe9bf6121eac533650"
|
||
# }
|
||
# })
|
||
# plate8 = get_tip_rack("RackT8")
|
||
# plate8.load_state({
|
||
# "Material": {
|
||
# "uuid": "068b3815e36b4a72a59bae017011b29f",
|
||
# "Code": "ZX-001-10+",
|
||
# "Name": "10μL加长 Tip头"
|
||
# }
|
||
# })
|
||
# plate9 = get_well_container("PlateT9")
|
||
# plate9.load_state({
|
||
# "Material": {
|
||
# "uuid": "04211a2dc93547fe9bf6121eac533650"
|
||
# }
|
||
# })
|
||
# plate10 = get_well_container("PlateT10")
|
||
# plate10.load_state({
|
||
# "Material": {
|
||
# "uuid": "04211a2dc93547fe9bf6121eac533650"
|
||
# }
|
||
# })
|
||
# plate11 = get_well_container("PlateT11")
|
||
# plate11.load_state({
|
||
# "Material": {
|
||
# "uuid": "57b1e4711e9e4a32b529f3132fc5931f",
|
||
# }
|
||
# })
|
||
# plate12 = get_well_container("PlateT12")
|
||
# plate12.load_state({
|
||
# "Material": {
|
||
# "uuid": "04211a2dc93547fe9bf6121eac533650"
|
||
# }
|
||
# })
|
||
# plate13 = get_well_container("PlateT13")
|
||
# plate13.load_state({
|
||
# "Material": {
|
||
# "uuid": "04211a2dc93547fe9bf6121eac533650"
|
||
# }
|
||
# })
|
||
|
||
# # container_for_nothing = PRCXI9300Container(name="container_for_nothing", size_x=50, size_y=50, size_z=10, category="plate", ordering=collections.OrderedDict())
|
||
|
||
# deck.assign_child_resource(plate1, location=Coordinate(0, 0, 0))
|
||
# deck.assign_child_resource(PRCXI9300Container(name="container_for_nothing1", size_x=50, size_y=50, size_z=10, category="plate", ordering=collections.OrderedDict()), location=Coordinate(0, 0, 0))
|
||
# deck.assign_child_resource(PRCXI9300Container(name="container_for_nothing2", size_x=50, size_y=50, size_z=10, category="plate", ordering=collections.OrderedDict()), location=Coordinate(0, 0, 0))
|
||
# deck.assign_child_resource(trash, location=Coordinate(0, 0, 0))
|
||
# deck.assign_child_resource(PRCXI9300Container(name="container_for_nothing3", size_x=50, size_y=50, size_z=10, category="plate", ordering=collections.OrderedDict()), location=Coordinate(0, 0, 0))
|
||
# deck.assign_child_resource(PRCXI9300Container(name="container_for_nothing", size_x=50, size_y=50, size_z=10, category="plate", ordering=collections.OrderedDict()), location=Coordinate(0, 0, 0))
|
||
# deck.assign_child_resource(PRCXI9300Container(name="container_for_nothing4", size_x=50, size_y=50, size_z=10, category="plate", ordering=collections.OrderedDict()), location=Coordinate(0, 0, 0))
|
||
# deck.assign_child_resource(plate8, location=Coordinate(0, 0, 0))
|
||
# deck.assign_child_resource(PRCXI9300Container(name="container_for_nothing5", size_x=50, size_y=50, size_z=10, category="plate", ordering=collections.OrderedDict()), location=Coordinate(0, 0, 0))
|
||
# deck.assign_child_resource(PRCXI9300Container(name="container_for_nothing6", size_x=50, size_y=50, size_z=10, category="plate", ordering=collections.OrderedDict()), location=Coordinate(0, 0, 0))
|
||
# deck.assign_child_resource(plate11, location=Coordinate(0, 0, 0))
|
||
# deck.assign_child_resource(PRCXI9300Container(name="container_for_nothing7", size_x=50, size_y=50, size_z=10, category="plate", ordering=collections.OrderedDict()), location=Coordinate(0, 0, 0))
|
||
# deck.assign_child_resource(PRCXI9300Container(name="container_for_nothing8", size_x=50, size_y=50, size_z=10, category="plate", ordering=collections.OrderedDict()), location=Coordinate(0, 0, 0))
|
||
|
||
# handler = PRCXI9300Handler(deck=deck, host="10.181.102.13", port=9999,
|
||
# timeout=10.0, setup=False, debug=False,
|
||
# matrix_id="fd383e6d-2d0e-40b5-9c01-1b2870b1f1b1",
|
||
# channel_num=8, axis="Right") # Initialize the handler with the deck and host settings
|
||
|
||
# handler.set_tiprack([plate8]) # Set the tip rack for the handler
|
||
# asyncio.run(handler.setup()) # Initialize the handler and setup the connection
|
||
# from pylabrobot.resources import set_volume_tracking
|
||
# # from pylabrobot.resources import set_tip_tracking
|
||
# set_volume_tracking(enabled=True)
|
||
|
||
# plate11.set_well_liquids([("Water", 100) if (i % 8 == 0 and i // 8 < 6) else (None, 100) for i in range(96)]) # Set liquids for every 8 wells in plate8
|
||
|
||
# from unilabos.resources.graphio import *
|
||
|
||
# A = tree_to_list([resource_plr_to_ulab(deck)])
|
||
# # with open("deck.json", "w", encoding="utf-8") as f:
|
||
# # json.dump(A, f, indent=4, ensure_ascii=False)
|
||
|
||
# print(plate11.get_well(0).tracker.get_used_volume())
|
||
# asyncio.run(handler.create_protocol(protocol_name="Test Protocol")) # Initialize the backend and setup the connection
|
||
|
||
# # asyncio.run(handler.pick_up_tips([plate8.children[8]],[0]))
|
||
# # print(plate8.children[8])
|
||
# # # asyncio.run(handler.run_protocol())
|
||
# # asyncio.run(handler.aspirate([plate11.children[0]],[10], [0]))
|
||
# # print(plate11.children[0])
|
||
# # # asyncio.run(handler.run_protocol())
|
||
# # asyncio.run(handler.dispense([plate1.children[0]],[10],[0]))
|
||
# # print(plate1.children[0])
|
||
# # # asyncio.run(handler.run_protocol())
|
||
# # asyncio.run(handler.mix([plate1.children[0]], mix_time=3, mix_vol=5, height_to_bottom=0.5, offsets=Coordinate(0, 0, 0), mix_rate=100))
|
||
# # print(plate1.children[0])
|
||
# # asyncio.run(handler.discard_tips())
|
||
|
||
# asyncio.run(handler.add_liquid(
|
||
# asp_vols=[10]*7,
|
||
# dis_vols=[10]*7,
|
||
# reagent_sources=plate11.children[:7],
|
||
# targets=plate1.children[2:9],
|
||
# use_channels=[0],
|
||
# flow_rates=[None] * 7,
|
||
# offsets=[Coordinate(0, 0, 0)] * 7,
|
||
# liquid_height=[None] * 7,
|
||
# blow_out_air_volume=[None] * 2,
|
||
# delays=None,
|
||
# mix_time=3,
|
||
# mix_vol=5,
|
||
# spread="custom",
|
||
# ))
|
||
|
||
# asyncio.run(handler.run_protocol()) # Run the protocol
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
# # # asyncio.run(handler.transfer_liquid(
|
||
# # # asp_vols=[10]*2,
|
||
# # # dis_vols=[10]*2,
|
||
# # # sources=plate11.children[:2],
|
||
# # # targets=plate11.children[-2:],
|
||
# # # use_channels=[0],
|
||
# # # offsets=[Coordinate(0, 0, 0)] * 4,
|
||
# # # liquid_height=[None] * 2,
|
||
# # # blow_out_air_volume=[None] * 2,
|
||
# # # delays=None,
|
||
# # # mix_times=3,
|
||
# # # mix_vol=5,
|
||
# # # spread="wide",
|
||
# # # tip_racks=[plate8]
|
||
# # # ))
|
||
|
||
# # # asyncio.run(handler.remove_liquid(
|
||
# # # vols=[10]*2,
|
||
# # # sources=plate11.children[:2],
|
||
# # # waste_liquid=plate11.children[43],
|
||
# # # use_channels=[0],
|
||
# # # offsets=[Coordinate(0, 0, 0)] * 4,
|
||
# # # liquid_height=[None] * 2,
|
||
# # # blow_out_air_volume=[None] * 2,
|
||
# # # delays=None,
|
||
# # # spread="wide"
|
||
# # # ))
|
||
# # asyncio.run(handler.run_protocol())
|
||
|
||
# # # 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))
|
||
# # #print(json.dumps(handler._unilabos_backend.steps_todo_list, indent=2)) # Print matrix info
|
||
|
||
|
||
# # # 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],
|
||
# # # flow_rates=[None] * 32,
|
||
# # # offsets=[Coordinate(0, 0, 0)] * 32,
|
||
# # # liquid_height=[None] * 32,
|
||
# # # blow_out_air_volume=[None] * 32,
|
||
# # # spread="wide",
|
||
# # # ))
|
||
# # # asyncio.run(handler.transfer_liquid(
|
||
# # # asp_vols=[100]*16,
|
||
# # # dis_vols=[100]*16,
|
||
# # # tip_racks=[tip_rack],
|
||
# # # sources=well_containers.children[-16:],
|
||
# # # targets=well_containers.children[:16],
|
||
# # # use_channels=[0, 1, 2, 3, 4, 5, 6, 7],
|
||
# # # offsets=[Coordinate(0, 0, 0)] * 32,
|
||
# # # asp_flow_rates=[None] * 16,
|
||
# # # dis_flow_rates=[None] * 16,
|
||
# # # liquid_height=[None] * 32,
|
||
# # # blow_out_air_volume=[None] * 32,
|
||
# # # mix_times=3,
|
||
# # # mix_vol=50,
|
||
# # # spread="wide",
|
||
# # # ))
|
||
# # print(json.dumps(handler._unilabos_backend.steps_todo_list, indent=2)) # Print matrix info
|
||
# # # input("pick_up_tips add step")
|
||
# # #asyncio.run(handler.run_protocol()) # Run the protocol
|
||
# # # input("Running protocol...")
|
||
# # # input("Press Enter to continue...") # Wait for user input before proceeding
|
||
# # # print("PRCXI9300Handler initialized with deck and host settings.")
|
||
|