create warehouse by factory func

unilabos/devices/workstation/bioyond_studio/station.py

@@ -4,15 +4,20 @@ Bioyond Workstation Implementation
 
 集成Bioyond物料管理的工作站示例
 """
+import traceback
 from typing import Dict, Any, List, Optional, Union
 import json
 
 from unilabos.devices.workstation.workstation_base import WorkstationBase, ResourceSynchronizer
 from unilabos.devices.workstation.bioyond_studio.bioyond_rpc import BioyondV1RPC
+from unilabos.resources.warehouse import WareHouse
 from unilabos.utils.log import logger
 from unilabos.resources.graphio import resource_bioyond_to_plr
 
-from unilabos.devices.workstation.bioyond_studio.config import API_CONFIG, WORKFLOW_MAPPINGS
+from unilabos.ros.nodes.base_device_node import ROS2DeviceNode, BaseROS2DeviceNode
+from unilabos.ros.nodes.presets.workstation import ROS2WorkstationNode
+
+from unilabos.devices.workstation.bioyond_studio.config import API_CONFIG, WORKFLOW_MAPPINGS, MATERIAL_TYPE_MAPPINGS
 
 
 class BioyondResourceSynchronizer(ResourceSynchronizer):
@@ -26,6 +31,7 @@ class BioyondResourceSynchronizer(ResourceSynchronizer):
         self.bioyond_api_client = None
         self.sync_interval = 60  # 默认60秒同步一次
         self.last_sync_time = 0
+        self.initialize()
 
     def initialize(self) -> bool:
         """初始化Bioyond资源同步器"""
@@ -36,7 +42,7 @@ class BioyondResourceSynchronizer(ResourceSynchronizer):
                 return False
 
             # 设置同步间隔
-            self.sync_interval = self.workstation.bioyond_config.get("sync_interval", 60)
+            self.sync_interval = self.workstation.bioyond_config.get("sync_interval", 600)
 
             logger.info("Bioyond资源同步器初始化完成")
             return True
@@ -51,18 +57,19 @@ class BioyondResourceSynchronizer(ResourceSynchronizer):
                 logger.error("Bioyond API客户端未初始化")
                 return False
 
-            bioyond_data = self.bioyond_api_client.fetch_materials()
+            bioyond_data = self.bioyond_api_client.stock_material('{"typeMode": 2, "includeDetail": true}')
             if not bioyond_data:
                 logger.warning("从Bioyond获取的物料数据为空")
                 return False
 
             # 转换为UniLab格式
-            unilab_resources = resource_bioyond_to_plr(bioyond_data, deck=self.workstation.deck)
+            unilab_resources = resource_bioyond_to_plr(bioyond_data, type_mapping=self.workstation.bioyond_config["material_type_mappings"], deck=self.workstation.deck)
 
             logger.info(f"从Bioyond同步了 {len(unilab_resources)} 个资源")
             return True
         except Exception as e:
             logger.error(f"从Bioyond同步物料数据失败: {e}")
+            traceback.print_exc()
             return False
 
     def sync_to_external(self, resource: Any) -> bool:
@@ -105,8 +112,6 @@ class BioyondWorkstation(WorkstationBase):
         *args,
         **kwargs,
     ):
-        self._create_communication_module(bioyond_config)
-
         # 初始化父类
         super().__init__(
             # 桌子
@@ -114,19 +119,34 @@ class BioyondWorkstation(WorkstationBase):
             *args,
             **kwargs,
         )
+        self.deck.warehouses = {}
+        for resource in self.deck.children:
+            if isinstance(resource, WareHouse):
+                self.deck.warehouses[resource.name] = resource
+        
+        self._create_communication_module(bioyond_config)
         self.resource_synchronizer = BioyondResourceSynchronizer(self)
         self.resource_synchronizer.sync_from_external()
         
         # TODO: self._ros_node里面拿属性
         logger.info(f"Bioyond工作站初始化完成")
+    
+    def post_init(self, ros_node: ROS2WorkstationNode):
+        self._ros_node = ros_node
+        #self.deck = create_a_coin_cell_deck()
+        ROS2DeviceNode.run_async_func(self._ros_node.update_resource, True, **{
+            "resources": [self.deck]
+        })
 
     def _create_communication_module(self, config: Optional[Dict[str, Any]] = None) -> None:
         """创建Bioyond通信模块"""
         self.bioyond_config = config or {
             **API_CONFIG,
-            "workflow_mappings": WORKFLOW_MAPPINGS
+            "workflow_mappings": WORKFLOW_MAPPINGS,
+            "material_type_mappings": MATERIAL_TYPE_MAPPINGS
         }
         self.hardware_interface = BioyondV1RPC(self.bioyond_config)
+        
         return None
     
     def _register_supported_workflows(self):

unilabos/registry/resources/bioyond/deck.yaml

@@ -0,0 +1,12 @@
+BIOYOND_PolymerReactionStation_Deck:
+  category:
+  - deck
+  class:
+    module: unilabos.resources.bioyond.decks:BIOYOND_PolymerReactionStation_Deck
+    type: pylabrobot
+  description: BIOYOND PolymerReactionStation Deck
+  handles: []
+  icon: ''
+  init_param_schema: {}
+  registry_type: resource
+  version: 1.0.0

unilabos/resources/bioyond/decks.py

@@ -1,12 +1,21 @@
-from pylabrobot.resources import Deck, Coordinate
+from pylabrobot.resources import Deck, Coordinate, Rotation
 
 from unilabos.resources.bioyond.warehouses import bioyond_warehouse_1x4x4, bioyond_warehouse_1x4x2, bioyond_warehouse_liquid_and_lid_handling
 
 
 class BIOYOND_PolymerReactionStation_Deck(Deck):
-    def __init__(self, name: str = "PolymerReactionStation_Deck") -> None:
+    def __init__(
+        self, 
+        name: str = "PolymerReactionStation_Deck",
+        size_x: float = 2700.0,
+        size_y: float = 1080.0,
+        size_z: float = 1500.0,
+        category: str = "deck",
+        setup: bool = False
+    ) -> None:
         super().__init__(name=name, size_x=2700.0, size_y=1080.0, size_z=1500.0)
-        self.setup()
+        if setup:
+            self.setup()
 
     def setup(self) -> None:
         # 添加仓库
@@ -20,7 +29,7 @@ class BIOYOND_PolymerReactionStation_Deck(Deck):
             "堆栈2": Coordinate(2550.0, 650.0, 0.0),
             "站内试剂存放堆栈": Coordinate(800.0, 475.0, 0.0),
         }
-        self.warehouses["站内试剂存放堆栈"].rotation = 90.0
+        self.warehouses["站内试剂存放堆栈"].rotation = Rotation(z=90)
 
         for warehouse_name, warehouse in self.warehouses.items():
             self.assign_child_resource(warehouse, location=self.warehouse_locations[warehouse_name])

unilabos/resources/bioyond/warehouses.py

@@ -1,9 +1,9 @@
-from unilabos.resources.warehouse import WareHouse
+from unilabos.resources.warehouse import WareHouse, warehouse_factory
 
 
 def bioyond_warehouse_1x4x4(name: str) -> WareHouse:
     """创建BioYond 4x1x4仓库"""
-    return WareHouse(
+    return warehouse_factory(
         name=name,
         num_items_x=1,
         num_items_y=4,
@@ -20,7 +20,7 @@ def bioyond_warehouse_1x4x4(name: str) -> WareHouse:
 
 def bioyond_warehouse_1x4x2(name: str) -> WareHouse:
     """创建BioYond 4x1x2仓库"""
-    return WareHouse(
+    return warehouse_factory(
         name=name,
         num_items_x=1,
         num_items_y=4,
@@ -38,7 +38,7 @@ def bioyond_warehouse_1x4x2(name: str) -> WareHouse:
 
 def bioyond_warehouse_liquid_and_lid_handling(name: str) -> WareHouse:
     """创建BioYond开关盖加液模块台面"""
-    return WareHouse(
+    return warehouse_factory(
         name=name,
         num_items_x=2,
         num_items_y=5,

unilabos/resources/itemized_carrier.py

@@ -71,6 +71,9 @@ class ItemizedCarrier(ResourcePLR):
     size_x: float,
     size_y: float,
     size_z: float,
+    num_items_x: int = 0,
+    num_items_y: int = 0,
+    num_items_z: int = 0,
     sites: Optional[Dict[Union[int, str], Optional[ResourcePLR]]] = None,
     category: Optional[str] = "carrier",
     model: Optional[str] = None,
@@ -83,18 +86,27 @@ class ItemizedCarrier(ResourcePLR):
       category=category,
       model=model,
     )
-    sites = sites or {}
-    self.sites: List[Optional[ResourcePLR]] = list(sites.values())
-    self._ordering = sites
-    self.num_items = len(self.sites)
-    self.child_locations: Dict[str, Coordinate] = {}
-    for spot, resource in sites.items():
-      if resource is not None and getattr(resource, "location", None) is None:
-        raise ValueError(f"resource {resource} has no location")
-      if resource is not None:
-        self.child_locations[spot] = resource.location
-      else:
-        self.child_locations[spot] = Coordinate.zero()
+    self.num_items = len(sites)
+    self.num_items_x, self.num_items_y, self.num_items_z = num_items_x, num_items_y, num_items_z
+    if isinstance(sites, dict):
+      sites = sites or {}
+      self.sites: List[Optional[ResourcePLR]] = list(sites.values())
+      self._ordering = sites
+      self.child_locations: Dict[str, Coordinate] = {}
+      for spot, resource in sites.items():
+        if resource is not None and getattr(resource, "location", None) is None:
+          raise ValueError(f"resource {resource} has no location")
+        if resource is not None:
+          self.child_locations[spot] = resource.location
+        else:
+          self.child_locations[spot] = Coordinate.zero()
+    elif isinstance(sites, list):
+      # deserialize时走这里；还需要根据 self.sites 索引children
+      self.child_locations = {site["label"]: Coordinate(**site["position"]) for site in sites}
+      self.sites = [site["occupied_by"] for site in sites]
+      self._ordering = {site["label"]: site["position"] for site in sites}
+    else:
+      print("sites:", sites)
 
   @property
   def capacity(self):
@@ -112,7 +124,20 @@ class ItemizedCarrier(ResourcePLR):
     reassign: bool = True,
     spot: Optional[int] = None,
   ):
-    idx = spot if spot is not None else len(self.sites)
+    idx = spot
+    # 如果只给 location，根据坐标和 deserialize 后的 self.sites（持有names）来寻找 resource 该摆放的位置
+    if spot is not None:
+      idx = spot
+    else:
+      for i, site in enumerate(self.sites):
+        site_location = list(self.child_locations.values())[i]
+        if type(site) == str and site == resource.name:
+          idx = i
+          break
+        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)
@@ -288,9 +313,15 @@ class ItemizedCarrier(ResourcePLR):
   def serialize(self):
     return {
       **super().serialize(),
-      "slots": [{
+      "num_items_x": self.num_items_x,
+      "num_items_y": self.num_items_y,
+      "num_items_z": self.num_items_z,
+      "sites": [{
         "label": str(identifier),
         "visible": True if self[identifier] is not None else False,
+        "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": {"width": self._size_x, "height": self._size_y, "depth": self._size_z},
         "content_type": ["bottle", "container", "tube", "bottle_carrier", "tip_rack"]

unilabos/resources/warehouse.py

@@ -1,63 +1,88 @@
-from typing import Optional, List
+from typing import Dict, Optional, List, Union
 from pylabrobot.resources import Coordinate
 from pylabrobot.resources.carrier import ResourceHolder, create_homogeneous_resources
 
-from unilabos.resources.itemized_carrier import ItemizedCarrier
+from unilabos.resources.itemized_carrier import ItemizedCarrier, ResourcePLR
+
+
+def warehouse_factory(
+    name: str,
+    num_items_x: int = 4,
+    num_items_y: int = 1,
+    num_items_z: int = 4,
+    dx: float = 137.0,
+    dy: float = 96.0,
+    dz: float = 120.0,
+    item_dx: float = 10.0,
+    item_dy: float = 10.0,
+    item_dz: float = 10.0,
+    removed_positions: Optional[List[int]] = None,
+    empty: bool = False,
+    category: str = "warehouse",
+    model: Optional[str] = None,
+):
+    # 创建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个位置)
+                # 计算位置
+                x = dx + col * item_dx
+                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:
+        locations = [loc for i, loc in enumerate(locations) if i not in removed_positions]
+    sites = create_homogeneous_resources(
+        klass=ResourceHolder,
+        locations=locations,
+        resource_size_x=127.0,
+        resource_size_y=86.0,
+        name_prefix=name,
+    )
+    
+    return WareHouse(
+        name=name,
+        size_x=dx + item_dx * num_items_x,
+        size_y=dy + item_dy * num_items_y,
+        size_z=dz + item_dz * num_items_z,
+        num_items_x = num_items_x,
+        num_items_y = num_items_y,
+        num_items_z = num_items_z,
+        # ordered_items=ordered_items,
+        # ordering=ordering,
+        sites=sites,
+        category=category,
+        model=model,
+    )
 
 
 class WareHouse(ItemizedCarrier):
-    """4x4x1堆栈载体类 - 可容纳16个板位的载体（4层x4行x1列）"""
-
+    """堆栈载体类 - 可容纳16个板位的载体（4层x4行x1列）"""
     def __init__(
         self,
         name: str,
-        num_items_x: int = 4,
-        num_items_y: int = 1,
-        num_items_z: int = 4,
-        dx: float = 137.0,
-        dy: float = 96.0,
-        dz: float = 120.0,
-        item_dx: float = 10.0,
-        item_dy: float = 10.0,
-        item_dz: float = 10.0,
-        removed_positions: Optional[List[int]] = None,
-        empty: bool = False,
+        size_x: float,
+        size_y: float,
+        size_z: float,
+        num_items_x: int,
+        num_items_y: int,
+        num_items_z: int,
+        sites: Optional[Dict[Union[int, str], Optional[ResourcePLR]]] = None,
         category: str = "warehouse",
         model: Optional[str] = None,
     ):
-        self.num_items_x = num_items_x
-        self.num_items_y = num_items_y
-        self.num_items_z = num_items_z
-        # 创建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个位置)
-                    # 计算位置
-                    x = dx + col * item_dx
-                    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:
-            locations = [loc for i, loc in enumerate(locations) if i not in removed_positions]
-
-        sites = create_homogeneous_resources(
-            klass=ResourceHolder,
-            locations=locations,
-            resource_size_x=127.0,
-            resource_size_y=86.0,
-            name_prefix=name,
-        )
 
         super().__init__(
             name=name,
-            size_x=dx + item_dx * num_items_x,
-            size_y=dy + item_dy * num_items_y,
-            size_z=dz + item_dz * num_items_z,
+            size_x=size_x,
+            size_y=size_y,
+            size_z=size_z,
             # ordered_items=ordered_items,
             # ordering=ordering,
+            num_items_x=num_items_x,
+            num_items_y=num_items_y,
+            num_items_z=num_items_z,
             sites=sites,
             category=category,
             model=model,