open_pub/Uni-Lab-OS
1
0
Fork 0
mirror of https://github.com/dptech-corp/Uni-Lab-OS.git synced 2025-12-19 05:51:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge dev branch: Add battery resources, bioyond_cell device registry, and fix file path resolution

Browse Source
This commit is contained in:
dijkstra402
2025-12-18 11:11:59 +08:00
parent 5072f00836 143104e9e3
commit 9f8b5eaf71
306 changed files with 47518 additions and 4826 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
unilabos/resources/battery/__init__.py
View File

@@ -0,0 +1,4 @@
"""Battery-related resource classes for coin cell assembly"""

81
unilabos/resources/battery/bottle_carriers.py
View File

@@ -1,56 +1,45 @@
from pylabrobot.resources import create_homogeneous_resources, Coordinate, ResourceHolder, create_ordered_items_2d
"""
瓶架类定义 - 用于纽扣电池组装工作站
Bottle Carrier Resource Classes
"""
from unilabos.resources.itemized_carrier import Bottle, BottleCarrier
from unilabos.resources.bioyond.YB_bottles import (
YB_pei_ye_xiao_Bottle,
)
# 命名约定：试剂瓶-Bottle，烧杯-Beaker，烧瓶-Flask，小瓶-Vial
from __future__ import annotations
from pylabrobot.resources import ResourceHolder
from pylabrobot.resources.utils import create_ordered_items_2d
from unilabos.resources.itemized_carrier import ItemizedCarrier
def YIHUA_Electrolyte_12VialCarrier(name: str) -> BottleCarrier:
"""12瓶载架 - 2x6布局"""
# 载架尺寸 (mm)
carrier_size_x = 120.0
carrier_size_y = 250.0
carrier_size_z = 50.0
# 瓶位尺寸
bottle_diameter = 35.0
bottle_spacing_x = 35.0 # X方向间距
bottle_spacing_y = 35.0 # Y方向间距
# 计算起始位置 (居中排列)
start_x = (carrier_size_x - (2 - 1) * bottle_spacing_x - bottle_diameter) / 2
start_y = (carrier_size_y - (6 - 1) * bottle_spacing_y - bottle_diameter) / 2
def YIHUA_Electrolyte_12VialCarrier(name: str) -> ItemizedCarrier:
"""依华电解液12瓶架 - 3x4布局
Args:
name: 瓶架名称
Returns:
ItemizedCarrier: 包含12个瓶位的瓶架
"""
sites = create_ordered_items_2d(
klass=ResourceHolder,
num_items_x=2,
num_items_y=6,
dx=start_x,
dy=start_y,
num_items_x=4,
num_items_y=3,
dx=10.0,
dy=10.0,
dz=5.0,
item_dx=bottle_spacing_x,
item_dy=bottle_spacing_y,
size_x=bottle_diameter,
size_y=bottle_diameter,
size_z=carrier_size_z,
item_dx=70.0,
item_dy=26.67,
size_x=60.0,
size_y=20.0,
size_z=70.0,
)
for k, v in sites.items():
v.name = f"{name}_{v.name}"
carrier = BottleCarrier(
return ItemizedCarrier(
name=name,
size_x=carrier_size_x,
size_y=carrier_size_y,
size_z=carrier_size_z,
size_x=300.0,
size_y=100.0,
size_z=80.0,
num_items_x=4,
num_items_y=3,
sites=sites,
model="Electrolyte_12VialCarrier",
category="bottle_carrier",
)
carrier.num_items_x = 2
carrier.num_items_y = 6
carrier.num_items_z = 1
for i in range(12):
carrier[i] = YB_pei_ye_xiao_Bottle(f"{name}_vial_{i+1}")
return carrier

198
unilabos/resources/battery/electrode_sheet.py
View File

@@ -1,44 +1,35 @@
from typing import Any, Dict, Optional, TypedDict
"""
电极片类定义
Electrode Sheet Resource Classes
"""
from pylabrobot.resources import Resource as ResourcePLR
from pylabrobot.resources import Container
from __future__ import annotations
from typing import Any, Dict, Optional
from pylabrobot.resources.resource import Resource
electrode_colors = {
"PositiveCan": "#ff0000",
"PositiveElectrode": "#cc3333",
"NegativeCan": "#000000",
"NegativeElectrode": "#666666",
"SpringWasher": "#8b7355",
"FlatWasher": "a9a9a9",
"AluminumFoil": "#ffcccc",
"Battery": "#00ff00",
}
class ElectrodeSheetState(TypedDict):
mass: float # 质量 (g)
material_type: str # 材料类型（铜、铝、不锈钢、弹簧钢等）
color: str # 材料类型对应的颜色
class ElectrodeSheet(ResourcePLR):
"""极片类 - 包含正负极片、隔膜、弹片、垫片、铝箔等所有片状材料"""
class ElectrodeSheet(Resource):
"""电极片类 - 用于纽扣电池组装"""
def __init__(
self,
name: str = "极片",
size_x=10,
size_y=10,
size_z=10,
name: str,
size_x: float = 12.0,
size_y: float = 12.0,
size_z: float = 0.1,
category: str = "electrode_sheet",
model: Optional[str] = None,
electrode_type: str = "anode", # "anode" 负极, "cathode" 正极, "separator" 隔膜
**kwargs
):
"""初始化极片
"""初始化电极片
Args:
name: 极片名称
name: 电极片名称
size_x: X方向尺寸 (mm)
size_y: Y方向尺寸 (mm)
size_z: Z方向尺寸/厚度 (mm)
category: 类别
model: 型号
electrode_type: 电极类型
"""
super().__init__(
name=name,
@@ -46,134 +37,31 @@ class ElectrodeSheet(ResourcePLR):
size_y=size_y,
size_z=size_z,
category=category,
model=model,
**kwargs
)
self._unilabos_state: ElectrodeSheetState = ElectrodeSheetState(
diameter=14,
thickness=0.1,
mass=0.5,
material_type="copper",
info=None
)
# TODO: 这个还要不要？给self._unilabos_state赋值的？
self._electrode_type = electrode_type
self._unilabos_state: Dict[str, Any] = {
"electrode_type": electrode_type,
"material": "",
"thickness": size_z,
}
@property
def electrode_type(self) -> str:
"""获取电极类型"""
return self._electrode_type
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]]:
"""格式不变"""
if isinstance(state, dict):
self._unilabos_state.update(state)
def serialize_state(self) -> Dict[str, Any]:
"""序列化状态"""
data = super().serialize_state()
data.update(self._unilabos_state) # Container自身的信息，云端物料将保存这一data，本地也通过这里的data进行读写，当前类用来表示这个物料的长宽高大小的属性，而data（state用来表示物料的内容，细节等）
data.update(self._unilabos_state)
return data
def PositiveCan(name: str) -> ElectrodeSheet:
"""创建正极壳"""
sheet = ElectrodeSheet(name=name, size_x=12, size_y=12, size_z=3.0, model="PositiveCan")
sheet.load_state({"material_type": "aluminum", "color": electrode_colors["PositiveCan"]})
return sheet
def PositiveElectrode(name: str) -> ElectrodeSheet:
"""创建正极片"""
sheet = ElectrodeSheet(name=name, size_x=10, size_y=10, size_z=0.1, model="PositiveElectrode")
sheet.load_state({"material_type": "positive_electrode", "color": electrode_colors["PositiveElectrode"]})
return sheet
def NegativeCan(name: str) -> ElectrodeSheet:
"""创建负极壳"""
sheet = ElectrodeSheet(name=name, size_x=12, size_y=12, size_z=2.0, model="NegativeCan")
sheet.load_state({"material_type": "steel", "color": electrode_colors["NegativeCan"]})
return sheet
def NegativeElectrode(name: str) -> ElectrodeSheet:
"""创建负极片"""
sheet = ElectrodeSheet(name=name, size_x=10, size_y=10, size_z=0.1, model="NegativeElectrode")
sheet.load_state({"material_type": "negative_electrode", "color": electrode_colors["NegativeElectrode"]})
return sheet
def SpringWasher(name: str) -> ElectrodeSheet:
"""创建弹片"""
sheet = ElectrodeSheet(name=name, size_x=10, size_y=10, size_z=0.5, model="SpringWasher")
sheet.load_state({"material_type": "spring_steel", "color": electrode_colors["SpringWasher"]})
return sheet
def FlatWasher(name: str) -> ElectrodeSheet:
"""创建垫片"""
sheet = ElectrodeSheet(name=name, size_x=10, size_y=10, size_z=0.2, model="FlatWasher")
sheet.load_state({"material_type": "steel", "color": electrode_colors["FlatWasher"]})
return sheet
def AluminumFoil(name: str) -> ElectrodeSheet:
"""创建铝箔"""
sheet = ElectrodeSheet(name=name, size_x=10, size_y=10, size_z=0.05, model="AluminumFoil")
sheet.load_state({"material_type": "aluminum", "color": electrode_colors["AluminumFoil"]})
return sheet
class BatteryState(TypedDict):
color: str # 材料类型对应的颜色
electrolyte_name: str
data_electrolyte_code: str
open_circuit_voltage: float
assembly_pressure: float
electrolyte_volume: float
info: Optional[str] # 附加信息
class Battery(Container):
"""电池类 - 包含组装好的电池"""
def __init__(
self,
name: str = "电池",
size_x=12,
size_y=12,
size_z=6,
category: str = "battery",
model: Optional[str] = None,
):
"""初始化电池
Args:
name: 电池名称
category: 类别
model: 型号
"""
super().__init__(
name=name,
size_x=size_x,
size_y=size_y,
size_z=size_z,
category=category,
model=model,
)
self._unilabos_state: BatteryState = BatteryState(
color=electrode_colors["Battery"],
electrolyte_name="无",
data_electrolyte_code="",
open_circuit_voltage=0.0,
assembly_pressure=0.0,
electrolyte_volume=0.0,
info=None
)
def load_state(self, state: Dict[str, Any]) -> None:
"""格式不变"""
super().load_state(state)
self._unilabos_state = state
#序列化
def serialize_state(self) -> Dict[str, Dict[str, Any]]:
"""格式不变"""
data = super().serialize_state()
data.update(self._unilabos_state) # Container自身的信息，云端物料将保存这一data，本地也通过这里的data进行读写，当前类用来表示这个物料的长宽高大小的属性，而data（state用来表示物料的内容，细节等）
return data

466
unilabos/resources/battery/magazine.py
View File

@@ -1,344 +1,152 @@
from typing import Dict, List, Optional, OrderedDict, Union, Callable
import math
"""
弹夹架类定义 - 用于纽扣电池组装工作站
Magazine Holder Resource Classes
"""
from __future__ import annotations
from typing import List, Optional
from pylabrobot.resources.coordinate import Coordinate
from pylabrobot.resources import Resource, ResourceStack, ItemizedResource
from pylabrobot.resources.carrier import create_homogeneous_resources
from unilabos.resources.battery.electrode_sheet import (
PositiveCan, PositiveElectrode,
NegativeCan, NegativeElectrode,
SpringWasher, FlatWasher,
AluminumFoil,
Battery
)
from pylabrobot.resources import ResourceHolder
from pylabrobot.resources.utils import create_ordered_items_2d
from unilabos.resources.itemized_carrier import ItemizedCarrier
class Magazine(ResourceStack):
"""子弹夹洞位类"""
def __init__(
self,
name: str,
direction: str = 'z',
resources: Optional[List[Resource]] = None,
max_sheets: int = 100,
**kwargs
):
"""初始化子弹夹洞位
Args:
name: 洞位名称
direction: 堆叠方向
resources: 资源列表
max_sheets: 最大极片数量
"""
super().__init__(
name=name,
direction=direction,
resources=resources,
)
self.max_sheets = max_sheets
@property
def size_x(self) -> float:
return self.get_size_x()
@property
def size_y(self) -> float:
return self.get_size_y()
@property
def size_z(self) -> float:
return self.get_size_z()
def serialize(self) -> dict:
return {
**super().serialize(),
"size_x": self.size_x or 10.0,
"size_y": self.size_y or 10.0,
"size_z": self.size_z or 10.0,
"max_sheets": self.max_sheets,
}
class MagazineHolder(ItemizedResource):
"""子弹夹类 - 有多个洞位，每个洞位放多个极片"""
def __init__(
self,
name: str,
size_x: float,
size_y: float,
size_z: float,
ordered_items: Optional[Dict[str, Magazine]] = None,
ordering: Optional[OrderedDict[str, str]] = None,
hole_diameter: float = 14.0,
hole_depth: float = 10.0,
max_sheets_per_hole: int = 100,
cross_section_type: str = "circle",
category: str = "magazine_holder",
model: Optional[str] = None,
):
"""初始化子弹夹
Args:
name: 子弹夹名称
size_x: 长度 (mm)
size_y: 宽度 (mm)
size_z: 高度 (mm)
hole_diameter: 洞直径 (mm)
hole_depth: 洞深度 (mm)
max_sheets_per_hole: 每个洞位最大极片数量
category: 类别
model: 型号
"""
super().__init__(
name=name,
size_x=size_x,
size_y=size_y,
size_z=size_z,
ordered_items=ordered_items,
ordering=ordering,
category=category,
model=model,
)
# 保存洞位的直径和深度
self.hole_diameter = hole_diameter
self.hole_depth = hole_depth
self.max_sheets_per_hole = max_sheets_per_hole
self.cross_section_type = cross_section_type
def serialize(self) -> dict:
return {
**super().serialize(),
"hole_diameter": self.hole_diameter,
"hole_depth": self.hole_depth,
"max_sheets_per_hole": self.max_sheets_per_hole,
"cross_section_type": self.cross_section_type,
}
def magazine_factory(
name: str,
size_x: float,
size_y: float,
size_z: float,
locations: List[Coordinate],
klasses: Optional[List[Callable[[str], str]]] = None,
hole_diameter: float = 14.0,
hole_depth: float = 10.0,
max_sheets_per_hole: int = 100,
category: str = "magazine_holder",
model: Optional[str] = None,
) -> 'MagazineHolder':
"""工厂函数：创建子弹夹
def MagazineHolder_4_Cathode(name: str) -> ItemizedCarrier:
"""正极&铝箔弹夹 - 4个洞位 (2x2布局)
Args:
name: 子弹夹名称
size_x: 长度 (mm)
size_y: 宽度 (mm)
size_z: 高度 (mm)
locations: 洞位坐标列表
klasses: 每个洞位中极片的类列表
hole_diameter: 洞直径 (mm)
hole_depth: 洞深度 (mm)
max_sheets_per_hole: 每个洞位最大极片数量
category: 类别
model: 型号
name: 弹夹名称
Returns:
ItemizedCarrier: 包含4个槽位的弹夹架
"""
for loc in locations:
loc.x -= hole_diameter / 2
loc.y -= hole_diameter / 2
# 创建洞位
_sites = create_homogeneous_resources(
klass=Magazine,
locations=locations,
resource_size_x=hole_diameter,
resource_size_y=hole_diameter,
name_prefix=name,
max_sheets=max_sheets_per_hole,
sites = create_ordered_items_2d(
klass=ResourceHolder,
num_items_x=2,
num_items_y=2,
dx=10.0,
dy=10.0,
dz=0.0,
item_dx=50.0,
item_dy=30.0,
size_x=40.0,
size_y=25.0,
size_z=40.0,
)
# 生成编号键
keys = [f"A{i+1}" for i in range(len(locations))]
sites = dict(zip(keys, _sites.values()))
return ItemizedCarrier(
name=name,
size_x=120.0,
size_y=80.0,
size_z=50.0,
num_items_x=2,
num_items_y=2,
sites=sites,
category="magazine_holder",
)
def MagazineHolder_6_Cathode(name: str) -> ItemizedCarrier:
"""正极壳&平垫片弹夹 - 6个洞位 (2x3布局)
holder = MagazineHolder(
name=name,
size_x=size_x,
size_y=size_y,
size_z=size_z,
ordered_items=sites,
hole_diameter=hole_diameter,
hole_depth=hole_depth,
max_sheets_per_hole=max_sheets_per_hole,
category=category,
model=model,
Args:
name: 弹夹名称
Returns:
ItemizedCarrier: 包含6个槽位的弹夹架
"""
sites = create_ordered_items_2d(
klass=ResourceHolder,
num_items_x=3,
num_items_y=2,
dx=10.0,
dy=10.0,
dz=0.0,
item_dx=40.0,
item_dy=30.0,
size_x=35.0,
size_y=25.0,
size_z=40.0,
)
if klasses is not None:
for i, klass in enumerate(klasses):
hole_key = keys[i]
hole = holder.children[i]
for j in reversed(range(max_sheets_per_hole)):
item_name = f"{hole_key}_sheet{j+1}"
item = klass(name=item_name)
hole.assign_child_resource(item)
return holder
def MagazineHolder_6_Cathode(
name: str,
size_x: float = 80.0,
size_y: float = 80.0,
size_z: float = 40.0,
hole_diameter: float = 14.0,
hole_depth: float = 10.0,
hole_spacing: float = 20.0,
max_sheets_per_hole: int = 100,
) -> MagazineHolder:
"""创建6孔子弹夹 - 六边形排布"""
center_x = size_x / 2
center_y = size_y / 2
locations = []
# 周围6个孔，按六边形排布
for i in range(6):
angle = i * 60 * math.pi / 180 # 每60度一个孔
x = center_x + hole_spacing * math.cos(angle)
y = center_y + hole_spacing * math.sin(angle)
locations.append(Coordinate(x, y, size_z - hole_depth))
return magazine_factory(
name=name,
size_x=size_x,
size_y=size_y,
size_z=size_z,
locations=locations,
klasses=[FlatWasher, PositiveCan, PositiveCan, FlatWasher, PositiveCan, PositiveCan],
hole_diameter=hole_diameter,
hole_depth=hole_depth,
max_sheets_per_hole=max_sheets_per_hole,
category="magazine_holder",
model="MagazineHolder_6_Cathode",
)
def MagazineHolder_6_Anode(
name: str,
size_x: float = 80.0,
size_y: float = 80.0,
size_z: float = 40.0,
hole_diameter: float = 14.0,
hole_depth: float = 10.0,
hole_spacing: float = 20.0,
max_sheets_per_hole: int = 100,
) -> MagazineHolder:
"""创建6孔子弹夹 - 六边形排布"""
center_x = size_x / 2
center_y = size_y / 2
locations = []
# 周围6个孔，按六边形排布
for i in range(6):
angle = i * 60 * math.pi / 180 # 每60度一个孔
x = center_x + hole_spacing * math.cos(angle)
y = center_y + hole_spacing * math.sin(angle)
locations.append(Coordinate(x, y, size_z - hole_depth))
return magazine_factory(
name=name,
size_x=size_x,
size_y=size_y,
size_z=size_z,
locations=locations,
klasses=[SpringWasher, NegativeCan, NegativeCan, SpringWasher, NegativeCan, NegativeCan],
hole_diameter=hole_diameter,
hole_depth=hole_depth,
max_sheets_per_hole=max_sheets_per_hole,
category="magazine_holder",
model="MagazineHolder_6_Anode",
)
def MagazineHolder_6_Battery(
name: str,
size_x: float = 80.0,
size_y: float = 80.0,
size_z: float = 40.0,
hole_diameter: float = 14.0,
hole_depth: float = 10.0,
hole_spacing: float = 20.0,
max_sheets_per_hole: int = 100,
) -> MagazineHolder:
"""创建6孔子弹夹 - 六边形排布"""
center_x = size_x / 2
center_y = size_y / 2
locations = []
# 周围6个孔，按六边形排布
for i in range(6):
angle = i * 60 * math.pi / 180 # 每60度一个孔
x = center_x + hole_spacing * math.cos(angle)
y = center_y + hole_spacing * math.sin(angle)
locations.append(Coordinate(x, y, size_z - hole_depth))
return magazine_factory(
name=name,
size_x=size_x,
size_y=size_y,
size_z=size_z,
locations=locations,
klasses=None, # 初始化时，不放入装好的电池
hole_diameter=hole_diameter,
hole_depth=hole_depth,
max_sheets_per_hole=max_sheets_per_hole,
category="magazine_holder",
model="MagazineHolder_6_Battery",
)
def MagazineHolder_4_Cathode(
name: str,
) -> MagazineHolder:
"""创建4孔子弹夹 - 正方形四角排布"""
size_x: float = 80.0
size_y: float = 80.0
size_z: float = 10.0
hole_diameter: float = 14.0
hole_depth: float = 10.0
hole_spacing: float = 25.0
max_sheets_per_hole: int = 100
# 计算4个洞位的坐标（正方形四角排布）
center_x = size_x / 2
center_y = size_y / 2
offset = hole_spacing / 2
locations = [
Coordinate(center_x - offset, center_y - offset, size_z - hole_depth), # 左下
Coordinate(center_x + offset, center_y - offset, size_z - hole_depth), # 右下
Coordinate(center_x - offset, center_y + offset, size_z - hole_depth), # 左上
Coordinate(center_x + offset, center_y + offset, size_z - hole_depth), # 右上
]
return magazine_factory(
return ItemizedCarrier(
name=name,
size_x=size_x,
size_y=size_y,
size_z=size_z,
locations=locations,
klasses=[AluminumFoil, PositiveElectrode, PositiveElectrode, PositiveElectrode],
hole_diameter=hole_diameter,
hole_depth=hole_depth,
max_sheets_per_hole=max_sheets_per_hole,
size_x=150.0,
size_y=80.0,
size_z=50.0,
num_items_x=3,
num_items_y=2,
sites=sites,
category="magazine_holder",
model="MagazineHolder_4_Cathode",
)
def MagazineHolder_6_Anode(name: str) -> ItemizedCarrier:
"""负极壳&弹垫片弹夹 - 6个洞位 (2x3布局)
Args:
name: 弹夹名称
Returns:
ItemizedCarrier: 包含6个槽位的弹夹架
"""
sites = create_ordered_items_2d(
klass=ResourceHolder,
num_items_x=3,
num_items_y=2,
dx=10.0,
dy=10.0,
dz=0.0,
item_dx=40.0,
item_dy=30.0,
size_x=35.0,
size_y=25.0,
size_z=40.0,
)
return ItemizedCarrier(
name=name,
size_x=150.0,
size_y=80.0,
size_z=50.0,
num_items_x=3,
num_items_y=2,
sites=sites,
category="magazine_holder",
)
def MagazineHolder_6_Battery(name: str) -> ItemizedCarrier:
"""成品弹夹 - 6个洞位 (3x2布局)
Args:
name: 弹夹名称
Returns:
ItemizedCarrier: 包含6个槽位的弹夹架
"""
sites = create_ordered_items_2d(
klass=ResourceHolder,
num_items_x=3,
num_items_y=2,
dx=10.0,
dy=10.0,
dz=0.0,
item_dx=33.0,
item_dy=40.0,
size_x=30.0,
size_y=35.0,
size_z=40.0,
)
return ItemizedCarrier(
name=name,
size_x=120.0,
size_y=100.0,
size_z=50.0,
num_items_x=3,
num_items_y=2,
sites=sites,
category="magazine_holder",
)

324
unilabos/resources/bioyond/bottle_carriers.py Normal file
View File

@@ -0,0 +1,324 @@
from pylabrobot.resources import create_homogeneous_resources, Coordinate, ResourceHolder, create_ordered_items_2d
from unilabos.resources.itemized_carrier import BottleCarrier
from unilabos.resources.bioyond.bottles import (
BIOYOND_PolymerStation_Solid_Stock,
BIOYOND_PolymerStation_Solid_Vial,
BIOYOND_PolymerStation_Liquid_Vial,
BIOYOND_PolymerStation_Solution_Beaker,
BIOYOND_PolymerStation_Reagent_Bottle,
BIOYOND_PolymerStation_Flask,
)
# 命名约定：试剂瓶-Bottle，烧杯-Beaker，烧瓶-Flask,小瓶-Vial
# ============================================================================
# 聚合站（PolymerStation）载体定义（统一入口）
# ============================================================================
def BIOYOND_PolymerStation_6StockCarrier(name: str) -> BottleCarrier:
"""聚合站-6孔样品板 - 2x3布局
参数:
- name: 载架名称前缀
说明:
- 统一站点命名为 PolymerStation，使用 PolymerStation 的 Vial 资源类
- A行（PLR y=0，对应 Bioyond 位置A01~A03）使用 Liquid_Vial（10% 分装小瓶）
- B行（PLR y=1，对应 Bioyond 位置B01~B03）使用 Solid_Vial（90% 分装小瓶）
"""
# 载架尺寸 (mm)
carrier_size_x = 127.8
carrier_size_y = 85.5
carrier_size_z = 50.0
# 瓶位尺寸
bottle_diameter = 20.0
bottle_spacing_x = 42.0 # X方向间距
bottle_spacing_y = 35.0 # Y方向间距
# 计算起始位置 (居中排列)
start_x = (carrier_size_x - (3 - 1) * bottle_spacing_x - bottle_diameter) / 2
start_y = (carrier_size_y - (2 - 1) * bottle_spacing_y - bottle_diameter) / 2
sites = create_ordered_items_2d(
klass=ResourceHolder,
num_items_x=3,
num_items_y=2,
dx=start_x,
dy=start_y,
dz=5.0,
item_dx=bottle_spacing_x,
item_dy=bottle_spacing_y,
size_x=bottle_diameter,
size_y=bottle_diameter,
size_z=carrier_size_z,
)
for k, v in sites.items():
v.name = f"{name}_{v.name}"
carrier = BottleCarrier(
name=name,
size_x=carrier_size_x,
size_y=carrier_size_y,
size_z=carrier_size_z,
sites=sites,
model="BIOYOND_PolymerStation_6StockCarrier",
)
carrier.num_items_x = 3
carrier.num_items_y = 2
carrier.num_items_z = 1
# 布局说明：
# - num_items_x=3, num_items_y=2 表示 3列×2行
# - create_ordered_items_2d 按先y后x的顺序创建(列优先)
# - 索引顺序: 0=A1(x=0,y=0), 1=B1(x=0,y=1), 2=A2(x=1,y=0), 3=B2(x=1,y=1), 4=A3(x=2,y=0), 5=B3(x=2,y=1)
#
# Bioyond坐标映射: PLR(x,y) → Bioyond(y+1,x+1)
# - A行(PLR y=0) → Bioyond x=1 → 10%分装小瓶
# - B行(PLR y=1) → Bioyond x=2 → 90%分装小瓶
ordering = ["A1", "B1", "A2", "B2", "A3", "B3"]
for col in range(3): # 3列
for row in range(2): # 2行
idx = col * 2 + row # 计算索引: 列优先顺序
if row == 0: # A行 (PLR y=0 → Bioyond x=1)
carrier[idx] = BIOYOND_PolymerStation_Liquid_Vial(f"{ordering[idx]}")
else: # B行 (PLR y=1 → Bioyond x=2)
carrier[idx] = BIOYOND_PolymerStation_Solid_Vial(f"{ordering[idx]}")
return carrier
def BIOYOND_PolymerStation_8StockCarrier(name: str) -> BottleCarrier:
"""聚合站-8孔样品板 - 2x4布局
参数:
- name: 载架名称前缀
说明:
- 统一站点命名为 PolymerStation，使用 PolymerStation 的 Solid_Stock 资源类
"""
# 载架尺寸 (mm)
carrier_size_x = 128.0
carrier_size_y = 85.5
carrier_size_z = 50.0
# 瓶位尺寸
bottle_diameter = 20.0
bottle_spacing_x = 30.0 # X方向间距
bottle_spacing_y = 35.0 # Y方向间距
# 计算起始位置 (居中排列)
start_x = (carrier_size_x - (4 - 1) * bottle_spacing_x - bottle_diameter) / 2
start_y = (carrier_size_y - (2 - 1) * bottle_spacing_y - bottle_diameter) / 2
sites = create_ordered_items_2d(
klass=ResourceHolder,
num_items_x=4,
num_items_y=2,
dx=start_x,
dy=start_y,
dz=5.0,
item_dx=bottle_spacing_x,
item_dy=bottle_spacing_y,
size_x=bottle_diameter,
size_y=bottle_diameter,
size_z=carrier_size_z,
)
for k, v in sites.items():
v.name = f"{name}_{v.name}"
carrier = BottleCarrier(
name=name,
size_x=carrier_size_x,
size_y=carrier_size_y,
size_z=carrier_size_z,
sites=sites,
model="BIOYOND_PolymerStation_8StockCarrier",
)
carrier.num_items_x = 4
carrier.num_items_y = 2
carrier.num_items_z = 1
ordering = ["A1", "B1", "A2", "B2", "A3", "B3", "A4", "B4"]
for i in range(8):
carrier[i] = BIOYOND_PolymerStation_Solid_Stock(f"{name}_vial_{ordering[i]}")
return carrier
def BIOYOND_PolymerStation_1BottleCarrier(name: str) -> BottleCarrier:
"""聚合站-单试剂瓶载架
参数:
- name: 载架名称前缀
"""
# 载架尺寸 (mm)
carrier_size_x = 127.8
carrier_size_y = 85.5
carrier_size_z = 20.0
# 烧杯/试剂瓶占位尺寸（使用圆形占位）
beaker_diameter = 60.0
# 计算中央位置
center_x = (carrier_size_x - beaker_diameter) / 2
center_y = (carrier_size_y - beaker_diameter) / 2
center_z = 5.0
carrier = BottleCarrier(
name=name,
size_x=carrier_size_x,
size_y=carrier_size_y,
size_z=carrier_size_z,
sites=create_homogeneous_resources(
klass=ResourceHolder,
locations=[Coordinate(center_x, center_y, center_z)],
resource_size_x=beaker_diameter,
resource_size_y=beaker_diameter,
name_prefix=name,
),
model="BIOYOND_PolymerStation_1BottleCarrier",
)
carrier.num_items_x = 1
carrier.num_items_y = 1
carrier.num_items_z = 1
# 统一后缀采用 "flask_1" 命名（可按需调整）
carrier[0] = BIOYOND_PolymerStation_Reagent_Bottle(f"{name}_flask_1")
return carrier
def BIOYOND_PolymerStation_1FlaskCarrier(name: str) -> BottleCarrier:
"""聚合站-单烧杯载架
说明:
- 使用 BIOYOND_PolymerStation_Flask 资源类
- 载架命名与 model 统一为 PolymerStation
"""
# 载架尺寸 (mm)
carrier_size_x = 127.8
carrier_size_y = 85.5
carrier_size_z = 20.0
# 烧杯尺寸
beaker_diameter = 60.0
# 计算中央位置
center_x = (carrier_size_x - beaker_diameter) / 2
center_y = (carrier_size_y - beaker_diameter) / 2
center_z = 5.0
carrier = BottleCarrier(
name=name,
size_x=carrier_size_x,
size_y=carrier_size_y,
size_z=carrier_size_z,
sites=create_homogeneous_resources(
klass=ResourceHolder,
locations=[Coordinate(center_x, center_y, center_z)],
resource_size_x=beaker_diameter,
resource_size_y=beaker_diameter,
name_prefix=name,
),
model="BIOYOND_PolymerStation_1FlaskCarrier",
)
carrier.num_items_x = 1
carrier.num_items_y = 1
carrier.num_items_z = 1
carrier[0] = BIOYOND_PolymerStation_Flask(f"{name}_flask_1")
return carrier
# ============================================================================
# 其他载体定义
# ============================================================================
def BIOYOND_Electrolyte_6VialCarrier(name: str) -> BottleCarrier:
"""6瓶载架 - 2x3布局"""
# 载架尺寸 (mm)
carrier_size_x = 127.8
carrier_size_y = 85.5
carrier_size_z = 50.0
# 瓶位尺寸
bottle_diameter = 30.0
bottle_spacing_x = 42.0 # X方向间距
bottle_spacing_y = 35.0 # Y方向间距
# 计算起始位置 (居中排列)
start_x = (carrier_size_x - (3 - 1) * bottle_spacing_x - bottle_diameter) / 2
start_y = (carrier_size_y - (2 - 1) * bottle_spacing_y - bottle_diameter) / 2
sites = create_ordered_items_2d(
klass=ResourceHolder,
num_items_x=3,
num_items_y=2,
dx=start_x,
dy=start_y,
dz=5.0,
item_dx=bottle_spacing_x,
item_dy=bottle_spacing_y,
size_x=bottle_diameter,
size_y=bottle_diameter,
size_z=carrier_size_z,
)
for k, v in sites.items():
v.name = f"{name}_{v.name}"
carrier = BottleCarrier(
name=name,
size_x=carrier_size_x,
size_y=carrier_size_y,
size_z=carrier_size_z,
sites=sites,
model="BIOYOND_Electrolyte_6VialCarrier",
)
carrier.num_items_x = 3
carrier.num_items_y = 2
carrier.num_items_z = 1
for i in range(6):
carrier[i] = BIOYOND_PolymerStation_Solid_Vial(f"{name}_vial_{i+1}")
return carrier
def BIOYOND_Electrolyte_1BottleCarrier(name: str) -> BottleCarrier:
"""1瓶载架 - 单个中央位置"""
# 载架尺寸 (mm)
carrier_size_x = 127.8
carrier_size_y = 85.5
carrier_size_z = 100.0
# 烧杯尺寸
beaker_diameter = 80.0
# 计算中央位置
center_x = (carrier_size_x - beaker_diameter) / 2
center_y = (carrier_size_y - beaker_diameter) / 2
center_z = 5.0
carrier = BottleCarrier(
name=name,
size_x=carrier_size_x,
size_y=carrier_size_y,
size_z=carrier_size_z,
sites=create_homogeneous_resources(
klass=ResourceHolder,
locations=[Coordinate(center_x, center_y, center_z)],
resource_size_x=beaker_diameter,
resource_size_y=beaker_diameter,
name_prefix=name,
),
model="BIOYOND_Electrolyte_1BottleCarrier",
)
carrier.num_items_x = 1
carrier.num_items_y = 1
carrier.num_items_z = 1
carrier[0] = BIOYOND_PolymerStation_Solution_Beaker(f"{name}_beaker_1")
return carrier

195
unilabos/resources/bioyond/bottles.py Normal file
View File

@@ -0,0 +1,195 @@
from unilabos.resources.itemized_carrier import Bottle
def BIOYOND_PolymerStation_Solid_Stock(
name: str,
diameter: float = 20.0,
height: float = 100.0,
max_volume: float = 30000.0, # 30mL
barcode: str = None,
) -> Bottle:
"""创建粉末瓶"""
return Bottle(
name=name,
diameter=diameter,
height=height,
max_volume=max_volume,
barcode=barcode,
model="BIOYOND_PolymerStation_Solid_Stock",
)
def BIOYOND_PolymerStation_Solid_Vial(
name: str,
diameter: float = 25.0,
height: float = 60.0,
max_volume: float = 30000.0, # 30mL
barcode: str = None,
) -> Bottle:
"""创建粉末瓶"""
return Bottle(
name=name,
diameter=diameter,
height=height,
max_volume=max_volume,
barcode=barcode,
model="BIOYOND_PolymerStation_Solid_Vial",
)
def BIOYOND_PolymerStation_Liquid_Vial(
name: str,
diameter: float = 25.0,
height: float = 60.0,
max_volume: float = 30000.0, # 30mL
barcode: str = None,
) -> Bottle:
"""创建滴定液瓶"""
return Bottle(
name=name,
diameter=diameter,
height=height,
max_volume=max_volume,
barcode=barcode,
model="BIOYOND_PolymerStation_Liquid_Vial",
)
def BIOYOND_PolymerStation_Solution_Beaker(
name: str,
diameter: float = 60.0,
height: float = 70.0,
max_volume: float = 200000.0, # 200mL
barcode: str = None,
) -> Bottle:
"""创建溶液烧杯"""
return Bottle(
name=name,
diameter=diameter,
height=height,
max_volume=max_volume,
barcode=barcode,
model="BIOYOND_PolymerStation_Solution_Beaker",
)
def BIOYOND_PolymerStation_Reagent_Bottle(
name: str,
diameter: float = 70.0,
height: float = 120.0,
max_volume: float = 500000.0, # 500mL
barcode: str = None,
) -> Bottle:
"""创建试剂瓶"""
return Bottle(
name=name,
diameter=diameter,
height=height,
max_volume=max_volume,
barcode=barcode,
model="BIOYOND_PolymerStation_Reagent_Bottle",
)
def BIOYOND_PolymerStation_Reactor(
name: str,
diameter: float = 30.0,
height: float = 80.0,
max_volume: float = 50000.0, # 50mL
barcode: str = None,
) -> Bottle:
"""创建反应器"""
return Bottle(
name=name,
diameter=diameter,
height=height,
max_volume=max_volume,
barcode=barcode,
model="BIOYOND_PolymerStation_Reactor",
)
def BIOYOND_PolymerStation_TipBox(
name: str,
size_x: float = 127.76, # 枪头盒宽度
size_y: float = 85.48, # 枪头盒长度
size_z: float = 100.0, # 枪头盒高度
barcode: str = None,
):
"""创建4×6枪头盒 (24个枪头)
Args:
name: 枪头盒名称
size_x: 枪头盒宽度 (mm)
size_y: 枪头盒长度 (mm)
size_z: 枪头盒高度 (mm)
barcode: 条形码
Returns:
TipBoxCarrier: 包含24个枪头孔位的枪头盒
"""
from pylabrobot.resources import Container, Coordinate
# 创建枪头盒容器
tip_box = Container(
name=name,
size_x=size_x,
size_y=size_y,
size_z=size_z,
category="tip_rack",
model="BIOYOND_PolymerStation_TipBox_4x6",
)
# 设置自定义属性
tip_box.barcode = barcode
tip_box.tip_count = 24 # 4行×6列
tip_box.num_items_x = 6 # 6列
tip_box.num_items_y = 4 # 4行
# 创建24个枪头孔位 (4行×6列)
# 假设孔位间距为 9mm
tip_spacing_x = 9.0 # 列间距
tip_spacing_y = 9.0 # 行间距
start_x = 14.38 # 第一个孔位的x偏移
start_y = 11.24 # 第一个孔位的y偏移
for row in range(4): # A, B, C, D
for col in range(6): # 1-6
spot_name = f"{chr(65 + row)}{col + 1}" # A1, A2, ..., D6
x = start_x + col * tip_spacing_x
y = start_y + row * tip_spacing_y
# 创建枪头孔位容器
tip_spot = Container(
name=spot_name,
size_x=8.0, # 单个枪头孔位大小
size_y=8.0,
size_z=size_z - 10.0, # 略低于盒子高度
category="tip_spot",
)
# 添加到枪头盒
tip_box.assign_child_resource(
tip_spot,
location=Coordinate(x=x, y=y, z=0)
)
return tip_box
def BIOYOND_PolymerStation_Flask(
name: str,
diameter: float = 60.0,
height: float = 70.0,
max_volume: float = 200000.0, # 200mL
barcode: str = None,
) -> Bottle:
"""聚合站-烧杯（统一 Flask 资源到 PolymerStation）"""
return Bottle(
name=name,
diameter=diameter,
height=height,
max_volume=max_volume,
barcode=barcode,
model="BIOYOND_PolymerStation_Flask",
)

115
unilabos/resources/bioyond/decks.py
View File

@@ -1,13 +1,29 @@
from os import name
from pickle import TRUE
from pylabrobot.resources import Deck, Coordinate, Rotation
from unilabos.resources.bioyond.YB_warehouses import bioyond_warehouse_1x4x4, bioyond_warehouse_1x4x2, bioyond_warehouse_liquid_and_lid_handling, bioyond_warehouse_1x2x2, bioyond_warehouse_1x3x3, bioyond_warehouse_10x1x1, bioyond_warehouse_3x3x1, bioyond_warehouse_3x3x1_2, bioyond_warehouse_5x1x1, bioyond_warehouse_20x1x1, bioyond_warehouse_2x2x1, bioyond_warehouse_3x5x1
from unilabos.resources.bioyond.warehouses import (
bioyond_warehouse_1x4x4,
bioyond_warehouse_1x4x4_right, # 新增：右侧仓库 (A05～D08)
bioyond_warehouse_1x4x2,
bioyond_warehouse_reagent_stack, # 新增：试剂堆栈 (A1-B4)
bioyond_warehouse_liquid_and_lid_handling,
bioyond_warehouse_1x2x2,
bioyond_warehouse_1x3x3,
bioyond_warehouse_10x1x1,
bioyond_warehouse_3x3x1,
bioyond_warehouse_3x3x1_2,
bioyond_warehouse_5x1x1,
bioyond_warehouse_1x8x4,
bioyond_warehouse_reagent_storage,
# bioyond_warehouse_liquid_preparation,
bioyond_warehouse_tipbox_storage, # 新增：Tip盒堆栈
bioyond_warehouse_density_vial,
)
class BIOYOND_PolymerReactionStation_Deck(Deck):
def __init__(
self,
self,
name: str = "PolymerReactionStation_Deck",
size_x: float = 2700.0,
size_y: float = 1080.0,
@@ -21,24 +37,35 @@ class BIOYOND_PolymerReactionStation_Deck(Deck):
def setup(self) -> None:
# 添加仓库
# 说明: 堆栈1物理上分为左右两部分
# - 堆栈1左: A01～D04 (4行×4列, 位于反应站左侧)
# - 堆栈1右: A05～D08 (4行×4列, 位于反应站右侧)
self.warehouses = {
"堆栈1": bioyond_warehouse_1x4x4("堆栈1"),
"堆栈2": bioyond_warehouse_1x4x4("堆栈2"),
"站内试剂存放堆栈": bioyond_warehouse_liquid_and_lid_handling("站内试剂存放堆栈"),
"堆栈1左": bioyond_warehouse_1x4x4("堆栈1左"), # 左侧堆栈: A01～D04
"堆栈1右": bioyond_warehouse_1x4x4_right("堆栈1右"), # 右侧堆栈: A05～D08
"站内试剂存放堆栈": bioyond_warehouse_reagent_storage("站内试剂存放堆栈"), # A01～A02
# "移液站内10%分装液体准备仓库": bioyond_warehouse_liquid_preparation("移液站内10%分装液体准备仓库"), # A01～B04
"站内Tip盒堆栈": bioyond_warehouse_tipbox_storage("站内Tip盒堆栈"), # A01～B03, 存放枪头盒.
"测量小瓶仓库(测密度)": bioyond_warehouse_density_vial("测量小瓶仓库(测密度)"), # A01～B03
}
self.warehouse_locations = {
"堆栈1": Coordinate(0.0, 430.0, 0.0),
"堆栈2": Coordinate(2550.0, 430.0, 0.0),
"站内试剂存放堆栈": Coordinate(800.0, 475.0, 0.0),
"堆栈1左": Coordinate(0.0, 430.0, 0.0), # 左侧位置
"堆栈1右": Coordinate(2500.0, 430.0, 0.0), # 右侧位置
"站内试剂存放堆栈": Coordinate(640.0, 480.0, 0.0),
# "移液站内10%分装液体准备仓库": Coordinate(1200.0, 600.0, 0.0),
"站内Tip盒堆栈": Coordinate(300.0, 150.0, 0.0),
"测量小瓶仓库(测密度)": Coordinate(922.0, 552.0, 0.0),
}
self.warehouses["站内试剂存放堆栈"].rotation = Rotation(z=90)
self.warehouses["测量小瓶仓库(测密度)"].rotation = Rotation(z=270)
for warehouse_name, warehouse in self.warehouses.items():
self.assign_child_resource(warehouse, location=self.warehouse_locations[warehouse_name])
class BIOYOND_PolymerPreparationStation_Deck(Deck):
def __init__(
self,
self,
name: str = "PolymerPreparationStation_Deck",
size_x: float = 2700.0,
size_y: float = 1080.0,
@@ -51,18 +78,21 @@ class BIOYOND_PolymerPreparationStation_Deck(Deck):
self.setup()
def setup(self) -> None:
# 添加仓库
# 添加仓库 - 配液站的3个堆栈，使用Bioyond系统中的实际名称
# 样品类型（typeMode=1）：烧杯、试剂瓶、分装板 → 试剂堆栈、溶液堆栈
# 试剂类型（typeMode=2）：样品板 → 粉末堆栈
self.warehouses = {
"io_warehouse_left": bioyond_warehouse_1x4x4("io_warehouse_left"),
"io_warehouse_right": bioyond_warehouse_1x4x4("io_warehouse_right"),
"solutions": bioyond_warehouse_1x4x2("warehouse_solutions"),
"liquid_and_lid_handling": bioyond_warehouse_liquid_and_lid_handling("warehouse_liquid_and_lid_handling"),
# 试剂类型 - 样品板
"粉末堆栈": bioyond_warehouse_1x4x4("粉末堆栈"), # 4行×4列 (A01-D04)
# 样品类型 - 烧杯、试剂瓶、分装板
"试剂堆栈": bioyond_warehouse_reagent_stack("试剂堆栈"), # 2行×4列 (A01-B04)
"溶液堆栈": bioyond_warehouse_1x4x4("溶液堆栈"), # 4行×4列 (A01-D04)
}
self.warehouse_locations = {
"io_warehouse_left": Coordinate(0.0, 650.0, 0.0),
"io_warehouse_right": Coordinate(2550.0, 650.0, 0.0),
"solutions": Coordinate(1915.0, 900.0, 0.0),
"liquid_and_lid_handling": Coordinate(1330.0, 490.0, 0.0),
"粉末堆栈": Coordinate(0.0, 450.0, 0.0),
"试剂堆栈": Coordinate(1850.0, 200.0, 0.0),
"溶液堆栈": Coordinate(2500.0, 450.0, 0.0),
}
for warehouse_name, warehouse in self.warehouses.items():
@@ -70,7 +100,7 @@ class BIOYOND_PolymerPreparationStation_Deck(Deck):
class BIOYOND_YB_Deck(Deck):
def __init__(
self,
self,
name: str = "YB_Deck",
size_x: float = 4150,
size_y: float = 1400.0,
@@ -85,32 +115,39 @@ class BIOYOND_YB_Deck(Deck):
def setup(self) -> None:
# 添加仓库
self.warehouses = {
"自动堆栈-左": bioyond_warehouse_2x2x1("自动堆栈-左"),
"自动堆栈-右": bioyond_warehouse_2x2x1("自动堆栈-右"),
"手动堆栈-左": bioyond_warehouse_3x5x1("手动堆栈-左"),
"手动堆栈-右": bioyond_warehouse_3x5x1("手动堆栈-右"),
"粉末加样头堆栈": bioyond_warehouse_20x1x1("粉末加样头堆栈"),
"配液站内试剂仓库": bioyond_warehouse_3x3x1("配液站内试剂仓库"),
"试剂替换仓库": bioyond_warehouse_10x1x1("试剂替换仓库"),
"321窗口": bioyond_warehouse_1x2x2("321窗口"),
"43窗口": bioyond_warehouse_1x2x2("43窗口"),
"手动传递窗左": bioyond_warehouse_1x3x3("手动传递窗左"),
"手动传递窗右": bioyond_warehouse_1x3x3("手动传递窗右"),
"加样头堆栈左": bioyond_warehouse_10x1x1("加样头堆栈左"),
"加样头堆栈右": bioyond_warehouse_10x1x1("加样头堆栈右"),
"15ml配液堆栈左": bioyond_warehouse_3x3x1("15ml配液堆栈左"),
"母液加样右": bioyond_warehouse_3x3x1_2("母液加样右"),
"大瓶母液堆栈左": bioyond_warehouse_5x1x1("大瓶母液堆栈左"),
"大瓶母液堆栈右": bioyond_warehouse_5x1x1("大瓶母液堆栈右"),
}
# warehouse 的位置
self.warehouse_locations = {
"自动堆栈-左": Coordinate(-100.3, 171.5, 0.0),
"自动堆栈-右": Coordinate(3960.1, 155.9, 0.0),
"手动堆栈-左": Coordinate(-213.3, 804.4, 0.0),
"手动堆栈-右": Coordinate(3960.1, 807.6, 0.0),
"粉末加样头堆栈": Coordinate(415.0, 1301.0, 0.0),
"配液站内试剂仓库": Coordinate(2162.0, 437.0, 0.0),
"试剂替换仓库": Coordinate(1173.0, 802.0, 0.0),
"321窗口": Coordinate(-150.0, 158.0, 0.0),
"43窗口": Coordinate(4160.0, 158.0, 0.0),
"手动传递窗左": Coordinate(-150.0, 877.0, 0.0),
"手动传递窗右": Coordinate(4160.0, 877.0, 0.0),
"加样头堆栈左": Coordinate(385.0, 1300.0, 0.0),
"加样头堆栈右": Coordinate(2187.0, 1300.0, 0.0),
"15ml配液堆栈左": Coordinate(749.0, 355.0, 0.0),
"母液加样右": Coordinate(2152.0, 333.0, 0.0),
"大瓶母液堆栈左": Coordinate(1164.0, 676.0, 0.0),
"大瓶母液堆栈右": Coordinate(2717.0, 676.0, 0.0),
}
for warehouse_name, warehouse in self.warehouses.items():
self.assign_child_resource(warehouse, location=self.warehouse_locations[warehouse_name])
# def YB_Deck(name: str) -> Deck:
# # by=BIOYOND_YB_Deck(name=name)
# # by.setup()
# return None
def YB_Deck(name: str) -> Deck:
by=BIOYOND_YB_Deck(name=name)
by.setup()
return by

10
unilabos/resources/container.py
View File

@@ -9,6 +9,7 @@ from unilabos.ros.msgs.message_converter import convert_from_ros_msg
class RegularContainer(Container):
def __init__(self, *args, **kwargs):
pose = kwargs.pop("pose", None)
if "size_x" not in kwargs:
kwargs["size_x"] = 0
if "size_y" not in kwargs:
@@ -17,10 +18,17 @@ class RegularContainer(Container):
kwargs["size_z"] = 0
self.kwargs = kwargs
self.state = {}
super().__init__(*args, **kwargs)
super().__init__(*args, category="container", **kwargs)
def load_state(self, state: Dict[str, Any]):
self.state = state
def get_regular_container(name="container"):
r = RegularContainer(name=name)
r.category = "container"
return RegularContainer(name=name)
#
# class RegularContainer(object):
# # 第一个参数必须是id传入

522
unilabos/resources/graphio.py
View File

@@ -11,7 +11,7 @@ from unilabos_msgs.msg import Resource
from unilabos.config.config import BasicConfig
from unilabos.resources.container import RegularContainer
from unilabos.resources.itemized_carrier import ItemizedCarrier
from unilabos.resources.itemized_carrier import ItemizedCarrier, BottleCarrier
from unilabos.ros.msgs.message_converter import convert_to_ros_msg
from unilabos.ros.nodes.resource_tracker import (
ResourceDictInstance,
@@ -42,13 +42,16 @@ def canonicalize_nodes_data(
Returns:
ResourceTreeSet: 标准化后的资源树集合
"""
print_status(f"{len(nodes)} Resources loaded:", "info")
print_status(f"{len(nodes)} Resources loaded", "info")
# 第一步：基本预处理（处理graphml的label字段）
for node in nodes:
outer_host_node_id = None
for idx, node in enumerate(nodes):
if node.get("label") is not None:
node_id = node.pop("label")
node["id"] = node["name"] = node_id
if node["id"] == "host_node":
outer_host_node_id = idx
if not isinstance(node.get("config"), dict):
node["config"] = {}
if not node.get("type"):
@@ -58,25 +61,26 @@ def canonicalize_nodes_data(
node["name"] = node.get("id")
print_status(f"Warning: Node {node.get('id', 'unknown')} missing 'name', defaulting to {node['name']}", "warning")
if not isinstance(node.get("position"), dict):
node["position"] = {"position": {}}
node["pose"] = {"position": {}}
x = node.pop("x", None)
if x is not None:
node["position"]["position"]["x"] = x
node["pose"]["position"]["x"] = x
y = node.pop("y", None)
if y is not None:
node["position"]["position"]["y"] = y
node["pose"]["position"]["y"] = y
z = node.pop("z", None)
if z is not None:
node["position"]["position"]["z"] = z
node["pose"]["position"]["z"] = z
if "sample_id" in node:
sample_id = node.pop("sample_id")
if sample_id:
logger.error(f"{node}的sample_id参数已弃用，sample_id: {sample_id}")
for k in list(node.keys()):
if k not in ["id", "uuid", "name", "description", "schema", "model", "icon", "parent_uuid", "parent", "type", "class", "position", "config", "data", "children"]:
if k not in ["id", "uuid", "name", "description", "schema", "model", "icon", "parent_uuid", "parent", "type", "class", "position", "config", "data", "children", "pose"]:
v = node.pop(k)
node["config"][k] = v
if outer_host_node_id is not None:
nodes.pop(outer_host_node_id)
# 第二步：处理parent_relation
id2idx = {node["id"]: idx for idx, node in enumerate(nodes)}
for parent, children in parent_relation.items():
@@ -93,7 +97,7 @@ def canonicalize_nodes_data(
for node in nodes:
try:
print_status(f"DeviceId: {node['id']}, Class: {node['class']}", "info")
# print_status(f"DeviceId: {node['id']}, Class: {node['class']}", "info")
# 使用标准化方法
resource_instance = ResourceDictInstance.get_resource_instance_from_dict(node)
known_nodes[node["id"]] = resource_instance
@@ -228,7 +232,7 @@ def handle_communications(G: nx.Graph):
if G.nodes[device_comm].get("class") == "serial":
G.nodes[device]["config"]["port"] = device_comm
elif G.nodes[device_comm].get("class") == "io_device":
print(f'!!! Modify {device}\'s io_device_port to {edata["port"][device_comm]}')
logger.warning(f'Modify {device}\'s io_device_port to {edata["port"][device_comm]}')
G.nodes[device]["config"]["io_device_port"] = int(edata["port"][device_comm])
@@ -580,11 +584,17 @@ def resource_plr_to_ulab(resource_plr: "ResourcePLR", parent_name: str = None, w
"trash": "trash",
"deck": "deck",
"tip_rack": "tip_rack",
"warehouse": "warehouse",
"container": "container",
"tube": "tube",
"bottle_carrier": "bottle_carrier",
"plate_adapter": "plate_adapter",
}
if source in replace_info:
return replace_info[source]
else:
print("转换pylabrobot的时候，出现未知类型", source)
if source is not None:
logger.warning(f"转换pylabrobot的时候，出现未知类型: {source}")
return source
def resource_plr_to_ulab_inner(d: dict, all_states: dict, child=True) -> dict:
@@ -619,131 +629,480 @@ def resource_bioyond_to_plr(bioyond_materials: list[dict], type_mapping: Dict[st
Args:
bioyond_materials: bioyond 系统的物料查询结果列表
type_mapping: 物料类型映射字典，格式 {bioyond_type: [plr_class_name, class_uuid]}
type_mapping: 物料类型映射字典，格式 {model: (显示名称, UUID)} 或 {显示名称: (model, UUID)}
location_id_mapping: 库位 ID 到名称的映射字典，格式 {location_id: location_name}
Returns:
pylabrobot 格式的物料列表
"""
print("1:bioyond_materials:",bioyond_materials)
# print("2:type_mapping:",type_mapping)
plr_materials = []
# 创建反向映射: {显示名称: (model, UUID)} -> 用于从 Bioyond typeName 查找 model
# 如果 type_mapping 的 key 已经是显示名称,则直接使用;否则创建反向映射
reverse_type_mapping = {}
for key, value in type_mapping.items():
# value 可能是 tuple 或 list: (显示名称, UUID) 或 [显示名称, UUID]
display_name = value[0] if isinstance(value, (tuple, list)) and len(value) >= 1 else None
if display_name:
# 反向映射: {显示名称: (原始key作为model, UUID)}
resource_uuid = value[1] if len(value) >= 2 else ""
# 如果已存在该显示名称,跳过(保留第一个遇到的映射)
if display_name not in reverse_type_mapping:
reverse_type_mapping[display_name] = (key, resource_uuid)
logger.debug(f"[反向映射表] 共 {len(reverse_type_mapping)} 个条目: {list(reverse_type_mapping.keys())}")
# 用于跟踪同名物料的计数器
name_counter = {}
for material in bioyond_materials:
className = (
type_mapping.get(material.get("typeName"), ("RegularContainer", ""))[0] if type_mapping else "RegularContainer"
# 从反向映射中查找: typeName(显示名称) -> (model, UUID)
type_info = reverse_type_mapping.get(material.get("typeName"))
className = type_info[0] if type_info else "RegularContainer"
# 为同名物料添加唯一后缀
base_name = material["name"]
if base_name in name_counter:
name_counter[base_name] += 1
unique_name = f"{base_name}_{name_counter[base_name]}"
else:
name_counter[base_name] = 1
unique_name = base_name
plr_material_result = initialize_resource(
{"name": unique_name, "class": className}, resource_type=ResourcePLR
)
plr_material: ResourcePLR = initialize_resource(
{"name": material["name"], "class": className}, resource_type=ResourcePLR
)
print("plr_material:",plr_material)
print("code:",material.get("code", ""))
# initialize_resource 可能返回列表或单个对象
if isinstance(plr_material_result, list):
if len(plr_material_result) == 0:
logger.warning(f"物料 {material['name']} 初始化失败，跳过")
continue
plr_material = plr_material_result[0]
else:
plr_material = plr_material_result
# 确保 plr_material 是 ResourcePLR 实例
if not isinstance(plr_material, ResourcePLR):
logger.warning(f"物料 {unique_name} 不是有效的 ResourcePLR 实例，类型: {type(plr_material)}")
continue
plr_material.code = material.get("code", "") and material.get("barCode", "") or ""
plr_material.unilabos_uuid = str(uuid.uuid4())
# ⭐ 保存 Bioyond 原始信息到 unilabos_extra（用于出库时查询）
plr_material.unilabos_extra = {
"material_bioyond_id": material.get("id"), # Bioyond 物料 UUID
"material_bioyond_name": material.get("name"), # Bioyond 原始名称（如 "MDA"）
"material_bioyond_type": material.get("typeName"), # Bioyond 物料类型名称
}
logger.debug(f"[转换物料] {material['name']} (ID:{material['id']}) → {unique_name} (类型:{className})")
# 处理子物料（detail）
if material.get("detail") and len(material["detail"]) > 0:
for bottle in reversed(plr_material.children):
plr_material.unassign_child_resource(bottle)
child_ids = []
# 确定detail物料的默认类型
# 样品板的detail通常是样品瓶
default_detail_type = "样品瓶" if "样品板" in material.get("typeName", "") else None
for detail in material["detail"]:
number = (
(detail.get("z", 0) - 1) * plr_material.num_items_x * plr_material.num_items_y
+ (detail.get("y", 0) - 1) * plr_material.num_items_y
+ (detail.get("x", 0) - 1)
)
typeName = detail.get("typeName", detail.get("name", ""))
if typeName in type_mapping:
# 检查索引是否超出范围
max_index = plr_material.num_items_x * plr_material.num_items_y - 1
if number < 0 or number > max_index:
logger.warning(
f" └─ [子物料警告] {detail['name']} 的坐标 (x={detail.get('x')}, y={detail.get('y')}, z={detail.get('z')}) "
f"计算出索引 {number} 超出载架范围 [0-{max_index}] (布局: {plr_material.num_items_x}×{plr_material.num_items_y})，跳过"
)
continue
# detail可能没有typeName，尝试从name推断，或使用默认类型
typeName = detail.get("typeName")
# 如果没有typeName，尝试根据父物料类型和位置推断
if not typeName:
if "分装板" in material.get("typeName", ""):
# 分装板: 根据行(x)判断类型
# 第一行(x=1)是10%分装小瓶，第二行(x=2)是90%分装小瓶
x_pos = detail.get("x", 0)
y_pos = detail.get("y", 0)
# logger.debug(f" └─ [推断类型] {detail['name']} 坐标(x={x_pos}, y={y_pos})")
if x_pos == 1:
typeName = "10%分装小瓶"
elif x_pos == 2:
typeName = "90%分装小瓶"
# logger.debug(f" └─ [推断结果] {detail['name']} → {typeName}")
else:
typeName = default_detail_type
if typeName and typeName in reverse_type_mapping:
bottle = plr_material[number] = initialize_resource(
{"name": f'{detail["name"]}_{number}', "class": type_mapping[typeName][0]}, resource_type=ResourcePLR
{"name": f'{detail["name"]}_{number}', "class": reverse_type_mapping[typeName][0]}, resource_type=ResourcePLR
)
bottle.tracker.liquids = [
(detail["name"], float(detail.get("quantity", 0)) if detail.get("quantity") else 0)
]
bottle.code = detail.get("code", "")
logger.debug(f" └─ [子物料] {detail['name']} → {plr_material.name}[{number}] (类型:{typeName})")
else:
logger.warning(f" └─ [子物料警告] {detail['name']} 的类型 '{typeName}' 不在mapping中，跳过")
else:
bottle = plr_material[0] if plr_material.capacity > 0 else plr_material
bottle.tracker.liquids = [
(material["name"], float(material.get("quantity", 0)) if material.get("quantity") else 0)
]
# 只对有 capacity 属性的容器（液体容器）处理液体追踪
if hasattr(plr_material, 'capacity'):
bottle = plr_material[0] if plr_material.capacity > 0 else plr_material
bottle.tracker.liquids = [
(material["name"], float(material.get("quantity", 0)) if material.get("quantity") else 0)
]
plr_materials.append(plr_material)
if deck and hasattr(deck, "warehouses"):
for loc in material.get("locations", []):
if hasattr(deck, "warehouses") and loc.get("whName") in deck.warehouses:
warehouse = deck.warehouses[loc["whName"]]
num_x = getattr(warehouse, "num_items_x", 0) or 0
num_y = getattr(warehouse, "num_items_y", 0) or 0
num_z = getattr(warehouse, "num_items_z", 0) or 0
if num_x <= 0 or num_y <= 0 or num_z <= 0:
locations = material.get("locations", [])
if not locations:
logger.debug(f"[物料位置] {unique_name} 没有location信息，跳过warehouse放置")
for loc in locations:
wh_name = loc.get("whName")
logger.debug(f"[物料位置] {unique_name} 尝试放置到 warehouse: {wh_name} (Bioyond坐标: x={loc.get('x')}, y={loc.get('y')}, z={loc.get('z')})")
# 特殊处理: Bioyond的"堆栈1"需要映射到"堆栈1左"或"堆栈1右"
# 根据列号(x)判断: 1-4映射到左侧, 5-8映射到右侧
if wh_name == "堆栈1":
x_val = loc.get("x", 1)
if 1 <= x_val <= 4:
wh_name = "堆栈1左"
elif 5 <= x_val <= 8:
wh_name = "堆栈1右"
else:
logger.warning(f"物料 {material['name']} 的列号 x={x_val} 超出范围，无法映射到堆栈1左或堆栈1右")
continue
idx = (
(loc.get("z", 0) - 1) * num_x * num_y
+ (loc.get("y", 0) - 1) * num_x
+ (loc.get("x", 0) - 1)
)
if hasattr(deck, "warehouses") and wh_name in deck.warehouses:
warehouse = deck.warehouses[wh_name]
logger.debug(f"[Warehouse匹配] 找到warehouse: {wh_name} (容量: {warehouse.capacity}, 行×列: {warehouse.num_items_x}×{warehouse.num_items_y})")
# Bioyond坐标映射 (重要！): x→行(1=A,2=B...), y→列(1=01,2=02...), z→层(通常=1)
# PyLabRobot warehouse是列优先存储: A01,B01,C01,D01, A02,B02,C02,D02, ...
x = loc.get("x", 1) # 行号 (1-based: 1=A, 2=B, 3=C, 4=D)
y = loc.get("y", 1) # 列号 (1-based: 1=01, 2=02, 3=03...)
z = loc.get("z", 1) # 层号 (1-based, 通常为1)
# 如果是右侧堆栈，需要调整列号 (5→1, 6→2, 7→3, 8→4)
if wh_name == "堆栈1右":
y = y - 4 # 将5-8映射到1-4
# 特殊处理：对于1行×N列的横向warehouse（如站内试剂存放堆栈）
# Bioyond的y坐标表示线性位置序号，而不是列号
if warehouse.num_items_y == 1:
# 1行warehouse: 直接用y作为线性索引
idx = y - 1
logger.debug(f"1行warehouse {wh_name}: y={y} → idx={idx}")
else:
# 多行warehouse: 根据 layout 使用不同的索引计算
row_idx = x - 1 # x表示行: 转为0-based
col_idx = y - 1 # y表示列: 转为0-based
layer_idx = z - 1 # 转为0-based
# 检查 warehouse 的排序方式属性
ordering_layout = getattr(warehouse, 'ordering_layout', 'col-major')
logger.debug(f"🔍 Warehouse {wh_name} layout检测: hasattr={hasattr(warehouse, 'ordering_layout')}, ordering_layout值='{ordering_layout}', warehouse类型={type(warehouse).__name__}")
if ordering_layout == "row-major":
# 行优先: A01,A02,A03,A04, B01,B02,B03,B04 (所有Bioyond堆栈)
# 索引计算: idx = (row) * num_cols + (col) + (layer) * (rows * cols)
idx = layer_idx * (warehouse.num_items_x * warehouse.num_items_y) + row_idx * warehouse.num_items_x + col_idx
logger.debug(f"行优先warehouse {wh_name}: x={x}(行),y={y}(列) → row={row_idx},col={col_idx} → idx={idx}")
else:
# 列优先 (后备): A01,B01,C01,D01, A02,B02,C02,D02
# 索引计算: idx = (col) * num_rows + (row) + (layer) * (rows * cols)
idx = layer_idx * (warehouse.num_items_x * warehouse.num_items_y) + col_idx * warehouse.num_items_y + row_idx
logger.debug(f"列优先warehouse {wh_name}: x={x}(行),y={y}(列) → row={row_idx},col={col_idx} → idx={idx}")
if 0 <= idx < warehouse.capacity:
if warehouse[idx] is None or isinstance(warehouse[idx], ResourceHolder):
warehouse[idx] = plr_material
logger.debug(f"✅ 物料 {unique_name} 放置到 {wh_name}[{idx}] (Bioyond坐标: x={loc.get('x')}, y={loc.get('y')})")
else:
logger.warning(f"❌ 物料 {unique_name} 的索引 {idx} 超出仓库 {wh_name} 容量 {warehouse.capacity}")
else:
if wh_name:
logger.warning(f"❌ 物料 {unique_name} 的warehouse '{wh_name}' 在deck中不存在。可用warehouses: {list(deck.warehouses.keys()) if hasattr(deck, 'warehouses') else '无'}")
return plr_materials
def resource_plr_to_bioyond(plr_resources: list[ResourcePLR], type_mapping: dict = {}, warehouse_mapping: dict = {}) -> list[dict]:
def resource_plr_to_bioyond(plr_resources: list[ResourcePLR], type_mapping: dict = {}, warehouse_mapping: dict = {}, material_params: dict = {}) -> list[dict]:
"""
将 PyLabRobot 资源转换为 Bioyond 格式
Args:
plr_resources: PyLabRobot 资源列表
type_mapping: 物料类型映射字典
warehouse_mapping: 仓库映射字典
material_params: 物料默认参数字典 (格式: {物料名称: {参数字典}})
Returns:
Bioyond 格式的物料列表
"""
bioyond_materials = []
# 定义不需要发送 details 的载架类型
# 说明：这些载架上自带试剂瓶或烧杯，作为整体物料上传即可，不需要在 details 中重复上传子物料
CARRIERS_WITHOUT_DETAILS = {
"BIOYOND_PolymerStation_1BottleCarrier", # 聚合站-单试剂瓶载架
"BIOYOND_PolymerStation_1FlaskCarrier", # 聚合站-单烧杯载架
}
for resource in plr_resources:
if hasattr(resource, "capacity") and resource.capacity > 1:
if isinstance(resource, BottleCarrier) and resource.capacity > 1:
# 获取 BottleCarrier 的类型映射
type_info = type_mapping.get(resource.model)
if not type_info:
logger.error(f"❌ [PLR→Bioyond] BottleCarrier 资源 '{resource.name}' 的 model '{resource.model}' 不在 type_mapping 中")
logger.debug(f"[PLR→Bioyond] 可用的 type_mapping 键: {list(type_mapping.keys())}")
raise ValueError(f"资源 model '{resource.model}' 未在 MATERIAL_TYPE_MAPPINGS 中配置")
material = {
"typeId": type_mapping.get(resource.model)[1],
"typeId": type_info[1],
"code": "",
"barCode": "",
"name": resource.name,
"unit": "个",
"quantity": 1,
"details": [],
"Parameters": "{}"
}
for bottle in resource.children:
if isinstance(resource, ItemizedCarrier):
site = resource.get_child_identifier(bottle)
else:
site = {"x": bottle.location.x - 1, "y": bottle.location.y - 1}
detail_item = {
"typeId": type_mapping.get(bottle.model)[1],
"name": bottle.name,
"code": bottle.code if hasattr(bottle, "code") else "",
"quantity": sum(qty for _, qty in bottle.tracker.liquids) if hasattr(bottle, "tracker") else 0,
"x": site["x"] + 1,
"y": site["y"] + 1,
"molecular": 1,
"Parameters": json.dumps({"molecular": 1})
}
material["details"].append(detail_item)
else:
bottle = resource[0] if resource.capacity > 0 else resource
material = {
"typeId": "3a14196b-24f2-ca49-9081-0cab8021bf1a",
"name": resource.get("name", ""),
"unit": "",
"quantity": sum(qty for _, qty in bottle.tracker.liquids) if hasattr(bottle, "tracker") else 0,
"Parameters": "{}"
"Parameters": "{}" # API 实际要求的字段（必需）
}
if resource.parent is not None and isinstance(resource.parent, ItemizedCarrier):
# 如果是自带试剂瓶的载架类型，不处理子物料（details留空）
if resource.model in CARRIERS_WITHOUT_DETAILS:
logger.info(f"[PLR→Bioyond] 载架 '{resource.name}' (model: {resource.model}) 自带试剂瓶，不添加 details")
else:
# 处理其他载架类型的子物料
for bottle in resource.children:
if isinstance(resource, ItemizedCarrier):
# ⭐ 优化：直接使用 get_child_identifier 获取真实的子物料坐标
# 这个方法会遍历 resource.children 找到 bottle 对象的实际位置
site = resource.get_child_identifier(bottle)
# 🔧 如果 get_child_identifier 失败或返回无效坐标 (0,0)
# 这通常发生在子物料名称使用纯数字后缀时（如 "BTDA_0", "BTDA_4"）
if not site or (site.get("x") == 0 and site.get("y") == 0):
# 方法1: 尝试从名称中提取标识符并解析
bottle_identifier = None
if "_" in bottle.name:
bottle_identifier = bottle.name.split("_")[-1]
# 只有非纯数字标识符才尝试解析（如 "A1", "B2"）
if bottle_identifier and not bottle_identifier.isdigit():
try:
x_idx, y_idx, z_idx = resource._parse_identifier_to_indices(bottle_identifier, 0)
site = {"x": x_idx, "y": y_idx, "z": z_idx, "identifier": bottle_identifier}
logger.debug(f" 🔧 [坐标修正-方法1] 从名称 {bottle.name} 解析标识符 {bottle_identifier} → ({x_idx}, {y_idx})")
except Exception as e:
logger.warning(f" ⚠️ [坐标解析] 标识符 {bottle_identifier} 解析失败: {e}")
# 方法2: 如果方法1失败，使用线性索引反推坐标
if not site or (site.get("x") == 0 and site.get("y") == 0):
# 找到bottle在children中的索引位置
try:
# 遍历所有槽位找到bottle的实际位置
for idx in range(resource.num_items_x * resource.num_items_y):
if resource[idx] is bottle:
# 根据载架布局计算行列坐标
# ItemizedCarrier 默认是列优先布局 (A1,B1,C1,D1, A2,B2,C2,D2...)
col_idx = idx // resource.num_items_y # 列索引 (0-based)
row_idx = idx % resource.num_items_y # 行索引 (0-based)
site = {"x": col_idx, "y": row_idx, "z": 0, "identifier": str(idx)}
logger.debug(f" 🔧 [坐标修正-方法2] {bottle.name} 在索引 {idx} → 列={col_idx}, 行={row_idx}")
break
except Exception as e:
logger.error(f" ❌ [坐标计算失败] {bottle.name}: {e}")
# 最后的兜底：使用 (0,0)
site = {"x": 0, "y": 0, "z": 0, "identifier": ""}
else:
site = {"x": bottle.location.x - 1, "y": bottle.location.y - 1, "identifier": ""}
# 获取子物料的类型映射
bottle_type_info = type_mapping.get(bottle.model)
if not bottle_type_info:
logger.error(f"❌ [PLR→Bioyond] 子物料 '{bottle.name}' 的 model '{bottle.model}' 不在 type_mapping 中")
raise ValueError(f"子物料 model '{bottle.model}' 未在 MATERIAL_TYPE_MAPPINGS 中配置")
# ⚠️ 坐标系转换说明:
# _parse_identifier_to_indices 返回: x=列索引, y=行索引 (0-based)
# Bioyond 系统要求: x=行号, y=列号 (1-based)
# 因此需要交换 x 和 y!
bioyond_x = site["y"] + 1 # 行索引 → Bioyond的x (行号)
bioyond_y = site["x"] + 1 # 列索引 → Bioyond的y (列号)
# 🐛 调试日志
logger.debug(f"🔍 [PLR→Bioyond] detail转换: {bottle.name} → PLR(x={site['x']},y={site['y']},id={site.get('identifier','?')}) → Bioyond(x={bioyond_x},y={bioyond_y})")
# 🔥 提取物料名称：从 tracker.liquids 中获取第一个液体的名称（去除PLR系统添加的后缀）
# tracker.liquids 格式: [(物料名称, 数量), ...]
material_name = bottle_type_info[0] # 默认使用类型名称（如"样品瓶"）
if hasattr(bottle, "tracker") and bottle.tracker.liquids:
# 如果有液体，使用液体的名称
first_liquid_name = bottle.tracker.liquids[0][0]
# 去除PLR系统为了唯一性添加的后缀（如 "_0", "_1" 等）
if "_" in first_liquid_name and first_liquid_name.split("_")[-1].isdigit():
material_name = "_".join(first_liquid_name.split("_")[:-1])
else:
material_name = first_liquid_name
logger.debug(f" 💧 [物料名称] {bottle.name} 液体: {first_liquid_name} → 转换为: {material_name}")
else:
logger.debug(f" 📭 [物料名称] {bottle.name} 无液体，使用类型名: {material_name}")
detail_item = {
"typeId": bottle_type_info[1],
"code": bottle.code if hasattr(bottle, "code") else "",
"name": material_name, # 使用物料名称（如"9090"），而不是类型名称（"样品瓶"）
"quantity": sum(qty for _, qty in bottle.tracker.liquids) if hasattr(bottle, "tracker") else 0,
"x": bioyond_x,
"y": bioyond_y,
"z": 1,
"unit": "微升",
"Parameters": "{}" # API 实际要求的字段（必需）
}
material["details"].append(detail_item)
else:
# 单个瓶子(非载架)类型的资源
bottle = resource[0] if hasattr(resource, "capacity") and resource.capacity > 0 else resource
# 根据 resource.model 从 type_mapping 获取正确的 typeId
type_info = type_mapping.get(resource.model)
if type_info:
type_id = type_info[1]
else:
# 如果找不到映射，记录警告并使用默认值
logger.warning(f"[PLR→Bioyond] 资源 {resource.name} 的 model '{resource.model}' 不在 type_mapping 中，使用默认烧杯类型")
type_id = "3a14196b-24f2-ca49-9081-0cab8021bf1a" # 默认使用烧杯类型
# 🔥 提取物料名称：优先使用液体名称，否则使用资源名称
material_name = resource.name if hasattr(resource, "name") else ""
if hasattr(bottle, "tracker") and bottle.tracker.liquids:
# 如果有液体，使用液体的名称
first_liquid_name = bottle.tracker.liquids[0][0]
# 去除PLR系统为了唯一性添加的后缀（如 "_0", "_1" 等）
if "_" in first_liquid_name and first_liquid_name.split("_")[-1].isdigit():
material_name = "_".join(first_liquid_name.split("_")[:-1])
else:
material_name = first_liquid_name
logger.debug(f" 💧 [单瓶物料] {resource.name} 液体: {first_liquid_name} → 转换为: {material_name}")
else:
logger.debug(f" 📭 [单瓶物料] {resource.name} 无液体，使用资源名: {material_name}")
# 🎯 处理物料默认参数和单位
# 检查是否有该物料名称的默认参数配置
default_unit = "个" # 默认单位
material_parameters = {}
if material_name in material_params:
params_config = material_params[material_name].copy()
# 提取 unit 字段（如果有）
if "unit" in params_config:
default_unit = params_config.pop("unit") # 从参数中移除，放到外层
# 剩余的字段放入 Parameters
material_parameters = params_config
logger.debug(f" 🔧 [物料参数] 为 {material_name} 应用配置: unit={default_unit}, parameters={material_parameters}")
# 转换为 JSON 字符串
parameters_json = json.dumps(material_parameters) if material_parameters else "{}"
material = {
"typeId": type_id,
"code": "",
"barCode": "",
"name": material_name, # 使用物料名称而不是资源名称
"unit": default_unit, # 使用配置的单位或默认单位
"quantity": sum(qty for _, qty in bottle.tracker.liquids) if hasattr(bottle, "tracker") else 0,
"Parameters": parameters_json # API 实际要求的字段（必需）
}
# ⭐ 处理 locations 信息
# 优先级: update_resource_site (位置更新请求) > 当前 parent 位置
extra_info = getattr(resource, "unilabos_extra", {})
update_site = extra_info.get("update_resource_site")
if update_site:
# 情况1: 有明确的位置更新请求 (如从 A02 移动到 A03)
# 需要从 warehouse_mapping 中查找目标库位的 UUID
logger.debug(f"🔄 [PLR→Bioyond] 检测到位置更新请求: {resource.name} → {update_site}")
# 遍历所有仓库查找目标库位
target_warehouse_name = None
target_location_uuid = None
for warehouse_name, warehouse_info in warehouse_mapping.items():
site_uuids = warehouse_info.get("site_uuids", {})
if update_site in site_uuids:
target_warehouse_name = warehouse_name
target_location_uuid = site_uuids[update_site]
break
if target_warehouse_name and target_location_uuid:
# 从库位代码解析坐标 (如 "A03" -> x=1, y=3)
# A=1, B=2, C=3, D=4...
# 01=1, 02=2, 03=3...
try:
row_letter = update_site[0] # 'A', 'B', 'C', 'D'
col_number = int(update_site[1:]) # '01', '02', '03'...
bioyond_x = ord(row_letter) - ord('A') + 1 # A→1, B→2, C→3, D→4
bioyond_y = col_number # 01→1, 02→2, 03→3
material["locations"] = [
{
"id": target_location_uuid,
"whid": warehouse_mapping[target_warehouse_name].get("uuid", ""),
"whName": target_warehouse_name,
"x": bioyond_x,
"y": bioyond_y,
"z": 1,
"quantity": 0
}
]
logger.debug(f"✅ [PLR→Bioyond] 位置更新: {resource.name} → {target_warehouse_name}/{update_site} (x={bioyond_x}, y={bioyond_y})")
except Exception as e:
logger.error(f"❌ [PLR→Bioyond] 解析库位代码失败: {update_site}, 错误: {e}")
else:
logger.warning(f"⚠️ [PLR→Bioyond] 未找到库位 {update_site} 的配置")
elif resource.parent is not None and isinstance(resource.parent, ItemizedCarrier):
# 情况2: 使用当前 parent 位置
site_in_parent = resource.parent.get_child_identifier(resource)
# ⚠️ 坐标系转换说明:
# get_child_identifier 返回: x_idx=列索引, y_idx=行索引 (0-based)
# Bioyond 系统要求: x=行号, y=列号 (1-based)
# 因此需要交换 x 和 y!
bioyond_x = site_in_parent["y"] + 1 # 行索引 → Bioyond的x (行号)
bioyond_y = site_in_parent["x"] + 1 # 列索引 → Bioyond的y (列号)
material["locations"] = [
{
"id": warehouse_mapping[resource.parent.name]["site_uuids"][site_in_parent["identifier"]],
"whid": warehouse_mapping[resource.parent.name]["uuid"],
"whName": resource.parent.name,
"x": site_in_parent["z"] + 1,
"y": site_in_parent["y"] + 1,
"x": bioyond_x,
"y": bioyond_y,
"z": 1,
"quantity": 0
}
],
]
logger.debug(f"🔄 [PLR→Bioyond] 坐标转换: {resource.name} 在 {resource.parent.name}[{site_in_parent['identifier']}] → UniLab(列={site_in_parent['x']},行={site_in_parent['y']}) → Bioyond(x={bioyond_x},y={bioyond_y})")
print(f"material_data: {material}")
bioyond_materials.append(material)
return bioyond_materials
@@ -768,6 +1127,8 @@ def initialize_resource(resource_config: dict, resource_type: Any = None) -> Uni
elif type(resource_class_config) == str:
# Allow special resource class names to be used
if resource_class_config not in lab_registry.resource_type_registry:
logger.warning(f"❌ 类 {resource_class_config} 不在 registry 中，返回原始配置")
logger.debug(f" 可用的类: {list(lab_registry.resource_type_registry.keys())[:10]}...")
return [resource_config]
# If the resource class is a string, look up the class in the
# resource_type_registry and import it
@@ -782,11 +1143,12 @@ def initialize_resource(resource_config: dict, resource_type: Any = None) -> Uni
if resource_class_config["type"] == "pylabrobot":
resource_plr = RESOURCE(name=resource_config["name"])
if resource_type != ResourcePLR:
r = resource_plr_to_ulab(resource_plr=resource_plr, parent_name=resource_config.get("parent", None))
# r = resource_plr_to_ulab(resource_plr=resource_plr)
if resource_config.get("position") is not None:
r["position"] = resource_config["position"]
r = tree_to_list([r])
tree_sets = ResourceTreeSet.from_plr_resources([resource_plr])
# r = resource_plr_to_ulab(resource_plr=resource_plr, parent_name=resource_config.get("parent", None))
# # r = resource_plr_to_ulab(resource_plr=resource_plr)
# if resource_config.get("position") is not None:
# r["position"] = resource_config["position"]
r = tree_sets.dump()
else:
r = resource_plr
elif resource_class_config["type"] == "unilabos":

36
unilabos/resources/itemized_carrier.py
View File

@@ -146,7 +146,7 @@ class ItemizedCarrier(ResourcePLR):
if site_location == location:
idx = i
break
if not reassign and self.sites[idx] is not None:
raise ValueError(f"a site with index {idx} already exists")
super().assign_child_resource(resource, location=location, reassign=reassign)
@@ -172,18 +172,18 @@ class ItemizedCarrier(ResourcePLR):
def get_child_identifier(self, child: ResourcePLR):
"""Get the identifier information for a given child resource.
Args:
child: The Resource object to find the identifier for
Returns:
dict: A dictionary containing:
- identifier: The string identifier (e.g. "A1", "B2")
- idx: The integer index in the sites list
- x: The x index (column index, 0-based)
- y: The y index (row index, 0-based)
- y: The y index (row index, 0-based)
- z: The z index (layer index, 0-based)
Raises:
ValueError: If the child resource is not found in this carrier
"""
@@ -192,10 +192,10 @@ class ItemizedCarrier(ResourcePLR):
if resource is child:
# Get the identifier from ordering keys
identifier = list(self._ordering.keys())[idx]
# Parse identifier to get x, y, z indices
x_idx, y_idx, z_idx = self._parse_identifier_to_indices(identifier, idx)
return {
"identifier": identifier,
"idx": idx,
@@ -203,17 +203,17 @@ class ItemizedCarrier(ResourcePLR):
"y": y_idx,
"z": z_idx
}
# If not found, raise an error
raise ValueError(f"Resource {child} is not assigned to this carrier")
def _parse_identifier_to_indices(self, identifier: str, idx: int) -> Tuple[int, int, int]:
"""Parse identifier string to get x, y, z indices.
Args:
identifier: String identifier like "A1", "B2", etc.
idx: Linear index as fallback for calculation
Returns:
Tuple of (x_idx, y_idx, z_idx)
"""
@@ -225,31 +225,31 @@ class ItemizedCarrier(ResourcePLR):
y_idx = remaining // self.num_items_x
x_idx = remaining % self.num_items_x
return x_idx, y_idx, z_idx
# Fallback: parse from Excel-style identifier
if isinstance(identifier, str) and len(identifier) >= 2:
# Extract row (letter) and column (number)
row_letters = ""
col_numbers = ""
for char in identifier:
if char.isalpha():
row_letters += char
elif char.isdigit():
col_numbers += char
if row_letters and col_numbers:
# Convert letter(s) to row index (A=0, B=1, etc.)
y_idx = 0
for char in row_letters:
y_idx = y_idx * 26 + (ord(char.upper()) - ord('A'))
# Convert number to column index (1-based to 0-based)
x_idx = int(col_numbers) - 1
z_idx = 0 # Default layer
return x_idx, y_idx, z_idx
# If all else fails, assume linear arrangement
return idx, 0, 0
@@ -413,8 +413,8 @@ class ItemizedCarrier(ResourcePLR):
"sites": [{
"label": str(identifier),
"visible": False if identifier in self.invisible_slots else True,
"occupied_by": self[identifier].name
if isinstance(self[identifier], ResourcePLR) and not isinstance(self[identifier], ResourceHolder) else
"occupied_by": self[identifier].name
if isinstance(self[identifier], ResourcePLR) and not isinstance(self[identifier], ResourceHolder) else
self[identifier] if isinstance(self[identifier], str) else None,
"position": {"x": location.x, "y": location.y, "z": location.z},
"size": self.child_size[identifier],

7
unilabos/resources/plr_additional_res_reg.py
View File

@@ -7,5 +7,10 @@ def register():
from unilabos.devices.liquid_handling.prcxi.prcxi import PRCXI9300Container
# noinspection PyUnresolvedReferences
from unilabos.devices.workstation.workstation_base import WorkStationContainer
from unilabos.devices.liquid_handling.laiyu.laiyu import TransformXYZDeck
from unilabos.devices.liquid_handling.laiyu.laiyu import TransformXYZContainer
from unilabos.devices.liquid_handling.rviz_backend import UniLiquidHandlerRvizBackend
from unilabos.devices.liquid_handling.laiyu.backend.laiyu_v_backend import UniLiquidHandlerLaiyuBackend

60
unilabos/resources/warehouse.py
View File

@@ -8,7 +8,7 @@ from unilabos.resources.itemized_carrier import ItemizedCarrier, ResourcePLR
LETTERS = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
def YB_warehouse_factory(
def warehouse_factory(
name: str,
num_items_x: int = 1,
num_items_y: int = 4,
@@ -19,19 +19,33 @@ def YB_warehouse_factory(
item_dx: float = 10.0,
item_dy: float = 10.0,
item_dz: float = 10.0,
resource_size_x: float = 127.0,
resource_size_y: float = 86.0,
resource_size_z: float = 25.0,
removed_positions: Optional[List[int]] = None,
empty: bool = False,
category: str = "warehouse",
model: Optional[str] = None,
col_offset: int = 0, # 列起始偏移量，用于生成A05-D08等命名
layout: str = "col-major", # 新增：排序方式，"col-major"=列优先，"row-major"=行优先
):
# 创建16个板架位 (4层 x 4位置)
# 创建位置坐标
locations = []
for layer in range(num_items_z): # 4层
for row in range(num_items_y): # 4行
for col in range(num_items_x): # 1列 (每层4x1=4个位置)
for layer in range(num_items_z): # 层
for row in range(num_items_y): # 行
for col in range(num_items_x): # 列
# 计算位置
x = dx + col * item_dx
y = dy + (num_items_y - row - 1) * item_dy
# 根据 layout 决定 y 坐标计算
if layout == "row-major":
# 行优先：row=0(A行) 应该显示在上方，需要较小的 y 值
y = dy + row * item_dy
else:
# 列优先：保持原逻辑（row=0 对应较大的 y）
y = dy + (num_items_y - row - 1) * item_dy
z = dz + (num_items_z - layer - 1) * item_dz
locations.append(Coordinate(x, y, z))
if removed_positions:
@@ -39,15 +53,25 @@ def YB_warehouse_factory(
_sites = create_homogeneous_resources(
klass=ResourceHolder,
locations=locations,
resource_size_x=127.0,
resource_size_y=86.0,
resource_size_x=resource_size_x,
resource_size_y=resource_size_y,
resource_size_z=resource_size_z,
name_prefix=name,
)
len_x, len_y = (num_items_x, num_items_y) if num_items_z == 1 else (num_items_y, num_items_z) if num_items_x == 1 else (num_items_x, num_items_z)
keys = [f"{LETTERS[len_y-1-j]}{str(i+1).zfill(2)}" for j in range(len_y) for i in range(len_x)]
# 根据 layout 参数生成不同的排序方式
# 注意：物理位置的 y 坐标是倒序的 (row=0 时 y 最大，对应前端显示的顶部)
if layout == "row-major":
# 行优先顺序: A01,A02,A03,A04, B01,B02,B03,B04
# locations[0] 对应 row=0, y最大（前端顶部）→ 应该是 A01
keys = [f"{LETTERS[j]}{i + 1 + col_offset:02d}" for j in range(len_y) for i in range(len_x)]
else:
# 列优先顺序: A01,B01,C01,D01, A02,B02,C02,D02
keys = [f"{LETTERS[j]}{i + 1 + col_offset:02d}" for i in range(len_x) for j in range(len_y)]
sites = {i: site for i, site in zip(keys, _sites.values())}
return WareHouse(
name=name,
size_x=dx + item_dx * num_items_x,
@@ -56,6 +80,7 @@ def YB_warehouse_factory(
num_items_x = num_items_x,
num_items_y = num_items_y,
num_items_z = num_items_z,
ordering_layout=layout, # 传递排序方式到 ordering_layout
# ordered_items=ordered_items,
# ordering=ordering,
sites=sites,
@@ -79,8 +104,9 @@ class WareHouse(ItemizedCarrier):
sites: Optional[Dict[Union[int, str], Optional[ResourcePLR]]] = None,
category: str = "warehouse",
model: Optional[str] = None,
ordering_layout: str = "col-major",
**kwargs
):
super().__init__(
name=name,
size_x=size_x,
@@ -97,6 +123,16 @@ class WareHouse(ItemizedCarrier):
model=model,
)
# 保存排序方式，供graphio.py的坐标映射使用
# 使用独立属性避免与父类的layout冲突
self.ordering_layout = ordering_layout
def serialize(self) -> dict:
"""序列化时保存 ordering_layout 属性"""
data = super().serialize()
data['ordering_layout'] = self.ordering_layout
return data
def get_site_by_layer_position(self, row: int, col: int, layer: int) -> ResourceHolder:
if not (0 <= layer < 4 and 0 <= row < 4 and 0 <= col < 1):
raise ValueError("无效的位置: layer={}, row={}, col={}".format(layer, row, col))
@@ -110,4 +146,4 @@ class WareHouse(ItemizedCarrier):
def get_rack_at_position(self, row: int, col: int, layer: int):
site = self.get_site_by_layer_position(row, col, layer)
return site.resource
return site.resource