PRCXI9320 json

This commit is contained in:
Xuwznln
2025-09-19 16:29:18 +08:00
parent 762c3c737c
commit 768f43880e
2 changed files with 36170 additions and 8483 deletions

File diff suppressed because it is too large Load Diff

View File

@@ -14,7 +14,9 @@ from pylabrobot.liquid_handling import (
SingleChannelDispense, SingleChannelDispense,
PickupTipRack, PickupTipRack,
DropTipRack, DropTipRack,
MultiHeadAspirationPlate, ChatterBoxBackend, LiquidHandlerChatterboxBackend, MultiHeadAspirationPlate,
ChatterBoxBackend,
LiquidHandlerChatterboxBackend,
) )
from pylabrobot.liquid_handling.standard import ( from pylabrobot.liquid_handling.standard import (
MultiHeadAspirationContainer, MultiHeadAspirationContainer,
@@ -72,7 +74,16 @@ class PRCXI9300Container(Plate, TipRack):
该类定义了 PRCXI 9300 的工作台布局和槽位信息。 该类定义了 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,): 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) super().__init__(name, size_x, size_y, size_z, category=category, ordering=ordering, model=model)
self._unilabos_state = {} self._unilabos_state = {}
@@ -102,7 +113,7 @@ class PRCXI9300Trash(Trash):
def load_state(self, state: Dict[str, Any]) -> None: def load_state(self, state: Dict[str, Any]) -> None:
"""从给定的状态加载工作台信息。""" """从给定的状态加载工作台信息。"""
#super().load_state(state) # super().load_state(state)
self._unilabos_state = state self._unilabos_state = state
def serialize_state(self) -> Dict[str, Dict[str, Any]]: def serialize_state(self) -> Dict[str, Dict[str, Any]]:
@@ -110,6 +121,7 @@ class PRCXI9300Trash(Trash):
data.update(self._unilabos_state) data.update(self._unilabos_state)
return data return data
class PRCXI9300Handler(LiquidHandlerAbstract): class PRCXI9300Handler(LiquidHandlerAbstract):
support_touch_tip = False support_touch_tip = False
@@ -120,7 +132,20 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
return True return True
return self._unilabos_backend.is_reset_ok 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=""): def __init__(
self,
deck: Deck,
host: str,
port: int,
timeout: float,
channel_num=8,
axis="Left",
setup=True,
debug=False,
simulator=False,
matrix_id="",
is_9320=False,
):
tablets_info = [] tablets_info = []
count = 0 count = 0
for child in deck.children: for child in deck.children:
@@ -129,7 +154,11 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
tablets_info.append( tablets_info.append(
WorkTablets(Number=count, Code=f"T{count}", Material=child._unilabos_state["Material"]) 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) if is_9320:
print("当前设备是9320")
self._unilabos_backend = PRCXI9300Backend(
tablets_info, host, port, timeout, channel_num, axis, setup, debug, matrix_id, is_9320
)
super().__init__(backend=self._unilabos_backend, deck=deck, simulator=simulator, channel_num=channel_num) 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]): def set_liquid(self, wells: list[Well], liquid_names: list[str], volumes: list[float]):
@@ -295,41 +324,88 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
mix_rate: Optional[float] = None, mix_rate: Optional[float] = None,
none_keys: List[str] = [], none_keys: List[str] = [],
): ):
return await self._unilabos_backend.mix(targets, mix_time, mix_vol, height_to_bottom, offsets, mix_rate, none_keys) 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]: def iter_tips(self, tip_racks: Sequence[TipRack]) -> Iterator[Resource]:
return super().iter_tips(tip_racks) return super().iter_tips(tip_racks)
async def pick_up_tips(self, tip_spots: List[TipSpot], use_channels: Optional[List[int]] = None, async def pick_up_tips(
offsets: Optional[List[Coordinate]] = None, **backend_kwargs): 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) return await super().pick_up_tips(tip_spots, use_channels, offsets, **backend_kwargs)
async def aspirate(self, resources: Sequence[Container], vols: List[float], async def aspirate(
use_channels: Optional[List[int]] = None, flow_rates: Optional[List[Optional[float]]] = None, self,
offsets: Optional[List[Coordinate]] = None, resources: Sequence[Container],
liquid_height: Optional[List[Optional[float]]] = None, vols: List[float],
blow_out_air_volume: Optional[List[Optional[float]]] = None, use_channels: Optional[List[int]] = None,
spread: Literal["wide", "tight", "custom"] = "wide", **backend_kwargs): 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, return await super().aspirate(
blow_out_air_volume, spread, **backend_kwargs) 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, async def drop_tips(
offsets: Optional[List[Coordinate]] = None, allow_nonzero_volume: bool = False, self,
**backend_kwargs): 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) 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], async def dispense(
use_channels: Optional[List[int]] = None, flow_rates: Optional[List[Optional[float]]] = None, self,
offsets: Optional[List[Coordinate]] = None, resources: Sequence[Container],
liquid_height: Optional[List[Optional[float]]] = None, vols: List[float],
blow_out_air_volume: Optional[List[Optional[float]]] = None, use_channels: Optional[List[int]] = None,
spread: Literal["wide", "tight", "custom"] = "wide", **backend_kwargs): flow_rates: Optional[List[Optional[float]]] = None,
return await super().dispense(resources, vols, use_channels, flow_rates, offsets, liquid_height, offsets: Optional[List[Coordinate]] = None,
blow_out_air_volume, spread, **backend_kwargs) 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, async def discard_tips(
offsets: Optional[List[Coordinate]] = None, **backend_kwargs): 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) return await super().discard_tips(use_channels, allow_nonzero_volume, offsets, **backend_kwargs)
def set_tiprack(self, tip_racks: Sequence[TipRack]): def set_tiprack(self, tip_racks: Sequence[TipRack]):
@@ -363,16 +439,17 @@ class PRCXI9300Backend(LiquidHandlerBackend):
host: str = "127.0.0.1", host: str = "127.0.0.1",
port: int = 9999, port: int = 9999,
timeout: float = 10.0, timeout: float = 10.0,
channel_num: int=8, channel_num: int = 8,
axis: str="Left", axis: str = "Left",
setup=True, setup=True,
debug=False, debug=False,
matrix_id="", matrix_id="",
is_9320=False,
) -> None: ) -> None:
super().__init__() super().__init__()
self.tablets_info = tablets_info self.tablets_info = tablets_info
self.matrix_id = matrix_id self.matrix_id = matrix_id
self.api_client = PRCXI9300Api(host, port, timeout, axis, debug) self.api_client = PRCXI9300Api(host, port, timeout, axis, debug, is_9320)
self.host, self.port, self.timeout = host, port, timeout self.host, self.port, self.timeout = host, port, timeout
self._num_channels = channel_num self._num_channels = channel_num
self._execute_setup = setup self._execute_setup = setup
@@ -391,7 +468,7 @@ class PRCXI9300Backend(LiquidHandlerBackend):
MatrixCount=len(self.tablets_info), MatrixCount=len(self.tablets_info),
WorkTablets=self.tablets_info, WorkTablets=self.tablets_info,
) )
#print(json.dumps(self.matrix_info, indent=2)) # print(json.dumps(self.matrix_info, indent=2))
if not len(self.matrix_id): if not len(self.matrix_id):
res = self.api_client.add_WorkTablet_Matrix(self.matrix_info) res = self.api_client.add_WorkTablet_Matrix(self.matrix_info)
assert res["Success"], f"Failed to create matrix: {res.get('Message', 'Unknown error')}" assert res["Success"], f"Failed to create matrix: {res.get('Message', 'Unknown error')}"
@@ -401,9 +478,7 @@ class PRCXI9300Backend(LiquidHandlerBackend):
) )
else: else:
print(f"PRCXI9300Backend using predefined worktable {self.matrix_id}, skipping matrix creation.") print(f"PRCXI9300Backend using predefined worktable {self.matrix_id}, skipping matrix creation.")
solution_id = self.api_client.add_solution( solution_id = self.api_client.add_solution(f"protocol_{run_time}", self.matrix_id, self.steps_todo_list)
f"protocol_{run_time}", self.matrix_id, self.steps_todo_list
)
print(f"PRCXI9300Backend created solution with ID: {solution_id}") print(f"PRCXI9300Backend created solution with ID: {solution_id}")
self.api_client.load_solution(solution_id) self.api_client.load_solution(solution_id)
print(json.dumps(self.steps_todo_list, indent=2)) print(json.dumps(self.steps_todo_list, indent=2))
@@ -464,9 +539,17 @@ class PRCXI9300Backend(LiquidHandlerBackend):
if self._num_channels == 1: if self._num_channels == 1:
hole_row = tipspot_index % 8 + 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, step = self.api_client.Load(
blending_times=0, balance_height=0, plate_or_hole=f"H{hole_col}-8,T{PlateNo}", dosage=0,
hole_numbers="1,2,3,4,5,6,7,8") 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) self.steps_todo_list.append(step)
async def drop_tips(self, ops: List[Drop], use_channels: List[int] = None): async def drop_tips(self, ops: List[Drop], use_channels: List[int] = None):
@@ -487,11 +570,10 @@ class PRCXI9300Backend(LiquidHandlerBackend):
balance_height=0, balance_height=0,
plate_or_hole=f"H{1}-8,T{PlateNo}", plate_or_hole=f"H{1}-8,T{PlateNo}",
hole_numbers="1,2,3,4,5,6,7,8", hole_numbers="1,2,3,4,5,6,7,8",
) )
self.steps_todo_list.append(step) self.steps_todo_list.append(step)
return return
#print(ops[0].resource.parent.children.index(ops[0].resource)) # print(ops[0].resource.parent.children.index(ops[0].resource))
plate_indexes = [] plate_indexes = []
for op in ops: for op in ops:
@@ -500,7 +582,9 @@ class PRCXI9300Backend(LiquidHandlerBackend):
plate_index = deck.children.index(plate) plate_index = deck.children.index(plate)
plate_indexes.append(plate_index) plate_indexes.append(plate_index)
if len(set(plate_indexes)) != 1: if len(set(plate_indexes)) != 1:
raise ValueError("All drop_tips must be from the same plate. Found different plates: " + str(plate_indexes)) raise ValueError(
"All drop_tips must be from the same plate. Found different plates: " + str(plate_indexes)
)
tip_columns = [] tip_columns = []
for op in ops: for op in ops:
@@ -508,7 +592,9 @@ class PRCXI9300Backend(LiquidHandlerBackend):
tipspot_index = tipspot.parent.children.index(tipspot) tipspot_index = tipspot.parent.children.index(tipspot)
tip_columns.append(tipspot_index // 8) tip_columns.append(tipspot_index // 8)
if len(set(tip_columns)) != 1: if len(set(tip_columns)) != 1:
raise ValueError("All drop_tips must be from the same tip column. Found different columns: " + str(tip_columns)) raise ValueError(
"All drop_tips must be from the same tip column. Found different columns: " + str(tip_columns)
)
PlateNo = plate_indexes[0] + 1 PlateNo = plate_indexes[0] + 1
hole_col = tip_columns[0] + 1 hole_col = tip_columns[0] + 1
@@ -530,16 +616,15 @@ class PRCXI9300Backend(LiquidHandlerBackend):
self.steps_todo_list.append(step) self.steps_todo_list.append(step)
async def mix( async def mix(
self, self,
targets: Sequence[Container], targets: Sequence[Container],
mix_time: int = None, mix_time: int = None,
mix_vol: Optional[int] = None, mix_vol: Optional[int] = None,
height_to_bottom: Optional[float] = None, height_to_bottom: Optional[float] = None,
offsets: Optional[Coordinate] = None, offsets: Optional[Coordinate] = None,
mix_rate: Optional[float] = None, mix_rate: Optional[float] = None,
none_keys: List[str] = [], none_keys: List[str] = [],
): ):
"""Mix liquid in the specified resources.""" """Mix liquid in the specified resources."""
plate_indexes = [] plate_indexes = []
@@ -558,7 +643,9 @@ class PRCXI9300Backend(LiquidHandlerBackend):
tip_columns.append(tipspot_index // 8) tip_columns.append(tipspot_index // 8)
if len(set(tip_columns)) != 1: if len(set(tip_columns)) != 1:
raise ValueError("All pickups must be from the same tip column. Found different columns: " + str(tip_columns)) raise ValueError(
"All pickups must be from the same tip column. Found different columns: " + str(tip_columns)
)
PlateNo = plate_indexes[0] + 1 PlateNo = plate_indexes[0] + 1
hole_col = tip_columns[0] + 1 hole_col = tip_columns[0] + 1
@@ -581,7 +668,6 @@ class PRCXI9300Backend(LiquidHandlerBackend):
self.steps_todo_list.append(step) self.steps_todo_list.append(step)
async def aspirate(self, ops: List[SingleChannelAspiration], use_channels: List[int] = None): async def aspirate(self, ops: List[SingleChannelAspiration], use_channels: List[int] = None):
"""Aspirate liquid from the specified resources.""" """Aspirate liquid from the specified resources."""
plate_indexes = [] plate_indexes = []
@@ -601,7 +687,9 @@ class PRCXI9300Backend(LiquidHandlerBackend):
tip_columns.append(tipspot_index // 8) tip_columns.append(tipspot_index // 8)
if len(set(tip_columns)) != 1: if len(set(tip_columns)) != 1:
raise ValueError("All pickups must be from the same tip column. Found different columns: " + str(tip_columns)) raise ValueError(
"All pickups must be from the same tip column. Found different columns: " + str(tip_columns)
)
volumes = [op.volume for op in ops] volumes = [op.volume for op in ops]
if len(set(volumes)) != 1: if len(set(volumes)) != 1:
@@ -613,15 +701,20 @@ class PRCXI9300Backend(LiquidHandlerBackend):
if self.num_channels == 1: if self.num_channels == 1:
hole_row = tipspot_index % 8 + 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, step = self.api_client.Imbibing(
hole_col=hole_col, blending_times=0, balance_height=0, dosage=int(volumes[0]),
plate_or_hole=f"H{hole_col}-8,T{PlateNo}", hole_numbers="1,2,3,4,5,6,7,8") 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) self.steps_todo_list.append(step)
async def dispense(self, ops: List[SingleChannelDispense], use_channels: List[int] = None): async def dispense(self, ops: List[SingleChannelDispense], use_channels: List[int] = None):
"""Dispense liquid into the specified resources.""" """Dispense liquid into the specified resources."""
plate_indexes = [] plate_indexes = []
@@ -641,7 +734,9 @@ class PRCXI9300Backend(LiquidHandlerBackend):
tip_columns.append(tipspot_index // 8) tip_columns.append(tipspot_index // 8)
if len(set(tip_columns)) != 1: if len(set(tip_columns)) != 1:
raise ValueError("All dispense must be from the same tip column. Found different columns: " + str(tip_columns)) raise ValueError(
"All dispense must be from the same tip column. Found different columns: " + str(tip_columns)
)
volumes = [op.volume for op in ops] volumes = [op.volume for op in ops]
if len(set(volumes)) != 1: if len(set(volumes)) != 1:
@@ -700,10 +795,19 @@ class PRCXI9300Backend(LiquidHandlerBackend):
class PRCXI9300Api: class PRCXI9300Api:
def __init__(self, host: str = "127.0.0.1", port: int = 9999, timeout: float = 10.0, axis="Left", debug: bool = False) -> None: def __init__(
self,
host: str = "127.0.0.1",
port: int = 9999,
timeout: float = 10.0,
axis="Left",
debug: bool = False,
is_9320: bool = False,
) -> None:
self.host, self.port, self.timeout = host, port, timeout self.host, self.port, self.timeout = host, port, timeout
self.debug = debug self.debug = debug
self.axis = axis self.axis = axis
self.is_9320 = is_9320
@staticmethod @staticmethod
def _len_prefix(n: int) -> bytes: def _len_prefix(n: int) -> bytes:
@@ -799,6 +903,10 @@ class PRCXI9300Api:
return self.call("IMachineState", "GetLocation", [axis_num]) return self.call("IMachineState", "GetLocation", [axis_num])
# ---------------------------------------------------- 版位矩阵IMatrix # ---------------------------------------------------- 版位矩阵IMatrix
def get_all_materials(self) -> Dict[str, Any]:
"""GetStepState"""
return self.call("IMatrix", "GetAllMaterial", [])
def list_matrices(self) -> List[Dict[str, Any]]: def list_matrices(self) -> List[Dict[str, Any]]:
"""GetWorkTabletMatrices""" """GetWorkTabletMatrices"""
return self.call("IMatrix", "GetWorkTabletMatrices") return self.call("IMatrix", "GetWorkTabletMatrices")
@@ -808,12 +916,26 @@ class PRCXI9300Api:
return self.call("IMatrix", "GetWorkTabletMatrixById", [matrix_id]) return self.call("IMatrix", "GetWorkTabletMatrixById", [matrix_id])
def add_WorkTablet_Matrix(self, matrix: MatrixInfo): def add_WorkTablet_Matrix(self, matrix: MatrixInfo):
return self.call("IMatrix", "AddWorkTabletMatrix2", [matrix]) return self.call("IMatrix", "AddWorkTabletMatrix2" if self.is_9320 else "AddWorkTabletMatrix", [matrix])
def Load(self, dosage: int, plate_no: int, is_whole_plate: bool, hole_row: int, hole_col: int, blending_times: int, def Load(
balance_height: int, plate_or_hole: str, hole_numbers: str, assist_fun1: str = "", assist_fun2: str = "", self,
assist_fun3: str = "", assist_fun4: str = "", assist_fun5: str = "", dosage: int,
liquid_method: str = "NormalDispense") -> Dict[str, Any]: 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 { return {
"StepAxis": self.axis, "StepAxis": self.axis,
"Function": "Load", "Function": "Load",
@@ -834,10 +956,24 @@ class PRCXI9300Api:
"LiquidDispensingMethod": liquid_method, "LiquidDispensingMethod": liquid_method,
} }
def Imbibing(self, dosage: int, plate_no: int, is_whole_plate: bool, hole_row: int, hole_col: int, def Imbibing(
blending_times: int, balance_height: int, plate_or_hole: str, hole_numbers: str, assist_fun1: str = "", self,
assist_fun2: str = "", assist_fun3: str = "", assist_fun4: str = "", assist_fun5: str = "", dosage: int,
liquid_method: str = "NormalDispense") -> Dict[str, Any]: 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 { return {
"StepAxis": self.axis, "StepAxis": self.axis,
"Function": "Imbibing", "Function": "Imbibing",
@@ -1077,7 +1213,6 @@ if __name__ == "__main__":
# 3. 设计一个单点动作流程,可以跑 # 3. 设计一个单点动作流程,可以跑
# 4. # 4.
# deck = PRCXI9300Deck(name="PRCXI_Deck_9300", size_x=100, size_y=100, size_z=100) # 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.tip_racks import opentrons_96_tiprack_300ul,opentrons_96_tiprack_10ul
@@ -1121,7 +1256,6 @@ if __name__ == "__main__":
# } # }
# }) # })
# plate3 = PRCXI9300Trash("trash", size_x=50, size_y=100, size_z=10, category="trash") # plate3 = PRCXI9300Trash("trash", size_x=50, size_y=100, size_z=10, category="trash")
# plate3.load_state({ # plate3.load_state({
# "Material": { # "Material": {
@@ -1263,13 +1397,14 @@ if __name__ == "__main__":
deck = PRCXI9300Deck(name="PRCXI_Deck", size_x=100, size_y=100, size_z=100) 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.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 from pylabrobot.resources.opentrons.plates import corning_96_wellplate_360ul_flat, nest_96_wellplate_2ml_deep
def get_well_container(name: str) -> PRCXI9300Container: def get_well_container(name: str) -> PRCXI9300Container:
well_containers = corning_96_wellplate_360ul_flat(name).serialize() well_containers = corning_96_wellplate_360ul_flat(name).serialize()
plate = PRCXI9300Container(name=name, size_x=50, size_y=50, size_z=10, category="plate", plate = PRCXI9300Container(
ordering=collections.OrderedDict()) name=name, size_x=50, size_y=50, size_z=10, category="plate", ordering=well_containers["ordering"]
)
plate_serialized = plate.serialize() plate_serialized = plate.serialize()
plate_serialized["parent_name"] = deck.name plate_serialized["parent_name"] = deck.name
well_containers.update({k: v for k, v in plate_serialized.items() if k not in ["children"]}) well_containers.update({k: v for k, v in plate_serialized.items() if k not in ["children"]})
@@ -1278,121 +1413,223 @@ if __name__ == "__main__":
def get_tip_rack(name: str) -> PRCXI9300Container: def get_tip_rack(name: str) -> PRCXI9300Container:
tip_racks = opentrons_96_tiprack_10ul("name").serialize() 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", tip_rack = PRCXI9300Container(
ordering=collections.OrderedDict()) 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 = tip_rack.serialize()
tip_rack_serialized["parent_name"] = deck.name 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"]}) 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) new_tip_rack: PRCXI9300Container = PRCXI9300Container.deserialize(tip_racks)
return new_tip_rack return new_tip_rack
plate1 = get_well_container("HPLCPlateT1") plate1 = get_well_container("PlateT1")
plate1.load_state({ plate1.load_state(
"Material": { {"Material": {"uuid": "068b3815e36b4a72a59bae017011b29f", "Code": "ZX-001-10+", "Name": "10μL加长 Tip头"}}
"uuid": "548bbc3df0d4447586f2c19d2c0c0c55", )
"Code": "HPLC01",
"Name": "HPLC料盘"
}
})
plate2 = get_well_container("PlateT2") plate2 = get_well_container("PlateT2")
plate2.load_state({ plate2.load_state(
"Material": { {"Material": {"uuid": "b05b3b2aafd94ec38ea0cd3215ecea8f", "Code": "ZX-78-096", "Name": "细菌培养皿"}}
"uuid": "04211a2dc93547fe9bf6121eac533650", )
}
})
plate3 = get_well_container("PlateT3") plate3 = get_well_container("PlateT3")
plate3.load_state({ plate3.load_state(
"Material": { {
"uuid": "04211a2dc93547fe9bf6121eac533650", "Material": {
"uuid": "04211a2dc93547fe9bf6121eac533650",
}
} }
}) )
trash = PRCXI9300Trash(name="trash", size_x=50, size_y=50, size_z=10, category="trash") plate4 = get_well_container("PlateT4")
trash.load_state({ plate4.load_state(
"Material": { {"Material": {"uuid": "b05b3b2aafd94ec38ea0cd3215ecea8f", "Code": "ZX-78-096", "Name": "细菌培养皿"}}
"uuid": "730067cf07ae43849ddf4034299030e9" )
}
})
plate5 = get_well_container("PlateT5") plate5 = get_well_container("PlateT5")
plate5.load_state({ plate5.load_state(
"Material": { {
"uuid": "04211a2dc93547fe9bf6121eac533650", "Material": {
"uuid": "076250742950465b9d6ea29a225dfb00",
"Code": "ZX-001-300",
"SupplyType": 1,
"Name": "300μL Tip头",
}
} }
}) )
plate6 = get_well_container("PlateT6") plate6 = get_well_container("PlateT6")
plate6.load_state({ plate6.load_state(
"Material": { {
"uuid": "04211a2dc93547fe9bf6121eac533650" "Material": {
"uuid": "e146697c395e4eabb3d6b74f0dd6aaf7",
"Code": "1",
"SupplyType": 1,
"Name": "ep适配器",
"SummaryName": "ep适配器",
}
} }
}) )
plate7 = PRCXI9300Container(name="plateT7", size_x=50, size_y=50, size_z=10, category="plate", ordering=collections.OrderedDict()) plate7 = PRCXI9300Container(
plate7.load_state({ name="plateT7", size_x=50, size_y=50, size_z=10, category="plate", ordering=collections.OrderedDict()
"Material": { )
"uuid": "04211a2dc93547fe9bf6121eac533650" plate7.load_state({"Material": {"uuid": "04211a2dc93547fe9bf6121eac533650"}})
}
})
plate8 = get_tip_rack("RackT8") plate8 = get_tip_rack("RackT8")
plate8.load_state({ plate8.load_state({"Material": {"uuid": "04211a2dc93547fe9bf6121eac533650"}})
"Material": {
"uuid": "068b3815e36b4a72a59bae017011b29f",
"Code": "ZX-001-10+",
"Name": "10μL加长 Tip头"
}
})
plate9 = get_well_container("PlateT9") plate9 = get_well_container("PlateT9")
plate9.load_state({ plate9.load_state(
"Material": { {
"uuid": "04211a2dc93547fe9bf6121eac533650" "Material": {
"uuid": "4a043a07c65a4f9bb97745e1f129b165",
"Code": "ZX-58-0001",
"SupplyType": 2,
"Name": "全裙边 PCR适配器",
"SummaryName": "全裙边 PCR适配器",
}
} }
}) )
plate10 = get_well_container("PlateT10") plate10 = get_well_container("PlateT10")
plate10.load_state({ plate10.load_state(
"Material": { {
"uuid": "04211a2dc93547fe9bf6121eac533650" "Material": {
"uuid": "4a043a07c65a4f9bb97745e1f129b165",
"Code": "ZX-58-0001",
"SupplyType": 2,
"Name": "全裙边 PCR适配器",
"SummaryName": "全裙边 PCR适配器",
}
} }
}) )
plate11 = get_well_container("PlateT11") plate11 = get_well_container("PlateT11")
plate11.load_state({ plate11.load_state(
"Material": { {
"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "Material": {
"uuid": "04211a2dc93547fe9bf6121eac533650",
}
} }
}) )
plate12 = get_well_container("PlateT12") plate12 = get_well_container("PlateT12")
plate12.load_state({ plate12.load_state({"Material": {"uuid": "04211a2dc93547fe9bf6121eac533650"}})
"Material": {
"uuid": "04211a2dc93547fe9bf6121eac533650"
}
})
plate13 = get_well_container("PlateT13") plate13 = get_well_container("PlateT13")
plate13.load_state({ plate13.load_state(
"Material": { {
"uuid": "04211a2dc93547fe9bf6121eac533650" "Material": {
"uuid": "4a043a07c65a4f9bb97745e1f129b165",
"Code": "ZX-58-0001",
"SupplyType": 2,
"Name": "全裙边 PCR适配器",
"SummaryName": "全裙边 PCR适配器",
}
} }
}) ),
plate14 = get_well_container("PlateT14")
plate14.load_state(
{
"Material": {
"uuid": "4a043a07c65a4f9bb97745e1f129b165",
"Code": "ZX-58-0001",
"SupplyType": 2,
"Name": "全裙边 PCR适配器",
"SummaryName": "全裙边 PCR适配器",
}
}
),
plate15 = get_well_container("PlateT15")
plate15.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"}})
# container_for_nothing = PRCXI9300Container(name="container_for_nothing", size_x=50, size_y=50, size_z=10, category="plate", ordering=collections.OrderedDict()) # 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(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(plate2, 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(
PRCXI9300Container(
name="container_for_nothin3",
size_x=50,
size_y=50,
size_z=10,
category="plate",
ordering=collections.OrderedDict(),
),
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))
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),
)
deck.assign_child_resource(plate9, location=Coordinate(0, 0, 0))
deck.assign_child_resource(plate10, location=Coordinate(0, 0, 0))
deck.assign_child_resource(
PRCXI9300Container(
name="container_for_nothing11",
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_nothing12",
size_x=50,
size_y=50,
size_z=10,
category="plate",
ordering=collections.OrderedDict(),
),
location=Coordinate(0, 0, 0),
)
deck.assign_child_resource(plate13, location=Coordinate(0, 0, 0))
deck.assign_child_resource(plate14, location=Coordinate(0, 0, 0))
deck.assign_child_resource(plate15, location=Coordinate(0, 0, 0))
deck.assign_child_resource(trash, 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="172.21.5.75", port=9999, from unilabos.resources.graphio import tree_to_list, resource_plr_to_ulab
timeout=10.0, setup=False, debug=True,
matrix_id="c1d0d5dc-40f2-4f24-97ac-9cc49c68496c",
channel_num=1, axis="Left",simulator=True) # Initialize the handler with the deck and host settings
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)
handler = PRCXI9300Handler(
deck=deck,
host="192.168.0.121",
port=9999,
timeout=10.0,
setup=False,
debug=False,
matrix_id="5de524d0-3f95-406c-86dd-f83626ebc7cb",
channel_num=1,
axis="Left",
simulator=False,
is_9320=True,
) # Initialize the handler with the deck and host settings
backend: PRCXI9300Backend = handler.backend
res = backend.api_client.get_all_materials()
handler.set_tiprack([plate8]) # Set the tip rack for the handler handler.set_tiprack([plate8]) # 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
from pylabrobot.resources import set_volume_tracking from pylabrobot.resources import set_volume_tracking
# from pylabrobot.resources import set_tip_tracking # from pylabrobot.resources import set_tip_tracking
set_volume_tracking(enabled=True) set_volume_tracking(enabled=True)
@@ -1406,7 +1643,7 @@ if __name__ == "__main__":
# 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 # 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 * # 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
# A = tree_to_list([resource_plr_to_ulab(deck)]) # A = tree_to_list([resource_plr_to_ulab(deck)])
# # with open("deck.json", "w", encoding="utf-8") as f: # # with open("deck.json", "w", encoding="utf-8") as f:
@@ -1521,6 +1758,7 @@ if __name__ == "__main__":
# # # print("PRCXI9300Handler initialized with deck and host settings.") # # # print("PRCXI9300Handler initialized with deck and host settings.")
# 一些推荐版位组合的测试样例:
# 一些推荐版位组合的测试样例: # 一些推荐版位组合的测试样例:
@@ -1545,3 +1783,21 @@ if __name__ == "__main__":
("reagent_4", "10μL加长 Tip头", 1), ("reagent_4", "10μL加长 Tip头", 1),
]) ])
# with open("prcxi_material.json", "r") as f:
# material_info = json.load(f)
# layout = DefaultLayout("PRCXI9320")
# layout.add_lab_resource(material_info)
# MatrixLayout_1, dict_1 = layout.recommend_layout([
# ("reagent_1", "96 细胞培养皿", 3),
# ("reagent_2", "12道储液槽", 1),
# ("reagent_3", "200μL Tip头", 7),
# ("reagent_4", "10μL加长 Tip头", 1),
# ])
# print(dict_1)
# MatrixLayout_2, dict_2 = layout.recommend_layout([
# ("reagent_1", "96深孔板", 4),
# ("reagent_2", "12道储液槽", 1),
# ("reagent_3", "200μL Tip头", 1),
# ("reagent_4", "10μL加长 Tip头", 1),
# ])