From 94cdcbf24eb1dacfd0b3da8b070e8d382a5deba3 Mon Sep 17 00:00:00 2001 From: Guangxin Zhang Date: Mon, 15 Sep 2025 00:29:16 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AF=B9=E4=BA=8EPRCXI9320=E7=9A=84transfer=5F?= =?UTF-8?q?group=EF=BC=8C=E4=B8=80=E5=AF=B9=E5=A4=9A=E5=92=8C=E5=A4=9A?= =?UTF-8?q?=E5=AF=B9=E5=A4=9A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- test/experiments/prcxi_9320.json | 8 +- .../liquid_handler_abstract.py | 45 +- .../prcxi/abstract_protocol.py | 568 ++++++++++++++++++ .../devices/liquid_handling/prcxi/prcxi.py | 477 +++++++-------- 4 files changed, 844 insertions(+), 254 deletions(-) create mode 100644 unilabos/devices/liquid_handling/prcxi/abstract_protocol.py diff --git a/test/experiments/prcxi_9320.json b/test/experiments/prcxi_9320.json index d72768f7..c938ef6d 100644 --- a/test/experiments/prcxi_9320.json +++ b/test/experiments/prcxi_9320.json @@ -16,15 +16,15 @@ "_resource_child_name": "PRCXI_Deck", "_resource_type": "unilabos.devices.liquid_handling.prcxi.prcxi:PRCXI9300Deck" }, - "host": "10.181.102.13", + "host": "172.21.5.75", "port": 9999, "timeout": 10.0, "axis": "Right", "channel_num": 1, "setup": false, - "debug": false, - "simulator": false, - "matrix_id": "fd383e6d-2d0e-40b5-9c01-1b2870b1f1b1" + "debug": true, + "simulator": true, + "matrix_id": "c1d0d5dc-40f2-4f24-97ac-9cc49c68496c" }, "data": {}, "children": [ diff --git a/unilabos/devices/liquid_handling/liquid_handler_abstract.py b/unilabos/devices/liquid_handling/liquid_handler_abstract.py index f5958129..4586f808 100644 --- a/unilabos/devices/liquid_handling/liquid_handler_abstract.py +++ b/unilabos/devices/liquid_handling/liquid_handler_abstract.py @@ -1,5 +1,5 @@ from __future__ import annotations - +import re import traceback from typing import List, Sequence, Optional, Literal, Union, Iterator, Dict, Any, Callable, Set, cast from collections import Counter @@ -558,14 +558,16 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware): # --------------------------------------------------------------- def set_group(self, group_name: str, wells: List[Well], volumes: List[float]): - if len(wells) != 8: + if self.channel_num == 8 and len(wells) != 8: raise RuntimeError(f"Expected 8 wells, got {len(wells)}") self.group_info[group_name] = wells self.set_liquid(wells, [group_name] * len(wells), volumes) async def transfer_group(self, source_group_name: str, target_group_name: str, unit_volume: float): + source_wells = self.group_info.get(source_group_name, []) target_wells = self.group_info.get(target_group_name, []) + rack_info = dict() for child in self.deck.children: if issubclass(child.__class__, TipRack): @@ -576,6 +578,7 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware): break else: rack_info[rack.name] = (rack, tip.maximal_volume - unit_volume) + if len(rack_info) == 0: raise ValueError(f"No tip rack can support volume {unit_volume}.") @@ -583,24 +586,40 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware): for child in self.deck.children: if child.name == rack_info[0][0]: target_rack = child - target_rack = cast(TipRack, target_rack) - available_tips = {} - for (idx, name), tipSpot in zip(target_rack._ordering.items(), target_rack.get_all_items()): + for (idx, tipSpot) in enumerate(target_rack.get_all_items()): if tipSpot.has_tip(): available_tips[idx] = tipSpot continue + # 一般移动液体有两种方式，一对多和多对多 + if self.channel_num == 8: - colnum_list = [int(holename[1:]) for holename in available_tips.keys()] - available_cols = [colnum for colnum, count in dict(Counter(colnum_list)).items() if count == 8] - available_cols.sort() # 这样就确定了列号 - tips_to_use = [available_tips[f"{chr(65 + i)}{available_cols[0]}"] for i in range(8)] + tip_prefix = list(available_tips.values())[0].name.split('_')[0] + colnum_list = [int(tip.name.split('_')[-1][1:]) for tip in available_tips.values()] + available_cols = [colnum for colnum, count in dict(Counter(colnum_list)).items() if count == 8] + available_cols.sort() + available_tips_dict = {tip.name: tip for tip in available_tips.values()} + tips_to_use = [available_tips_dict[f"{tip_prefix}_{chr(65 + i)}{available_cols[0]}"] for i in range(8)] + await self.pick_up_tips(tips_to_use, use_channels=list(range(0, 8))) + await self.aspirate(source_wells, [unit_volume] * 8, use_channels=list(range(0, 8))) + await self.dispense(target_wells, [unit_volume] * 8, use_channels=list(range(0, 8))) + await self.discard_tips(use_channels=list(range(0, 8))) - await self.pick_up_tips(tips_to_use, use_channels=list(range(0, 8))) - await self.aspirate(source_wells, [10] * 8, use_channels=list(range(0, 8))) - await self.dispense(target_wells, [10] * 8, use_channels=list(range(0, 8))) - await self.discard_tips(use_channels=list(range(0, 8))) + elif self.channel_num == 1: + + for num_well in range(len(target_wells)): + tip_to_use = available_tips[list(available_tips.keys())[num_well]] + await self.pick_up_tips([tip_to_use], use_channels=[0]) + if len(source_wells) == 1: + await self.aspirate([source_wells[0]], [unit_volume], use_channels=[0]) + else: + await self.aspirate([source_wells[num_well]], [unit_volume], use_channels=[0]) + await self.dispense([target_wells[num_well]], [unit_volume], use_channels=[0]) + await self.discard_tips(use_channels=[0]) + + else: + raise ValueError(f"Unsupported channel number {self.channel_num}.") async def create_protocol( self, diff --git a/unilabos/devices/liquid_handling/prcxi/abstract_protocol.py b/unilabos/devices/liquid_handling/prcxi/abstract_protocol.py new file mode 100644 index 00000000..83def461 --- /dev/null +++ b/unilabos/devices/liquid_handling/prcxi/abstract_protocol.py @@ -0,0 +1,568 @@ +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 MaterialResource: + """统一的液体/反应器资源，支持多孔（wells）场景： + - wells: 列表，每个元素代表一个物料孔（unit）； + - units: 与 wells 对齐的列表，每个元素是 {liquid_id: volume}； + - 若传入 liquid_id + volume 或 composition，总量将**等分**到各 unit； + """ + def __init__( + self, + resource_name: str, + slot: int, + well: List[int], + composition: Optional[Dict[str, float]] = None, + liquid_id: Optional[str] = None, + volume: Union[float, int] = 0.0, + is_supply: Optional[bool] = None, + ): + self.resource_name = resource_name + self.slot = int(slot) + self.well = list(well or []) + self.is_supply = bool(is_supply) if is_supply is not None else (bool(composition) or (liquid_id is not None)) + + # 规范化：至少有 1 个 unit + n = max(1, len(self.well)) + self.units: List[Dict[str, float]] = [dict() for _ in range(n)] + + # 初始化内容：等分到各 unit + if composition: + for k, v in composition.items(): + share = float(v) / n + for u in self.units: + if share > 0: + u[k] = u.get(k, 0.0) + share + elif liquid_id is not None and float(volume) > 0: + share = float(volume) / n + for u in self.units: + u[liquid_id] = u.get(liquid_id, 0.0) + share + + # 位置描述 + def location(self) -> Dict[str, Any]: + return {"slot": self.slot, "well": self.well} + + def unit_count(self) -> int: + return len(self.units) + + def unit_volume(self, idx: int) -> float: + return float(sum(self.units[idx].values())) + + def total_volume(self) -> float: + return float(sum(self.unit_volume(i) for i in range(self.unit_count()))) + + def add_to_unit(self, idx: int, liquid_id: str, vol: Union[float, int]): + v = float(vol) + if v < 0: + return + u = self.units[idx] + if liquid_id not in u: + u[liquid_id] = 0.0 + if v > 0: + u[liquid_id] += v + + def remove_from_unit(self, idx: int, total: Union[float, int]) -> Dict[str, float]: + take = float(total) + if take <= 0: return {} + u = self.units[idx] + avail = sum(u.values()) + if avail <= 0: return {} + take = min(take, avail) + ratio = take / avail + removed: Dict[str, float] = {} + for k, v in list(u.items()): + dv = v * ratio + nv = v - dv + if nv < 1e-9: nv = 0.0 + u[k] = nv + removed[k] = dv + + self.units[idx] = {k: v for k, v in u.items() if v > 0} + return removed + + def transfer_unit_to(self, src_idx: int, other: "MaterialResource", dst_idx: int, total: Union[float, int]): + moved = self.remove_from_unit(src_idx, total) + for k, v in moved.items(): + other.add_to_unit(dst_idx, k, v) + + def get_resource(self) -> Dict[str, Any]: + return { + "resource_name": self.resource_name, + "slot": self.slot, + "well": self.well, + "units": [dict(u) for u in self.units], + "total_volume": self.total_volume(), + "is_supply": self.is_supply, + } + +def transfer_liquid( + sources: MaterialResource, + targets: MaterialResource, + unit_volume: Optional[Union[float, int]] = None, + tip: Optional[str] = None, #这里应该是指定种类的 +) -> Dict[str, Any]: + try: + vol_each = float(unit_volume) + except (TypeError, ValueError): + return {"action": "transfer_liquid", "error": "invalid unit_volume"} + if vol_each <= 0: + return {"action": "transfer_liquid", "error": "non-positive volume"} + + ns, nt = sources.unit_count(), targets.unit_count() + # one-to-many: 从单个 source unit(0) 扇出到目标各 unit + if ns == 1 and nt >= 1: + for j in range(nt): + sources.transfer_unit_to(0, targets, j, vol_each) + # many-to-many: 数量相同，逐一对应 + elif ns == nt and ns > 0: + for i in range(ns): + sources.transfer_unit_to(i, targets, i, vol_each) + else: + raise ValueError(f"Unsupported mapping: sources={ns} units, targets={nt} units. Only 1->N or N->N are allowed.") + + return { + "action": "transfer_liquid", + "sources": sources.get_resource(), + "targets": targets.get_resource(), + "unit_volume": unit_volume, + "tip": tip, + } + +def plan_transfer(pm: "ProtocolManager", **kwargs) -> Dict[str, Any]: + """Shorthand to add a non-committing transfer to a ProtocolManager. + Accepts the same kwargs as ProtocolManager.add_transfer. + """ + return pm.add_transfer(**kwargs) + +class ProtocolManager: + """Plan/track transfers and back‑solve minimum initial volumes. + + Use add_transfer(...) to register steps (no mutation). + Use compute_min_initials(...) to infer the minimal starting volume of each liquid + per resource required to execute the plan in order. + """ + + # ---------- lifecycle ---------- + def __init__(self): + # queued logical steps (keep live refs to MaterialResource) + self.steps: List[Dict[str, Any]] = [] + + # simple tip catalog; choose the smallest that meets min_aspirate and capacity*safety + self.tip_catalog = [ + {"name": "TIP_10uL", "capacity": 10.0, "min_aspirate": 0.5}, + {"name": "TIP_20uL", "capacity": 20.0, "min_aspirate": 1.0}, + {"name": "TIP_50uL", "capacity": 50.0, "min_aspirate": 2.0}, + {"name": "TIP_200uL", "capacity": 200.0, "min_aspirate": 5.0}, + {"name": "TIP_300uL", "capacity": 300.0, "min_aspirate": 10.0}, + {"name": "TIP_1000uL", "capacity": 1000.0, "min_aspirate": 20.0}, + ] + + # stable labels for unknown liquids per resource (A, B, C, ..., AA, AB, ...) + self._unknown_labels: Dict[MaterialResource, str] = {} + self._unknown_label_counter: int = 0 + + # ---------- public API ---------- + def recommend_tip(self, unit_volume: float, safety: float = 1.10) -> str: + v = float(unit_volume) + # prefer: meets min_aspirate and capacity with safety margin; else fallback to capacity-only; else max capacity + eligible = [t for t in self.tip_catalog if t["min_aspirate"] <= v and t["capacity"] >= v * safety] + if not eligible: + eligible = [t for t in self.tip_catalog if t["capacity"] >= v] + return min(eligible or self.tip_catalog, key=lambda t: t["capacity"]) ["name"] + + def get_tip_capacity(self, tip_name: str) -> Optional[float]: + for t in self.tip_catalog: + if t["name"] == tip_name: + return t["capacity"] + return None + + def add_transfer( + self, + sources: MaterialResource, + targets: MaterialResource, + unit_volume: Union[float, int], + tip: Optional[str] = None, + ) -> Dict[str, Any]: + step = { + "action": "transfer_liquid", + "sources": sources, + "targets": targets, + "unit_volume": float(unit_volume), + "tip": tip or self.recommend_tip(unit_volume), + } + self.steps.append(step) + # return a serializable shadow (no mutation) + return { + "action": "transfer_liquid", + "sources": sources.get_resource(), + "targets": targets.get_resource(), + "unit_volume": step["unit_volume"], + "tip": step["tip"], + } + + @staticmethod + def _liquid_keys_of(resource: MaterialResource) -> List[str]: + keys: set[str] = set() + for u in resource.units: + keys.update(u.keys()) + return sorted(keys) + + @staticmethod + def _fanout_multiplier(ns: int, nt: int) -> Optional[int]: + """Return the number of liquid movements for a mapping shape. + 1->N: N moves; N->N: N moves; otherwise unsupported (None). + """ + if ns == 1 and nt >= 1: + return nt + if ns == nt and ns > 0: + return ns + return None + + # ---------- planning core ---------- + def compute_min_initials( + self, + use_initial: bool = False, + external_only: bool = True, + ) -> Dict[str, Dict[str, float]]: + """Simulate the plan (non‑mutating) and return minimal starting volumes per resource/liquid.""" + ledger: Dict[MaterialResource, Dict[str, float]] = {} + min_seen: Dict[MaterialResource, Dict[str, float]] = {} + + def _ensure(res: MaterialResource) -> None: + if res in ledger: + return + declared = self._liquid_keys_of(res) + if use_initial: + # sum actual held amounts across units + totals = {k: 0.0 for k in declared} + for u in res.units: + for k, v in u.items(): + totals[k] = totals.get(k, 0.0) + float(v) + ledger[res] = totals + else: + ledger[res] = {k: 0.0 for k in declared} + min_seen[res] = {k: ledger[res].get(k, 0.0) for k in ledger[res]} + + def _proportions(src: MaterialResource, src_bal: Dict[str, float]) -> tuple[List[str], Dict[str, float]]: + keys = list(src_bal.keys()) + total_pos = sum(x for x in src_bal.values() if x > 0) + + # if ledger has no keys yet, seed from declared types on the resource + if not keys: + keys = self._liquid_keys_of(src) + for k in keys: + src_bal.setdefault(k, 0.0) + min_seen[src].setdefault(k, 0.0) + + if total_pos > 0: + # proportional to current positive balances + props = {k: (src_bal.get(k, 0.0) / total_pos) for k in keys} + return keys, props + + # no material currently: evenly from known keys, or assign an unknown label + if keys: + eq = 1.0 / len(keys) + return keys, {k: eq for k in keys} + + unk = self._label_for_unknown(src) + keys = [unk] + src_bal.setdefault(unk, 0.0) + min_seen[src].setdefault(unk, 0.0) + return keys, {unk: 1.0} + + for step in self.steps: + if step.get("action") != "transfer_liquid": + continue + + src: MaterialResource = step["sources"] + dst: MaterialResource = step["targets"] + vol = float(step["unit_volume"]) + if vol <= 0: + continue + + _ensure(src) + _ensure(dst) + + mult = self._fanout_multiplier(src.unit_count(), dst.unit_count()) + if not mult: + continue # unsupported mapping shape for this planner + + eff_vol = vol * mult + src_bal = ledger[src] + keys, props = _proportions(src, src_bal) + + # subtract from src; track minima; accumulate to dst + moved: Dict[str, float] = {} + for k in keys: + dv = eff_vol * props[k] + src_bal[k] = src_bal.get(k, 0.0) - dv + moved[k] = dv + prev_min = min_seen[src].get(k, 0.0) + if src_bal[k] < prev_min: + min_seen[src][k] = src_bal[k] + + dst_bal = ledger[dst] + for k, dv in moved.items(): + dst_bal[k] = dst_bal.get(k, 0.0) + dv + min_seen[dst].setdefault(k, dst_bal[k]) + + # convert minima (negative) to required initials + result: Dict[str, Dict[str, float]] = {} + for res, mins in min_seen.items(): + if external_only and not getattr(res, "is_supply", False): + continue + need = {liq: max(0.0, -mn) for liq, mn in mins.items() if mn < 0.0} + if need: + result[res.resource_name] = need + return result + + def compute_tip_consumption(self) -> Dict[str, Any]: + """Compute how many tips are consumed at each transfer step, and aggregate by tip type. + Rule: each liquid movement (source unit -> target unit) consumes one tip. + For supported shapes: 1->N uses N tips; N->N uses N tips. + """ + per_step: List[Dict[str, Any]] = [] + totals_by_tip: Dict[str, int] = {} + + for i, s in enumerate(self.steps): + if s.get("action") != "transfer_liquid": + continue + ns = s["sources"].unit_count() + nt = s["targets"].unit_count() + moves = self._fanout_multiplier(ns, nt) or 0 + tip_name = s.get("tip") or self.recommend_tip(s["unit_volume"]) # per-step tip may vary + per_step.append({ + "idx": i, + "tip": tip_name, + "tips_used": moves, + "moves": moves, + }) + totals_by_tip[tip_name] = totals_by_tip.get(tip_name, 0) + int(moves) + + return {"per_step": per_step, "totals_by_tip": totals_by_tip} + + def compute_min_initials_with_tips( + self, + use_initial: bool = False, + external_only: bool = True, + ) -> Dict[str, Any]: + needs = self.compute_min_initials(use_initial=use_initial, external_only=external_only) + step_tips: List[Dict[str, Any]] = [] + totals_by_tip: Dict[str, int] = {} + + for i, s in enumerate(self.steps): + if s.get("action") != "transfer_liquid": + continue + ns = s["sources"].unit_count() + nt = s["targets"].unit_count() + moves = self._fanout_multiplier(ns, nt) or 0 + tip_name = s.get("tip") or self.recommend_tip(s["unit_volume"]) # step-specific tip + totals_by_tip[self.get_tip_capacity(tip_name)] = totals_by_tip.get(tip_name, 0) + int(moves) + + step_tips.append({ + "idx": i, + "tip": tip_name, + "tip_capacity": self.get_tip_capacity(tip_name), + "unit_volume": s["unit_volume"], + "tips_used": moves, + }) + return {"liquid_setup": needs, "step_tips": step_tips, "totals_by_tip": totals_by_tip} + + # ---------- unknown labels ---------- + def _index_to_letters(self, idx: int) -> str: + """0->A, 1->B, ... 25->Z, 26->AA, 27->AB ... (Excel-like)""" + s: List[str] = [] + idx = int(idx) + while True: + idx, r = divmod(idx, 26) + s.append(chr(ord('A') + r)) + if idx == 0: + break + idx -= 1 # Excel-style carry + return "".join(reversed(s)) + + def _label_for_unknown(self, res: MaterialResource) -> str: + """Assign a stable unknown-liquid label (A/B/C/...) per resource.""" + if res not in self._unknown_labels: + lab = self._index_to_letters(self._unknown_label_counter) + self._unknown_label_counter += 1 + self._unknown_labels[res] = lab + return self._unknown_labels[res] + + +# 在这一步传输目前有的物料 +class LabResource: + def __init__(self): + self.tipracks = [] + self.plates = [] + self.trash = [] + + def add_tipracks(self, tiprack: List[TipRack]): + self.tipracks.extend(tiprack) + def add_plates(self, plate: List[Plate]): + self.plates.extend(plate) + def add_trash(self, trash: List[Plate]): + self.trash.extend(trash) + def get_resources_info(self) -> Dict[str, Any]: + tipracks = [{"name": tr.name, "max_volume": tr.children[0].tracker._tip.maximal_volume, "count": len(tr.children)} for tr in self.tipracks] + plates = [{"name": pl.name, "max_volume": pl.children[0].max_volume, "count": len(pl.children)} for pl in self.plates] + trash = [{"name": t.name, "max_volume": t.children[0].max_volume, "count": len(t.children)} for t in self.trash] + return { + "tipracks": tipracks, + "plates": plates, + "trash": trash + } + +from typing import Dict, Any + +class DefaultLayout: + + def __init__(self, product_name: str = "PRCXI9300"): + self.labresource = None + if product_name not in ["PRCXI9300", "PRCXI9320"]: + raise ValueError(f"Unsupported product_name: {product_name}. Only 'PRCXI9300' and 'PRCXI9320' are supported.") + + if product_name == "PRCXI9300": + self.rows = 2 + self.columns = 3 + self.layout = [1, 2, 3, 4, 5, 6] + self.trash_slot = 3 + self.waste_liquid_slot = 6 + elif product_name == "PRCXI9320": + self.rows = 3 + self.columns = 4 + self.layout = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12] + self.trash_slot = 3 + self.waste_liquid_slot = 12 + + def get_layout(self) -> Dict[str, Any]: + return { + "rows": self.rows, + "columns": self.columns, + "layout": self.layout, + "trash_slot": self.trash_slot, + "waste_liquid_slot": self.waste_liquid_slot + } + + def get_trash_slot(self) -> int: + return self.trash_slot + + def get_waste_liquid_slot(self) -> int: + return self.waste_liquid_slot + + def set_liquid_handler_layout(self, product_name: str): + if product_name == "PRCXI9300": + self.rows = 2 + self.columns = 3 + self.layout = [1, 2, 3, 4, 5, 6] + self.trash_slot = 3 + self.waste_liquid_slot = 6 + + elif product_name == "PRCXI9320": + self.rows = 3 + self.columns = 4 + self.layout = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12] + self.trash_slot = 3 + self.waste_liquid_slot = 12 + + def set_trash_slot(self, slot: int): + self.trash_slot = slot + + def set_waste_liquid_slot(self, slot: int): + self.waste_liquid_slot = slot + + def add_lab_resource(self, lab_resource: LabResource): + self.labresource = lab_resource.get_resources_info() + + def recommend_layout(self, needs: Dict[str, Any]) -> Dict[str, Any]: + """根据 needs 推荐布局""" + liquid_info = needs['liquid_setup'] + tip_info = needs['totals_by_tip'] # 修改这里：直接访问 totals_by_tip + print("当前实验所需物料信息：", liquid_info) + print("当前实验所需枪头信息：", tip_info) + print(self.labresource) + + for liquid in liquid_info: + # total_volume = liquid.values() + print(liquid) + #print(f"资源 {liquid} 需要的总体积: {total_volume}") + +if __name__ == "__main__": + # ---- 资源：SUP 供液（X），中间板 R1（4 孔空），目标板 R2（4 孔空）---- + sup = MaterialResource("SUP", slot=5, well=[1], liquid_id="X", volume=10000) + r1 = MaterialResource("R1", slot=6, well=[1,2,3,4,5,6,7,8]) + r2 = MaterialResource("R2", slot=7, well=[1,2,3,4,5,6,7,8]) + + pm = ProtocolManager() + # 步骤1：SUP -> R1，1->N 扇出，每孔 50 uL（总 200 uL） + pm.add_transfer(sup, r1, unit_volume=10.0) + # 步骤2：R1 -> R2，N->N 对应，每对 25 uL（总 100 uL；来自 R1 中已存在的混合物 X） + pm.add_transfer(r1, r2, unit_volume=120.0) + + out = pm.compute_min_initials_with_tips() + + + + + # layout_planer = DefaultLayout('PRCXI9320') + # print(layout_planer.get_layout()) + # print("回推最小需求：", out["liquid_setup"]) # {'SUP': {'X': 200.0}} + # print("步骤枪头建议：", out["step_tips"]) # [{'idx':0,'tip':'TIP_200uL','unit_volume':50.0}, {'idx':1,'tip':'TIP_50uL','unit_volume':25.0}] + +# # 实际执行（可选） +# transfer_liquid(sup, r1, unit_volume=50.0) +# transfer_liquid(r1, r2, unit_volume=25.0) +# print("执行后 SUP：", sup.get_resource()) # 总体积 -200 +# print("执行后 R1：", r1.get_resource()) # 每孔 25 uL（50 进 -25 出） +# print("执行后 R2：", r2.get_resource()) # 每孔 25 uL + + + from pylabrobot.resources.opentrons.tube_racks import * + from pylabrobot.resources.opentrons.plates import * + from pylabrobot.resources.opentrons.tip_racks import * + from pylabrobot.resources.opentrons.reservoirs import * + + plate = [locals()['nest_96_wellplate_2ml_deep'](name="thermoscientificnunc_96_wellplate_2000ul"), locals()['corning_96_wellplate_360ul_flat'](name="corning_96_wellplate_360ul_flat")] + tiprack = [locals()['opentrons_96_tiprack_300ul'](name="opentrons_96_tiprack_300ul"), locals()['opentrons_96_tiprack_1000ul'](name="opentrons_96_tiprack_1000ul")] + trash = [locals()['axygen_1_reservoir_90ml'](name="axygen_1_reservoir_90ml")] + + from pprint import pprint + + lab_resource = LabResource() + lab_resource.add_tipracks(tiprack) + lab_resource.add_plates(plate) + lab_resource.add_trash(trash) + + layout_planer = DefaultLayout('PRCXI9300') + layout_planer.add_lab_resource(lab_resource) + layout_planer.recommend_layout(out) \ No newline at end of file diff --git a/unilabos/devices/liquid_handling/prcxi/prcxi.py b/unilabos/devices/liquid_handling/prcxi/prcxi.py index 2c69faa4..f1652f89 100644 --- a/unilabos/devices/liquid_handling/prcxi/prcxi.py +++ b/unilabos/devices/liquid_handling/prcxi/prcxi.py @@ -981,121 +981,119 @@ if __name__ == "__main__": # 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.plates import corning_96_wellplate_360ul_flat, nest_96_wellplate_2ml_deep + # 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_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 + # 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头" - } - }) + # 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深孔板" - } - }) + # 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" - } - }) + # 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深孔板" - } - }) + # 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") + # 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深孔板" - } - }) + # 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)) + # 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 + # # # plate_2_liquids = [[('water', 500)]]*96 - # plate2.set_well_liquids(plate_2_liquids) + # # # 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 + # 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) - - - 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 + # 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) + # 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.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]) @@ -1166,164 +1164,172 @@ if __name__ == "__main__": ### 9320 ### -# 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.plates import corning_96_wellplate_360ul_flat, nest_96_wellplate_2ml_deep + 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_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 + 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" -# } -# }) + 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()) + # 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)) + 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 = PRCXI9300Handler(deck=deck, host="172.21.5.75", port=9999, + 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 -# 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) + 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 + plate_2_liquids = handler.set_group("water", [plate2.children[0]], [300]) + #print(plate_2_liquids) + plate5_liquids = handler.set_group("master_mix", plate5.children[:23], [100]*23) + #print(plate5_liquids) -# 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 + + 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.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.transfer_group("water", "master_mix", 10)) # Reset tip tracking + + + # 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([0])) # asyncio.run(handler.add_liquid( # asp_vols=[10]*7, @@ -1341,10 +1347,7 @@ if __name__ == "__main__": # spread="custom", # )) -# asyncio.run(handler.run_protocol()) # Run the protocol - - - + # asyncio.run(handler.run_protocol()) # Run the protocol @@ -1413,7 +1416,7 @@ if __name__ == "__main__": # # # )) # # 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 + #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.")