workstation_by_hhm

宜宾扣电工站与奔曜配液工站，更新截止10月20日

unilabos/devices/workstation/bioyond_material_management.py

@@ -0,0 +1,374 @@
+"""
+Bioyond物料管理实现
+Bioyond Material Management Implementation
+
+基于Bioyond系统的物料管理，支持从Bioyond系统同步物料到UniLab工作站
+"""
+from typing import Dict, Any, List, Optional, Union
+import json
+import asyncio
+from abc import ABC, abstractmethod
+
+from pylabrobot.resources import (
+    Resource as PLRResource,
+    Container,
+    Deck,
+    Coordinate as PLRCoordinate,
+)
+
+from unilabos.ros.nodes.resource_tracker import DeviceNodeResourceTracker
+from unilabos.utils.log import logger
+from unilabos.resources.graphio import (
+    resource_plr_to_ulab, 
+    resource_ulab_to_plr,
+    resource_bioyond_to_ulab,
+    resource_bioyond_container_to_ulab,
+    resource_ulab_to_bioyond
+)
+from .workstation_material_management import MaterialManagementBase
+
+
+class BioyondMaterialManagement(MaterialManagementBase):
+    """Bioyond物料管理类
+    
+    实现从Bioyond系统同步物料到UniLab工作站的功能：
+    1. 从Bioyond系统获取物料数据
+    2. 转换为UniLab格式
+    3. 同步到PyLabRobot Deck
+    4. 支持双向同步
+    """
+    
+    def __init__(
+        self,
+        device_id: str,
+        deck_config: Dict[str, Any],
+        resource_tracker: DeviceNodeResourceTracker,
+        children_config: Dict[str, Dict[str, Any]] = None,
+        bioyond_config: Dict[str, Any] = None
+    ):
+        self.bioyond_config = bioyond_config or {}
+        self.bioyond_api_client = None
+        self.sync_interval = self.bioyond_config.get("sync_interval", 30)  # 同步间隔（秒）
+        
+        # 初始化父类
+        super().__init__(device_id, deck_config, resource_tracker, children_config)
+        
+        # 初始化Bioyond API客户端
+        self._initialize_bioyond_client()
+        
+        # 启动同步任务
+        self._start_sync_task()
+    
+    def _initialize_bioyond_client(self):
+        """初始化Bioyond API客户端"""
+        try:
+            # 这里应该根据实际的Bioyond API实现
+            # 暂时使用模拟客户端
+            self.bioyond_api_client = BioyondAPIClient(self.bioyond_config)
+            logger.info(f"Bioyond API客户端初始化成功")
+        except Exception as e:
+            logger.error(f"Bioyond API客户端初始化失败: {e}")
+            self.bioyond_api_client = None
+    
+    def _start_sync_task(self):
+        """启动同步任务"""
+        if self.bioyond_api_client:
+            # 创建异步同步任务
+            asyncio.create_task(self._periodic_sync())
+            logger.info(f"Bioyond同步任务已启动，间隔: {self.sync_interval}秒")
+    
+    async def _periodic_sync(self):
+        """定期同步任务"""
+        while True:
+            try:
+                await self.sync_from_bioyond()
+                await asyncio.sleep(self.sync_interval)
+            except Exception as e:
+                logger.error(f"Bioyond同步任务出错: {e}")
+                await asyncio.sleep(self.sync_interval)
+    
+    async def sync_from_bioyond(self) -> bool:
+        """从Bioyond系统同步物料"""
+        try:
+            if not self.bioyond_api_client:
+                logger.warning("Bioyond API客户端未初始化")
+                return False
+            
+            # 1. 从Bioyond获取物料数据
+            bioyond_data = await self.bioyond_api_client.get_materials()
+            if not bioyond_data:
+                logger.warning("从Bioyond获取物料数据为空")
+                return False
+            
+            # 2. 转换为UniLab格式
+            if isinstance(bioyond_data, dict) and "data" in bioyond_data:
+                # 容器格式数据
+                unilab_resources = resource_bioyond_container_to_ulab(bioyond_data)
+            else:
+                # 物料列表格式数据
+                unilab_resources = resource_bioyond_to_ulab(bioyond_data)
+            
+            # 3. 转换为PLR格式并分配到Deck
+            await self._assign_resources_to_deck(unilab_resources)
+            
+            logger.info(f"从Bioyond同步了 {len(unilab_resources)} 个资源")
+            return True
+            
+        except Exception as e:
+            logger.error(f"从Bioyond同步物料失败: {e}")
+            return False
+    
+    async def sync_to_bioyond(self, plr_resource: PLRResource) -> bool:
+        """将本地物料变更同步到Bioyond系统"""
+        try:
+            if not self.bioyond_api_client:
+                logger.warning("Bioyond API客户端未初始化")
+                return False
+            
+            # 1. 转换为UniLab格式
+            unilab_resource = resource_plr_to_ulab(plr_resource)
+            
+            # 2. 转换为Bioyond格式
+            bioyond_materials = resource_ulab_to_bioyond([unilab_resource])
+            
+            # 3. 发送到Bioyond系统
+            success = await self.bioyond_api_client.update_materials(bioyond_materials)
+            
+            if success:
+                logger.info(f"成功同步物料 {plr_resource.name} 到Bioyond")
+            else:
+                logger.warning(f"同步物料 {plr_resource.name} 到Bioyond失败")
+            
+            return success
+            
+        except Exception as e:
+            logger.error(f"同步物料到Bioyond失败: {e}")
+            return False
+    
+    async def _assign_resources_to_deck(self, unilab_resources: List[Dict[str, Any]]):
+        """将UniLab资源分配到Deck"""
+        try:
+            # 转换为PLR格式
+            from unilabos.resources.graphio import list_to_nested_dict
+            nested_resources = list_to_nested_dict(unilab_resources)
+            plr_resources = resource_ulab_to_plr(nested_resources)
+            
+            # 分配资源到Deck
+            if hasattr(plr_resources, 'children'):
+                resources_to_assign = plr_resources.children
+            elif isinstance(plr_resources, list):
+                resources_to_assign = plr_resources
+            else:
+                resources_to_assign = [plr_resources]
+            
+            for resource in resources_to_assign:
+                try:
+                    # 获取资源位置
+                    if hasattr(resource, 'location') and resource.location:
+                        location = PLRCoordinate(resource.location.x, resource.location.y, resource.location.z)
+                    else:
+                        location = PLRCoordinate(0, 0, 0)
+                    
+                    # 分配资源到Deck
+                    self.plr_deck.assign_child_resource(resource, location)
+                    
+                    # 注册到resource tracker
+                    self.resource_tracker.add_resource(resource)
+                    
+                    # 保存资源引用
+                    self.plr_resources[resource.name] = resource
+                    
+                except Exception as e:
+                    logger.error(f"分配资源 {resource.name} 到Deck失败: {e}")
+            
+            logger.info(f"成功分配了 {len(resources_to_assign)} 个资源到Deck")
+            
+        except Exception as e:
+            logger.error(f"分配资源到Deck失败: {e}")
+    
+    def _create_resource_by_type(
+        self, 
+        resource_id: str, 
+        resource_type: str, 
+        config: Dict[str, Any], 
+        data: Dict[str, Any], 
+        location: PLRCoordinate
+    ) -> Optional[PLRResource]:
+        """根据类型创建Bioyond相关资源"""
+        try:
+            # 这里可以根据需要实现特定的Bioyond资源类型
+            # 目前使用通用的容器类型
+            if resource_type in ["container", "plate", "well"]:
+                return self._create_generic_container(resource_id, resource_type, config, data, location)
+            else:
+                logger.warning(f"未知的Bioyond资源类型: {resource_type}")
+                return None
+                
+        except Exception as e:
+            logger.error(f"创建Bioyond资源失败 {resource_id} ({resource_type}): {e}")
+            return None
+    
+    def _create_generic_container(
+        self, 
+        resource_id: str, 
+        resource_type: str, 
+        config: Dict[str, Any], 
+        data: Dict[str, Any], 
+        location: PLRCoordinate
+    ) -> Optional[PLRResource]:
+        """创建通用容器资源"""
+        try:
+            from pylabrobot.resources import Plate, Well
+            
+            if resource_type == "plate":
+                return Plate(
+                    name=resource_id,
+                    size_x=config.get("size_x", 127.76),
+                    size_y=config.get("size_y", 85.48),
+                    size_z=config.get("size_z", 14.35),
+                    location=location,
+                    category="plate"
+                )
+            elif resource_type == "well":
+                return Well(
+                    name=resource_id,
+                    size_x=config.get("size_x", 9.0),
+                    size_y=config.get("size_y", 9.0),
+                    size_z=config.get("size_z", 10.0),
+                    location=location,
+                    category="well"
+                )
+            else:
+                return Container(
+                    name=resource_id,
+                    size_x=config.get("size_x", 50.0),
+                    size_y=config.get("size_y", 50.0),
+                    size_z=config.get("size_z", 10.0),
+                    location=location,
+                    category="container"
+                )
+                
+        except Exception as e:
+            logger.error(f"创建通用容器失败 {resource_id}: {e}")
+            return None
+    
+    def get_bioyond_materials(self) -> List[Dict[str, Any]]:
+        """获取当前Bioyond物料列表"""
+        try:
+            # 将当前PLR资源转换为Bioyond格式
+            bioyond_materials = []
+            for resource in self.plr_resources.values():
+                unilab_resource = resource_plr_to_ulab(resource)
+                bioyond_materials.extend(resource_ulab_to_bioyond([unilab_resource]))
+            return bioyond_materials
+        except Exception as e:
+            logger.error(f"获取Bioyond物料列表失败: {e}")
+            return []
+    
+    def update_material_from_bioyond(self, material_id: str, bioyond_data: Dict[str, Any]) -> bool:
+        """从Bioyond数据更新指定物料"""
+        try:
+            # 查找现有物料
+            material = self.find_material_by_id(material_id)
+            if not material:
+                logger.warning(f"未找到物料: {material_id}")
+                return False
+            
+            # 转换Bioyond数据为UniLab格式
+            unilab_resources = resource_bioyond_to_ulab([bioyond_data])
+            if not unilab_resources:
+                logger.warning(f"转换Bioyond数据失败: {material_id}")
+                return False
+            
+            # 更新物料属性
+            unilab_resource = unilab_resources[0]
+            material.name = unilab_resource.get("name", material.name)
+            
+            # 更新位置
+            position = unilab_resource.get("position", {})
+            if position:
+                material.location = PLRCoordinate(
+                    position.get("x", 0),
+                    position.get("y", 0),
+                    position.get("z", 0)
+                )
+            
+            logger.info(f"成功更新物料: {material_id}")
+            return True
+            
+        except Exception as e:
+            logger.error(f"更新物料失败 {material_id}: {e}")
+            return False
+
+
+class BioyondAPIClient:
+    """Bioyond API客户端（模拟实现）
+    
+    实际使用时需要根据Bioyond系统的API接口实现
+    """
+    
+    def __init__(self, config: Dict[str, Any]):
+        self.config = config
+        self.base_url = config.get("base_url", "http://localhost:8080")
+        self.api_key = config.get("api_key", "")
+        self.timeout = config.get("timeout", 30)
+    
+    async def get_materials(self) -> Optional[Union[Dict[str, Any], List[Dict[str, Any]]]]:
+        """从Bioyond系统获取物料数据"""
+        try:
+            # 这里应该实现实际的API调用
+            # 暂时返回模拟数据
+            logger.info("从Bioyond API获取物料数据")
+            
+            # 模拟API调用延迟
+            await asyncio.sleep(0.1)
+            
+            # 返回模拟数据（实际应该从API获取）
+            return {
+                "data": [],
+                "code": 1,
+                "message": "success",
+                "timestamp": 1234567890
+            }
+            
+        except Exception as e:
+            logger.error(f"Bioyond API调用失败: {e}")
+            return None
+    
+    async def update_materials(self, materials: List[Dict[str, Any]]) -> bool:
+        """更新Bioyond系统中的物料数据"""
+        try:
+            # 这里应该实现实际的API调用
+            logger.info(f"更新Bioyond系统中的 {len(materials)} 个物料")
+            
+            # 模拟API调用延迟
+            await asyncio.sleep(0.1)
+            
+            # 模拟成功响应
+            return True
+            
+        except Exception as e:
+            logger.error(f"更新Bioyond物料失败: {e}")
+            return False
+    
+    async def get_material_by_id(self, material_id: str) -> Optional[Dict[str, Any]]:
+        """根据ID获取单个物料"""
+        try:
+            # 这里应该实现实际的API调用
+            logger.info(f"从Bioyond API获取物料: {material_id}")
+            
+            # 模拟API调用延迟
+            await asyncio.sleep(0.1)
+            
+            # 返回模拟数据
+            return {
+                "id": material_id,
+                "name": f"material_{material_id}",
+                "type": "container",
+                "quantity": 1.0,
+                "unit": "个"
+            }
+            
+        except Exception as e:
+            logger.error(f"获取Bioyond物料失败 {material_id}: {e}")
+            return None