open_pub/Uni-Lab-OS
1
0
Fork 0
mirror of https://github.com/deepmodeling/Uni-Lab-OS synced 2026-03-27 20:46:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'prcix9320' into feat/lab_resource

Browse Source
This commit is contained in:
q434343
2026-03-23 00:38:40 +08:00
parent 35bcf6765d 41be9e4e19
commit 9d65718f37
74 changed files with 21875 additions and 19202 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -45,6 +45,7 @@ from pylabrobot.resources import (
Trash,
PlateAdapter,
TubeRack,
create_homogeneous_resources,
)
from unilabos.devices.liquid_handling.liquid_handler_abstract import (
@@ -55,8 +56,8 @@ from unilabos.devices.liquid_handling.liquid_handler_abstract import (
TransferLiquidReturn,
)
from unilabos.registry.placeholder_type import ResourceSlot
from unilabos.resources.resource_tracker import ResourceTreeSet
from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode
from unilabos.resources.itemized_carrier import ItemizedCarrier
from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode, ROS2DeviceNode
class PRCXIError(RuntimeError):
@@ -92,12 +93,8 @@ class PRCXI9300Deck(Deck):
"""
# T1-T16 默认位置 (4列×4行)
_DEFAULT_SITE_POSITIONS = [
(0, 0, 0), (138, 0, 0), (276, 0, 0), (414, 0, 0), # T1-T4
(0, 96, 0), (138, 96, 0), (276, 96, 0), (414, 96, 0), # T5-T8
(0, 192, 0), (138, 192, 0), (276, 192, 0), (414, 192, 0), # T9-T12
(0, 288, 0), (138, 288, 0), (276, 288, 0), (414, 288, 0), # T13-T16
]
_DEFAULT_SITE_POSITIONS = [((i%4)*137.5+5, (3-int(i/4))*96+13, 0) for i in range(0, 16)]
_DEFAULT_SITE_SIZE = {"width": 128.0, "height": 86, "depth": 0}
_DEFAULT_CONTENT_TYPE = ["plate", "tip_rack", "plates", "tip_racks", "tube_rack", "adaptor"]
@@ -262,9 +259,30 @@ class PRCXI9300Plate(Plate):
elif value is None:
ordering_param = ordering
else:
# ordering 的值已经是对象，可以直接使用
items = ordering
ordering_param = None
# ordering 的值是对象（可能是 Well 对象），检查是否有有效的 location
# 如果是反序列化过程，Well 对象可能没有正确的 location，需要让 Plate 重新创建
sample_value = next(iter(ordering.values()), None)
if sample_value is not None and hasattr(sample_value, 'location'):
# 如果是 Well 对象但 location 为 None，说明是反序列化过程
# 让 Plate 自己创建 Well 对象
if sample_value.location is None:
items = None
ordering_param = collections.OrderedDict((k, None) for k in ordering.keys())
else:
# Well 对象有有效的 location，可以直接使用
items = ordering
ordering_param = None
elif sample_value is None:
# ordering 的值都是 None，让 Plate 自己创建 Well 对象
items = None
ordering_param = collections.OrderedDict((k, None) for k in ordering.keys())
else:
# 其他情况，直接使用
items = ordering
ordering_param = None
else:
items = None
ordering_param = collections.OrderedDict() # 提供空的 ordering
# 根据情况传递不同的参数
if items is not None:
@@ -346,9 +364,16 @@ class PRCXI9300TipRack(TipRack):
items = None
ordering_param = collections.OrderedDict((k, None) for k in ordering.keys())
else:
# ordering 的值已经是对象，可以直接使用
items = ordering
ordering_param = None
# ordering 的值已经是对象，需要过滤掉 None 值
# 只保留有效的对象，用于 ordered_items 参数
valid_items = {k: v for k, v in ordering.items() if v is not None}
if valid_items:
items = valid_items
ordering_param = None
else:
# 如果没有有效对象，使用 ordering 参数
items = None
ordering_param = collections.OrderedDict((k, None) for k in ordering.keys())
else:
items = None
ordering_param = None
@@ -407,17 +432,11 @@ class PRCXI9300Trash(Trash):
该类定义了 PRCXI 9300 的工作台布局和槽位信息。
"""
def __init__(
self,
name: str,
size_x: float,
size_y: float,
size_z: float,
category: str = "trash",
material_info: Optional[Dict[str, Any]] = None,
**kwargs,
):
def __init__(self, name: str, size_x: float, size_y: float, size_z: float,
category: str = "plate",
material_info: Optional[Dict[str, Any]] = None,
**kwargs):
if name != "trash":
print(f"Warning: PRCXI9300Trash usually expects name='trash' for backend logic, but got '{name}'.")
super().__init__(name, size_x, size_y, size_z, **kwargs)
@@ -485,20 +504,25 @@ class PRCXI9300TubeRack(TubeRack):
items_to_pass = ordered_items
ordering_param = None
elif ordering is not None:
# 检查 ordering 中的值类型来决定如何处理：
# - 字符串值（从 JSON 反序列化）: 只用键创建 ordering_param
# - None 值（从第二次往返序列化）: 同样只用键创建 ordering_param
# - 对象值（已经是实际的 Resource 对象）: 直接作为 ordered_items 使用
first_val = next(iter(ordering.values()), None) if ordering else None
if not ordering or first_val is None or isinstance(first_val, str):
# ordering 的值是字符串或 None，只使用键（位置信息）创建新的 OrderedDict
# 传递 ordering 参数而不是 ordered_items，让 TubeRack 自己创建 Tube 对象
# 检查 ordering 中的值是否是字符串（从 JSON 反序列化时的情况）
if ordering and isinstance(next(iter(ordering.values()), None), str):
# ordering 的值是字符串，这种情况下我们让 TubeRack 使用默认行为
# 不在初始化时创建 items，而是在 deserialize 后处理
items_to_pass = None
ordering_param = collections.OrderedDict((k, None) for k in ordering.keys())
ordering_param = collections.OrderedDict((k, None) for k in ordering.keys()) # 提供空的 ordering 来满足要求
# 保存 ordering 信息以便后续处理
self._temp_ordering = ordering
else:
# ordering 的值已经是对象，可以直接使用
items_to_pass = ordering
ordering_param = None
# ordering 的值已经是对象，需要过滤掉 None 值
# 只保留有效的对象，用于 ordered_items 参数
valid_items = {k: v for k, v in ordering.items() if v is not None}
if valid_items:
items_to_pass = valid_items
ordering_param = None
else:
# 如果没有有效对象，创建空的 ordered_items
items_to_pass = {}
ordering_param = None
elif items is not None:
# 兼容旧的 items 参数
items_to_pass = items
@@ -520,6 +544,29 @@ class PRCXI9300TubeRack(TubeRack):
if material_info:
self._unilabos_state["Material"] = material_info
# 如果有临时 ordering 信息，在初始化完成后处理
if hasattr(self, '_temp_ordering') and self._temp_ordering:
self._process_temp_ordering()
def _process_temp_ordering(self):
"""处理临时的 ordering 信息，创建相应的 Tube 对象"""
from pylabrobot.resources import Tube, Coordinate
for location, item_type in self._temp_ordering.items():
if item_type == 'Tube' or item_type == 'tube':
# 为每个位置创建 Tube 对象
tube = Tube(name=f"{self.name}_{location}", size_x=10, size_y=10, size_z=50, max_volume=2000.0)
# 使用 assign_child_resource 添加到 rack 中
self.assign_child_resource(tube, location=Coordinate(0, 0, 0))
# 清理临时数据
del self._temp_ordering
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]]:
try:
data = super().serialize_state()
@@ -546,6 +593,97 @@ class PRCXI9300TubeRack(TubeRack):
data.update(safe_state)
return data
class PRCXI9300PlateAdapterSite(ItemizedCarrier):
def __init__(self, name: str, size_x: float, size_y: float, size_z: float,
material_info: Optional[Dict[str, Any]] = None, **kwargs):
# 处理 sites 参数的不同格式
sites = create_homogeneous_resources(
klass=ResourceHolder,
locations=[Coordinate(0, 0, 0)],
resource_size_x=size_x,
resource_size_y=size_y,
resource_size_z=size_z,
name_prefix=name,
)[0]
# 确保不传递重复的参数
kwargs.pop('layout', None)
sites_in = kwargs.pop('sites', None)
# 创建默认的sites字典
sites_dict = {name: sites}
# 优先从 sites_in 读取 'content_type'，否则使用默认值
content_type = [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack"
]
# 如果提供了sites参数，则用sites_in中的值替换sites_dict中对应的元素
if sites_in is not None and isinstance(sites_in, dict):
for site_key, site_value in sites_in.items():
if site_key in sites_dict:
sites_dict[site_key] = site_value
super().__init__(name, size_x, size_y, size_z,
sites=sites_dict,
num_items_x=kwargs.pop('num_items_x', 1),
num_items_y=kwargs.pop('num_items_y', 1),
num_items_z=kwargs.pop('num_items_z', 1),
content_type=content_type,
**kwargs)
self._unilabos_state = {}
if material_info:
self._unilabos_state["Material"] = material_info
def assign_child_resource(self, resource, location=Coordinate(0, 0, 0), reassign=True, spot=None):
"""重写 assign_child_resource 方法，对于适配器位置，不使用索引分配"""
# 直接调用 Resource 的 assign_child_resource，避免 ItemizedCarrier 的索引逻辑
from pylabrobot.resources.resource import Resource
Resource.assign_child_resource(self, resource, location=location, reassign=reassign)
def unassign_child_resource(self, resource):
"""重写 unassign_child_resource 方法，对于适配器位置，不使用 sites 列表"""
# 直接调用 Resource 的 unassign_child_resource，避免 ItemizedCarrier 的 sites 逻辑
from pylabrobot.resources.resource import Resource
Resource.unassign_child_resource(self, resource)
def serialize_state(self) -> Dict[str, Dict[str, Any]]:
try:
data = super().serialize_state()
except AttributeError:
data = {}
# 包含 sites 配置信息，但避免序列化 ResourceHolder 对象
if hasattr(self, 'sites') and self.sites:
# 只保存 sites 的基本信息，不保存 ResourceHolder 对象本身
sites_info = []
for site in self.sites:
if hasattr(site, '__class__') and 'pylabrobot' in str(site.__class__.__module__):
# 对于 pylabrobot 对象，只保存基本信息
sites_info.append({
"__pylabrobot_object__": True,
"class": site.__class__.__name__,
"module": site.__class__.__module__,
"name": getattr(site, 'name', str(site))
})
else:
sites_info.append(site)
data['sites'] = sites_info
return data
def load_state(self, state: Dict[str, Any]) -> None:
"""加载状态，包括 sites 配置信息"""
super().load_state(state)
# 从状态中恢复 sites 配置信息
if 'sites' in state:
self.sites = [state['sites']]
class PRCXI9300PlateAdapter(PlateAdapter):
@@ -650,18 +788,51 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
step_mode=False,
matrix_id="",
is_9320=False,
start_rail=2,
rail_nums=4,
rail_interval=0,
x_increase = -0.003636,
y_increase = -0.003636,
x_offset = -0.8,
y_offset = -37.98,
deck_z = 300,
deck_y = 400,
rail_width=27.5,
xy_coupling = -0.0045,
):
tablets_info = []
count = 0
self.deck_x = (start_rail + rail_nums*5 + (rail_nums-1)*rail_interval) * rail_width
self.deck_y = deck_y
self.deck_z = deck_z
self.x_increase = x_increase
self.y_increase = y_increase
self.x_offset = x_offset
self.y_offset = y_offset
self.xy_coupling = xy_coupling
tablets_info = {}
plate_positions = []
for child in deck.children:
# 如果放其他类型的物料，是不可以的
if hasattr(child, "_unilabos_state") and "Material" in child._unilabos_state:
number = int(child.name.replace("T", ""))
tablets_info.append(
WorkTablets(
Number=number, Code=f"T{number}", Material=child._unilabos_state["Material"]
)
)
number = int(child.name.replace("T", ""))
if child.children:
if "Material" in child.children[0]._unilabos_state:
tablets_info[number] = child.children[0]._unilabos_state["Material"].get("uuid", "730067cf07ae43849ddf4034299030e9")
else:
tablets_info[number] = "730067cf07ae43849ddf4034299030e9"
else:
tablets_info[number] = "730067cf07ae43849ddf4034299030e9"
pos = self.plr_pos_to_prcxi(child)
plate_positions.append(
{
"Number": number,
"XPos": pos.x,
"YPos": pos.y,
"ZPos": pos.z
}
)
if is_9320:
print("当前设备是9320")
# 始终初始化 step_mode 属性
@@ -671,10 +842,38 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
self.step_mode = step_mode
else:
print("9300设备不支持 单点动作模式")
self._unilabos_backend = PRCXI9300Backend(
tablets_info, host, port, timeout, channel_num, axis, setup, debug, matrix_id, is_9320
tablets_info, plate_positions, host, port, timeout, channel_num, axis, setup, debug, matrix_id, is_9320,
x_increase, y_increase, x_offset, y_offset,
deck_z, deck_x=self.deck_x, deck_y=self.deck_y, xy_coupling=xy_coupling
)
super().__init__(backend=self._unilabos_backend, deck=deck, simulator=simulator, channel_num=channel_num)
def plr_pos_to_prcxi(self, resource: Resource):
resource_pos = resource.get_absolute_location(x="c",y="c",z="t")
x = resource_pos.x
y = resource_pos.y
z = resource_pos.z
# 如果z等于0，则递归resource.parent的高度并向z加，使用get_size_z方法
parent = resource.parent
res_z = resource.location.z
while not isinstance(parent, LiquidHandlerAbstract) and (res_z == 0) and parent is not None:
z += parent.get_size_z()
res_z = parent.location.z
parent = getattr(parent, "parent", None)
prcxi_x = (self.deck_x - x)*(1+self.x_increase) + self.x_offset + self.xy_coupling * (self.deck_y - y)
prcxi_y = (self.deck_y - y)*(1+self.y_increase) + self.y_offset
prcxi_z = self.deck_z - z
prcxi_x = min(max(0, prcxi_x),self.deck_x)
prcxi_y = min(max(0, prcxi_y),self.deck_y)
prcxi_z = min(max(0, prcxi_z),self.deck_z)
return Coordinate(prcxi_x, prcxi_y, prcxi_z)
def post_init(self, ros_node: BaseROS2DeviceNode):
super().post_init(ros_node)
@@ -888,18 +1087,32 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
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,
)
try:
return await super().aspirate(
resources,
vols,
use_channels,
flow_rates,
offsets,
liquid_height,
blow_out_air_volume,
spread,
**backend_kwargs,
)
except ValueError as e:
if "Resource is too small to space channels" in str(e) and spread != "custom":
return await super().aspirate(
resources,
vols,
use_channels,
flow_rates,
offsets,
liquid_height,
blow_out_air_volume,
spread="custom",
**backend_kwargs,
)
raise
async def drop_tips(
self,
@@ -923,17 +1136,33 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
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,
)
try:
return await super().dispense(
resources,
vols,
use_channels,
flow_rates,
offsets,
liquid_height,
blow_out_air_volume,
spread,
**backend_kwargs,
)
except ValueError as e:
if "Resource is too small to space channels" in str(e) and spread != "custom":
# 目标资源过小无法分布多通道时，退化为 custom（所有通道对准中心）
return await super().dispense(
resources,
vols,
use_channels,
flow_rates,
offsets,
liquid_height,
blow_out_air_volume,
"custom",
**backend_kwargs,
)
raise
async def discard_tips(
self,
@@ -953,6 +1182,11 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
async def shaker_action(self, time: int, module_no: int, amplitude: int, is_wait: bool):
return await self._unilabos_backend.shaker_action(time, module_no, amplitude, is_wait)
async def magnetic_action(self, time: int, module_no: int, height: int, is_wait: bool):
return await self._unilabos_backend.magnetic_action(time, module_no, height, is_wait)
async def shaking_incubation_action(self, time: int, module_no: int, amplitude: int, is_wait: bool, temperature: int):
return await self._unilabos_backend.shaking_incubation_action(time, module_no, amplitude, is_wait, temperature)
async def heater_action(self, temperature: float, time: int):
return await self._unilabos_backend.heater_action(temperature, time)
@@ -969,7 +1203,7 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
**backend_kwargs,
):
return await super().move_plate(
res = await super().move_plate(
plate,
to,
intermediate_locations,
@@ -981,6 +1215,12 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
target_plate_number=to,
**backend_kwargs,
)
plate.unassign()
to.assign_child_resource(plate, location=Coordinate(0, 0, 0))
ROS2DeviceNode.run_async_func(self._ros_node.update_resource, True, **{
"resources": [self.deck]
})
return res
class PRCXI9300Backend(LiquidHandlerBackend):
@@ -1005,6 +1245,7 @@ class PRCXI9300Backend(LiquidHandlerBackend):
def __init__(
self,
tablets_info: list[WorkTablets],
plate_positions: dict[int, Coordinate],
host: str = "127.0.0.1",
port: int = 9999,
timeout: float = 10.0,
@@ -1013,10 +1254,19 @@ class PRCXI9300Backend(LiquidHandlerBackend):
setup=True,
debug=False,
matrix_id="",
is_9320=False,
is_9320=False,
x_increase = 0,
y_increase = 0,
x_offset = 0,
y_offset = 0,
deck_z = 300,
deck_x = 0,
deck_y = 0,
xy_coupling = 0.0,
) -> None:
super().__init__()
self.tablets_info = tablets_info
self.plate_positions = plate_positions
self.matrix_id = matrix_id
self.api_client = PRCXI9300Api(host, port, timeout, axis, debug, is_9320)
self.host, self.port, self.timeout = host, port, timeout
@@ -1024,6 +1274,15 @@ class PRCXI9300Backend(LiquidHandlerBackend):
self._execute_setup = setup
self.debug = debug
self.axis = "Left"
self.x_increase = x_increase
self.y_increase = y_increase
self.xy_coupling = xy_coupling
self.x_offset = x_offset
self.y_offset = y_offset
self.deck_x = deck_x
self.deck_y = deck_y
self.deck_z = deck_z
self.tip_length = 0
async def shaker_action(self, time: int, module_no: int, amplitude: int, is_wait: bool):
step = self.api_client.shaker_action(
@@ -1035,6 +1294,27 @@ class PRCXI9300Backend(LiquidHandlerBackend):
self.steps_todo_list.append(step)
return step
async def shaking_incubation_action(self, time: int, module_no: int, amplitude: int, is_wait: bool, temperature: int):
step = self.api_client.shaking_incubation_action(
time=time,
module_no=module_no,
amplitude=amplitude,
is_wait=is_wait,
temperature=temperature,
)
self.steps_todo_list.append(step)
return step
async def magnetic_action(self, time: int, module_no: int, height: int, is_wait: bool):
step = self.api_client.magnetic_action(
time=time,
module_no=module_no,
height=height,
is_wait=is_wait,
)
self.steps_todo_list.append(step)
return step
async def pick_up_resource(self, pickup: ResourcePickup, **backend_kwargs):
resource = pickup.resource
@@ -1073,29 +1353,42 @@ class PRCXI9300Backend(LiquidHandlerBackend):
self._ros_node = ros_node
def create_protocol(self, protocol_name):
if protocol_name == "":
protocol_name = f"protocol_{time.time()}"
self.protocol_name = protocol_name
self.steps_todo_list = []
if not len(self.matrix_id):
self.matrix_id = str(uuid.uuid4())
material_list = self.api_client.get_all_materials()
material_dict = {material["uuid"]: material for material in material_list}
work_tablets = []
for num, material_id in self.tablets_info.items():
work_tablets.append({
"Number": num,
"Material": material_dict[material_id]
})
self.matrix_info = {
"MatrixId": self.matrix_id,
"MatrixName": self.matrix_id,
"WorkTablets": work_tablets,
}
# print(json.dumps(self.matrix_info, indent=2))
res = self.api_client.add_WorkTablet_Matrix(self.matrix_info)
if not res["Success"]:
self.matrix_id = ""
raise AssertionError(f"Failed to create matrix: {res.get('Message', 'Unknown error')}")
print(f"PRCXI9300Backend created matrix with ID: {self.matrix_info['MatrixId']}, result: {res}")
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,
solution_id = self.api_client.add_solution(
f"protocol_{run_time}", self.matrix_id, self.steps_todo_list
)
# 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))
@@ -1138,6 +1431,9 @@ class PRCXI9300Backend(LiquidHandlerBackend):
else:
await asyncio.sleep(1)
print("PRCXI9300 reset successfully.")
self.api_client.update_clamp_jaw_position(self.matrix_id, self.plate_positions)
except ConnectionRefusedError as e:
raise RuntimeError(
f"Failed to connect to PRCXI9300 API at {self.host}:{self.port}. "
@@ -1186,7 +1482,7 @@ class PRCXI9300Backend(LiquidHandlerBackend):
PlateNo = plate_indexes[0] + 1
hole_col = tip_columns[0] + 1
hole_row = 1
if self._num_channels == 1:
if self.num_channels != 8:
hole_row = tipspot_index % 8 + 1
step = self.api_client.Load(
@@ -1199,7 +1495,7 @@ class PRCXI9300Backend(LiquidHandlerBackend):
blending_times=0,
balance_height=0,
plate_or_hole=f"H{hole_col}-8,T{PlateNo}",
hole_numbers=f"{(hole_col - 1) * 8 + hole_row}" if self._num_channels == 1 else "1,2,3,4,5",
hole_numbers=f"{(hole_col - 1) * 8 + hole_row}" if self._num_channels != 8 else "1,2,3,4,5",
)
self.steps_todo_list.append(step)
@@ -1260,7 +1556,7 @@ class PRCXI9300Backend(LiquidHandlerBackend):
PlateNo = plate_indexes[0] + 1
hole_col = tip_columns[0] + 1
if self.channel_num == 1:
if self.num_channels != 8:
hole_row = tipspot_index % 8 + 1
step = self.api_client.UnLoad(
@@ -1318,7 +1614,7 @@ class PRCXI9300Backend(LiquidHandlerBackend):
PlateNo = plate_indexes[0] + 1
hole_col = tip_columns[0] + 1
hole_row = 1
if self.num_channels == 1:
if self.num_channels != 8:
hole_row = tipspot_index % 8 + 1
assert mix_time > 0
@@ -1375,7 +1671,7 @@ class PRCXI9300Backend(LiquidHandlerBackend):
PlateNo = plate_indexes[0] + 1
hole_col = tip_columns[0] + 1
hole_row = 1
if self.num_channels == 1:
if self.num_channels != 8:
hole_row = tipspot_index % 8 + 1
step = self.api_client.Imbibing(
@@ -1433,7 +1729,7 @@ class PRCXI9300Backend(LiquidHandlerBackend):
hole_col = tip_columns[0] + 1
hole_row = 1
if self.num_channels == 1:
if self.num_channels != 8:
hole_row = tipspot_index % 8 + 1
step = self.api_client.Tapping(
@@ -1559,10 +1855,10 @@ class PRCXI9300Api:
start = False
while not success:
status = self.step_state_list()
if len(status) == 1:
start = True
if status is None:
break
if len(status) == 1:
start = True
if len(status) == 0:
break
if status[-1]["State"] == 2 and start:
@@ -1641,6 +1937,13 @@ class PRCXI9300Api:
"""GetWorkTabletMatrixById"""
return self.call("IMatrix", "GetWorkTabletMatrixById", [matrix_id])
def update_clamp_jaw_position(self, target_matrix_id: str, plate_positions: List[Dict[str, Any]]):
position_params = {
"MatrixId": target_matrix_id,
"WorkTablets": plate_positions
}
return self.call("IMatrix", "UpdateClampJawPosition", [position_params])
def add_WorkTablet_Matrix(self, matrix: MatrixInfo):
return self.call("IMatrix", "AddWorkTabletMatrix2" if self.is_9320 else "AddWorkTabletMatrix", [matrix])
@@ -1883,6 +2186,26 @@ class PRCXI9300Api:
"AssistFun4": is_wait,
}
def shaking_incubation_action(self, time: int, module_no: int, amplitude: int, is_wait: bool, temperature: int):
return {
"StepAxis": "Left",
"Function": "Shaking_Incubation",
"AssistFun1": time,
"AssistFun2": module_no,
"AssistFun3": amplitude,
"AssistFun4": is_wait,
"AssistFun5": temperature,
}
def magnetic_action(self, time: int, module_no: int, height: int, is_wait: bool):
return {
"StepAxis": "Left",
"Function": "Magnetic",
"AssistFun1": time,
"AssistFun2": module_no,
"AssistFun3": height,
"AssistFun4": is_wait,
}
class DefaultLayout:
@@ -1914,82 +2237,82 @@ class DefaultLayout:
{
"Number": 1,
"Code": "T1",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"},
},
{
"Number": 2,
"Code": "T2",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"},
},
{
"Number": 3,
"Code": "T3",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"},
},
{
"Number": 4,
"Code": "T4",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"},
},
{
"Number": 5,
"Code": "T5",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"},
},
{
"Number": 6,
"Code": "T6",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"},
},
{
"Number": 7,
"Code": "T7",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"},
},
{
"Number": 8,
"Code": "T8",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"},
},
{
"Number": 9,
"Code": "T9",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"},
},
{
"Number": 10,
"Code": "T10",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"},
},
{
"Number": 11,
"Code": "T11",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"},
},
{
"Number": 12,
"Code": "T12",
"Material": {"uuid": "730067cf07ae43849ddf4034299030e9", "materialEnum": 0},
"Material": {"uuid": "730067cf07ae43849ddf4034299030e9"},
}, # 这个设置成废液槽，用储液槽表示
{
"Number": 13,
"Code": "T13",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"},
},
{
"Number": 14,
"Code": "T14",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"},
},
{
"Number": 15,
"Code": "T15",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"},
},
{
"Number": 16,
"Code": "T16",
"Material": {"uuid": "730067cf07ae43849ddf4034299030e9", "materialEnum": 0},
"Material": {"uuid": "730067cf07ae43849ddf4034299030e9"},
}, # 这个设置成垃圾桶，用储液槽表示
],
}