bioyond station with communication init and resource sync

unilabos/devices/workstation/bioyond_studio/station.py

@@ -8,10 +8,89 @@ from typing import Dict, Any, List, Optional, Union
 import json
 
 from unilabos.devices.workstation.workstation_base import WorkstationBase, ResourceSynchronizer
-from unilabos.ros.nodes.resource_tracker import DeviceNodeResourceTracker
+from unilabos.devices.workstation.bioyond_studio.bioyond_rpc import BioyondV1RPC
 from unilabos.utils.log import logger
 from unilabos.resources.graphio import resource_bioyond_to_plr
 
+from .config import API_CONFIG, WORKFLOW_MAPPINGS
+
+
+class BioyondResourceSynchronizer(ResourceSynchronizer):
+    """Bioyond资源同步器
+
+    负责与Bioyond系统进行物料数据的同步
+    """
+
+    def __init__(self, workstation: 'BioyondWorkstation'):
+        super().__init__(workstation)
+        self.bioyond_api_client = None
+        self.sync_interval = 60  # 默认60秒同步一次
+        self.last_sync_time = 0
+
+    def initialize(self) -> bool:
+        """初始化Bioyond资源同步器"""
+        try:
+            self.bioyond_api_client = self.workstation.hardware_interface
+            if self.bioyond_api_client is None:
+                logger.error("Bioyond API客户端未初始化")
+                return False
+
+            # 设置同步间隔
+            self.sync_interval = self.workstation.bioyond_config.get("sync_interval", 60)
+
+            logger.info("Bioyond资源同步器初始化完成")
+            return True
+        except Exception as e:
+            logger.error(f"Bioyond资源同步器初始化失败: {e}")
+            return False
+
+    def sync_from_external(self) -> bool:
+        """从Bioyond系统同步物料数据"""
+        try:
+            if self.bioyond_api_client is None:
+                logger.error("Bioyond API客户端未初始化")
+                return False
+
+            bioyond_data = self.bioyond_api_client.fetch_materials()
+            if not bioyond_data:
+                logger.warning("从Bioyond获取的物料数据为空")
+                return False
+
+            # 转换为UniLab格式
+            unilab_resources = resource_bioyond_to_plr(bioyond_data, deck=self.workstation.deck)
+
+            logger.info(f"从Bioyond同步了 {len(unilab_resources)} 个资源")
+            return True
+        except Exception as e:
+            logger.error(f"从Bioyond同步物料数据失败: {e}")
+            return False
+
+    def sync_to_external(self, resource: Any) -> bool:
+        """将本地物料数据变更同步到Bioyond系统"""
+        try:
+            if self.bioyond_api_client is None:
+                logger.error("Bioyond API客户端未初始化")
+                return False
+
+            # 调用入库、出库操作
+            # bioyond_format_data = self._convert_resource_to_bioyond_format(resource)
+            # success = await self.bioyond_api_client.update_material(bioyond_format_data)
+            #
+            # if success
+        except:
+            pass
+
+    def handle_external_change(self, change_info: Dict[str, Any]) -> bool:
+        """处理Bioyond系统的变更通知"""
+        try:
+            # 这里可以实现对Bioyond变更的处理逻辑
+            logger.info(f"处理Bioyond变更通知: {change_info}")
+
+            return True
+        except Exception as e:
+            logger.error(f"处理Bioyond变更通知失败: {e}")
+            return False
+
 
 class BioyondWorkstation(WorkstationBase):
     """Bioyond工作站
@@ -26,15 +105,7 @@ class BioyondWorkstation(WorkstationBase):
         *args,
         **kwargs,
     ):
-        # 设置Bioyond配置
+        self._create_communication_module(bioyond_config)
-        self.bioyond_config = bioyond_config or {
-            "base_url": "http://localhost:8080",
-            "api_key": "",
-            "sync_interval": 30,
-            "timeout": 30
-        }
-        
-        # 设置默认deck配置
 
         # 初始化父类
         super().__init__(
@@ -43,24 +114,21 @@ class BioyondWorkstation(WorkstationBase):
             *args,
             **kwargs,
         )
+        self.resource_synchronizer = BioyondResourceSynchronizer(self)
+        self.resource_synchronizer.sync_from_external()
         
         # TODO: self._ros_node里面拿属性
         logger.info(f"Bioyond工作站初始化完成")
 
-    def _create_communication_module(self):
+    def _create_communication_module(self, config: Optional[Dict[str, Any]] = None) -> None:
         """创建Bioyond通信模块"""
-        # 暂时返回None，因为工作站基类没有强制要求通信模块
+        self.bioyond_config = config or {
+            **API_CONFIG,
+            "workflow_mappings": WORKFLOW_MAPPINGS
+        }
+        self.hardware_interface = BioyondV1RPC(self.bioyond_config)
         return None
     
-    def _create_material_management_module(self) -> BioyondMaterialManagement:
-        """创建Bioyond物料管理模块"""
-        # 获取必要的属性，如果不存在则使用默认值
-        device_id = getattr(self, 'device_id', 'bioyond_workstation')
-        resource_tracker = getattr(self, 'resource_tracker', None)
-        children_config = getattr(self, '_children', {})
-        
-       
-    
     def _register_supported_workflows(self):
         """注册Bioyond支持的工作流"""
         from unilabos.devices.workstation.workstation_base import WorkflowInfo

unilabos/devices/workstation/workstation_base.py

@@ -112,17 +112,17 @@ class ResourceSynchronizer(ABC):
         self.workstation = workstation
 
     @abstractmethod
-    async def sync_from_external(self) -> bool:
+    def sync_from_external(self) -> bool:
         """从外部系统同步物料到本地deck"""
         pass
 
     @abstractmethod
-    async def sync_to_external(self, plr_resource: PLRResource) -> bool:
+    def sync_to_external(self, plr_resource: PLRResource) -> bool:
         """将本地物料同步到外部系统"""
         pass
 
     @abstractmethod
-    async def handle_external_change(self, change_info: Dict[str, Any]) -> bool:
+    def handle_external_change(self, change_info: Dict[str, Any]) -> bool:
         """处理外部系统的变更通知"""
         pass
 
@@ -147,17 +147,15 @@ class WorkstationBase(ABC):
 
     def __init__(
         self,
-        deck: PLRResource,
+        deck: Deck,
         *args,
         **kwargs,  # 必须有kwargs
     ):
-        # 基本配置
-        print(deck)
-        self.deck_config = deck
-
         # PLR 物料系统
-        self.deck: Optional[Deck] = None
+        self.deck: Optional[Deck] = deck
         self.plr_resources: Dict[str, PLRResource] = {}
+
+        self.resource_synchronizer = None  # type: Optional[ResourceSynchronizer]
         # 硬件接口
         self.hardware_interface: Union[Any, str] = None
 
@@ -173,46 +171,7 @@ class WorkstationBase(ABC):
     def post_init(self, ros_node: ROS2WorkstationNode) -> None:
         # 初始化物料系统
         self._ros_node = ros_node
-        self._initialize_material_system()
+        self._ros_node.update_resource([self.deck])
-
-    def _initialize_material_system(self):
-        """初始化物料系统 - 使用 graphio 转换"""
-        pass
-
-    def _create_complete_resource_config(self) -> Dict[str, Any]:
-        """创建完整的资源配置 - 合并 deck_config 和 children"""
-        # 创建主 deck 配置
-        return {}
-
-    def _normalize_child_resource(self, resource_id: str, config: Dict[str, Any], parent_id: str) -> Dict[str, Any]:
-        """标准化子资源配置"""
-        return {
-            "id": resource_id,
-            "name": config.get("name", resource_id),
-            "type": config.get("type", "container"),
-            "position": self._normalize_position(config.get("position", {})),
-            "config": config.get("config", {}),
-            "data": config.get("data", {}),
-            "children": [],  # 简化版本：只支持一层子资源
-            "parent": parent_id,
-        }
-
-    def _normalize_position(self, position: Any) -> Dict[str, float]:
-        """标准化位置信息"""
-        if isinstance(position, dict):
-            return {
-                "x": float(position.get("x", 0)),
-                "y": float(position.get("y", 0)),
-                "z": float(position.get("z", 0)),
-            }
-        elif isinstance(position, (list, tuple)) and len(position) >= 2:
-            return {
-                "x": float(position[0]),
-                "y": float(position[1]),
-                "z": float(position[2]) if len(position) > 2 else 0.0,
-            }
-        else:
-            return {"x": 0.0, "y": 0.0, "z": 0.0}
 
     def _build_resource_mappings(self, deck: Deck):
         """递归构建资源映射"""
@@ -296,14 +255,14 @@ class WorkstationBase(ABC):
         """按类型查找资源"""
         return [res for res in self.plr_resources.values() if isinstance(res, resource_type)]
 
-    async def sync_with_external_system(self) -> bool:
+    def sync_with_external_system(self) -> bool:
         """与外部物料系统同步"""
         if not self.resource_synchronizer:
             logger.info(f"工作站 {self._ros_node.device_id} 没有配置资源同步器")
             return True
 
         try:
-            success = await self.resource_synchronizer.sync_from_external()
+            success = self.resource_synchronizer.sync_from_external()
             if success:
                 logger.info(f"工作站 {self._ros_node.device_id} 外部同步成功")
             else: