mirror of
https://github.com/dptech-corp/Uni-Lab-OS.git
synced 2025-12-17 04:51:10 +00:00
584 lines
22 KiB
Python
584 lines
22 KiB
Python
"""
|
|
工作站物料管理基类
|
|
Workstation Material Management Base Class
|
|
|
|
基于PyLabRobot的物料管理系统
|
|
"""
|
|
from typing import Dict, Any, List, Optional, Union, Type
|
|
from abc import ABC, abstractmethod
|
|
import json
|
|
|
|
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
|
|
|
|
|
|
class MaterialManagementBase(ABC):
|
|
"""物料管理基类
|
|
|
|
定义工作站物料管理的标准接口：
|
|
1. 物料初始化 - 根据配置创建物料资源
|
|
2. 物料追踪 - 实时跟踪物料位置和状态
|
|
3. 物料查找 - 按类型、位置、状态查找物料
|
|
4. 物料转换 - PyLabRobot与UniLab资源格式转换
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
device_id: str,
|
|
deck_config: Dict[str, Any],
|
|
resource_tracker: DeviceNodeResourceTracker,
|
|
children_config: Dict[str, Dict[str, Any]] = None
|
|
):
|
|
self.device_id = device_id
|
|
self.deck_config = deck_config
|
|
self.resource_tracker = resource_tracker
|
|
self.children_config = children_config or {}
|
|
|
|
# 创建主台面
|
|
self.plr_deck = self._create_deck()
|
|
|
|
# 扩展ResourceTracker
|
|
self._extend_resource_tracker()
|
|
|
|
# 注册deck到resource tracker
|
|
self.resource_tracker.add_resource(self.plr_deck)
|
|
|
|
# 初始化子资源
|
|
self.plr_resources = {}
|
|
self._initialize_materials()
|
|
|
|
def _create_deck(self) -> Deck:
|
|
"""创建主台面"""
|
|
return Deck(
|
|
name=f"{self.device_id}_deck",
|
|
size_x=self.deck_config.get("size_x", 1000.0),
|
|
size_y=self.deck_config.get("size_y", 1000.0),
|
|
size_z=self.deck_config.get("size_z", 500.0),
|
|
origin=PLRCoordinate(0, 0, 0)
|
|
)
|
|
|
|
def _extend_resource_tracker(self):
|
|
"""扩展ResourceTracker以支持PyLabRobot特定功能"""
|
|
|
|
def find_by_type(resource_type):
|
|
"""按类型查找资源"""
|
|
return self._find_resources_by_type_recursive(self.plr_deck, resource_type)
|
|
|
|
def find_by_category(category: str):
|
|
"""按类别查找资源"""
|
|
found = []
|
|
for resource in self._get_all_resources():
|
|
if hasattr(resource, 'category') and resource.category == category:
|
|
found.append(resource)
|
|
return found
|
|
|
|
def find_by_name_pattern(pattern: str):
|
|
"""按名称模式查找资源"""
|
|
import re
|
|
found = []
|
|
for resource in self._get_all_resources():
|
|
if re.search(pattern, resource.name):
|
|
found.append(resource)
|
|
return found
|
|
|
|
# 动态添加方法到resource_tracker
|
|
self.resource_tracker.find_by_type = find_by_type
|
|
self.resource_tracker.find_by_category = find_by_category
|
|
self.resource_tracker.find_by_name_pattern = find_by_name_pattern
|
|
|
|
def _find_resources_by_type_recursive(self, resource, target_type):
|
|
"""递归查找指定类型的资源"""
|
|
found = []
|
|
if isinstance(resource, target_type):
|
|
found.append(resource)
|
|
|
|
# 递归查找子资源
|
|
children = getattr(resource, "children", [])
|
|
for child in children:
|
|
found.extend(self._find_resources_by_type_recursive(child, target_type))
|
|
|
|
return found
|
|
|
|
def _get_all_resources(self) -> List[PLRResource]:
|
|
"""获取所有资源"""
|
|
all_resources = []
|
|
|
|
def collect_resources(resource):
|
|
all_resources.append(resource)
|
|
children = getattr(resource, "children", [])
|
|
for child in children:
|
|
collect_resources(child)
|
|
|
|
collect_resources(self.plr_deck)
|
|
return all_resources
|
|
|
|
def _initialize_materials(self):
|
|
"""初始化物料"""
|
|
try:
|
|
# 确定创建顺序，确保父资源先于子资源创建
|
|
creation_order = self._determine_creation_order()
|
|
|
|
# 按顺序创建资源
|
|
for resource_id in creation_order:
|
|
config = self.children_config[resource_id]
|
|
self._create_plr_resource(resource_id, config)
|
|
|
|
logger.info(f"物料管理系统初始化完成，共创建 {len(self.plr_resources)} 个资源")
|
|
|
|
except Exception as e:
|
|
logger.error(f"物料初始化失败: {e}")
|
|
|
|
def _determine_creation_order(self) -> List[str]:
|
|
"""确定资源创建顺序"""
|
|
order = []
|
|
visited = set()
|
|
|
|
def visit(resource_id: str):
|
|
if resource_id in visited:
|
|
return
|
|
visited.add(resource_id)
|
|
|
|
config = self.children_config.get(resource_id, {})
|
|
parent_id = config.get("parent")
|
|
|
|
# 如果有父资源，先访问父资源
|
|
if parent_id and parent_id in self.children_config:
|
|
visit(parent_id)
|
|
|
|
order.append(resource_id)
|
|
|
|
for resource_id in self.children_config:
|
|
visit(resource_id)
|
|
|
|
return order
|
|
|
|
def _create_plr_resource(self, resource_id: str, config: Dict[str, Any]):
|
|
"""创建PyLabRobot资源"""
|
|
try:
|
|
resource_type = config.get("type", "unknown")
|
|
data = config.get("data", {})
|
|
location_config = config.get("location", {})
|
|
|
|
# 创建位置坐标
|
|
location = PLRCoordinate(
|
|
x=location_config.get("x", 0.0),
|
|
y=location_config.get("y", 0.0),
|
|
z=location_config.get("z", 0.0)
|
|
)
|
|
|
|
# 根据类型创建资源
|
|
resource = self._create_resource_by_type(resource_id, resource_type, config, data, location)
|
|
|
|
if resource:
|
|
# 设置父子关系
|
|
parent_id = config.get("parent")
|
|
if parent_id and parent_id in self.plr_resources:
|
|
parent_resource = self.plr_resources[parent_id]
|
|
parent_resource.assign_child_resource(resource, location)
|
|
else:
|
|
# 直接放在deck上
|
|
self.plr_deck.assign_child_resource(resource, location)
|
|
|
|
# 保存资源引用
|
|
self.plr_resources[resource_id] = resource
|
|
|
|
# 注册到resource tracker
|
|
self.resource_tracker.add_resource(resource)
|
|
|
|
logger.debug(f"创建资源成功: {resource_id} ({resource_type})")
|
|
|
|
except Exception as e:
|
|
logger.error(f"创建资源失败 {resource_id}: {e}")
|
|
|
|
@abstractmethod
|
|
def _create_resource_by_type(
|
|
self,
|
|
resource_id: str,
|
|
resource_type: str,
|
|
config: Dict[str, Any],
|
|
data: Dict[str, Any],
|
|
location: PLRCoordinate
|
|
) -> Optional[PLRResource]:
|
|
"""根据类型创建资源 - 子类必须实现"""
|
|
pass
|
|
|
|
# ============ 物料查找接口 ============
|
|
|
|
def find_materials_by_type(self, material_type: str) -> List[PLRResource]:
|
|
"""按材料类型查找物料"""
|
|
return self.resource_tracker.find_by_category(material_type)
|
|
|
|
def find_material_by_id(self, resource_id: str) -> Optional[PLRResource]:
|
|
"""按ID查找物料"""
|
|
return self.plr_resources.get(resource_id)
|
|
|
|
def find_available_positions(self, position_type: str) -> List[PLRResource]:
|
|
"""查找可用位置"""
|
|
positions = self.resource_tracker.find_by_category(position_type)
|
|
available = []
|
|
|
|
for pos in positions:
|
|
if hasattr(pos, 'is_available') and pos.is_available():
|
|
available.append(pos)
|
|
elif hasattr(pos, 'children') and len(pos.children) == 0:
|
|
available.append(pos)
|
|
|
|
return available
|
|
|
|
def get_material_inventory(self) -> Dict[str, int]:
|
|
"""获取物料库存统计"""
|
|
inventory = {}
|
|
|
|
for resource in self._get_all_resources():
|
|
if hasattr(resource, 'category'):
|
|
category = resource.category
|
|
inventory[category] = inventory.get(category, 0) + 1
|
|
|
|
return inventory
|
|
|
|
# ============ 物料状态更新接口 ============
|
|
|
|
def update_material_location(self, material_id: str, new_location: PLRCoordinate) -> bool:
|
|
"""更新物料位置"""
|
|
try:
|
|
material = self.find_material_by_id(material_id)
|
|
if material:
|
|
material.location = new_location
|
|
return True
|
|
return False
|
|
except Exception as e:
|
|
logger.error(f"更新物料位置失败: {e}")
|
|
return False
|
|
|
|
def move_material(self, material_id: str, target_container_id: str) -> bool:
|
|
"""移动物料到目标容器"""
|
|
try:
|
|
material = self.find_material_by_id(material_id)
|
|
target = self.find_material_by_id(target_container_id)
|
|
|
|
if material and target:
|
|
# 从原位置移除
|
|
if material.parent:
|
|
material.parent.unassign_child_resource(material)
|
|
|
|
# 添加到新位置
|
|
target.assign_child_resource(material)
|
|
return True
|
|
|
|
return False
|
|
|
|
except Exception as e:
|
|
logger.error(f"移动物料失败: {e}")
|
|
return False
|
|
|
|
# ============ 资源转换接口 ============
|
|
|
|
def convert_to_unilab_format(self, plr_resource: PLRResource) -> Dict[str, Any]:
|
|
"""将PyLabRobot资源转换为UniLab格式"""
|
|
return resource_plr_to_ulab(plr_resource)
|
|
|
|
def convert_from_unilab_format(self, unilab_resource: Dict[str, Any]) -> PLRResource:
|
|
"""将UniLab格式转换为PyLabRobot资源"""
|
|
return resource_ulab_to_plr(unilab_resource)
|
|
|
|
def get_deck_state(self) -> Dict[str, Any]:
|
|
"""获取Deck状态"""
|
|
try:
|
|
return {
|
|
"deck_info": {
|
|
"name": self.plr_deck.name,
|
|
"size": {
|
|
"x": self.plr_deck.size_x,
|
|
"y": self.plr_deck.size_y,
|
|
"z": self.plr_deck.size_z
|
|
},
|
|
"children_count": len(self.plr_deck.children)
|
|
},
|
|
"resources": {
|
|
resource_id: self.convert_to_unilab_format(resource)
|
|
for resource_id, resource in self.plr_resources.items()
|
|
},
|
|
"inventory": self.get_material_inventory()
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"获取Deck状态失败: {e}")
|
|
return {"error": str(e)}
|
|
|
|
# ============ 数据持久化接口 ============
|
|
|
|
def save_state_to_file(self, file_path: str) -> bool:
|
|
"""保存状态到文件"""
|
|
try:
|
|
state = self.get_deck_state()
|
|
with open(file_path, 'w', encoding='utf-8') as f:
|
|
json.dump(state, f, indent=2, ensure_ascii=False)
|
|
logger.info(f"状态已保存到: {file_path}")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"保存状态失败: {e}")
|
|
return False
|
|
|
|
def load_state_from_file(self, file_path: str) -> bool:
|
|
"""从文件加载状态"""
|
|
try:
|
|
with open(file_path, 'r', encoding='utf-8') as f:
|
|
state = json.load(f)
|
|
|
|
# 重新创建资源
|
|
self._recreate_resources_from_state(state)
|
|
logger.info(f"状态已从文件加载: {file_path}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"加载状态失败: {e}")
|
|
return False
|
|
|
|
def _recreate_resources_from_state(self, state: Dict[str, Any]):
|
|
"""从状态重新创建资源"""
|
|
# 清除现有资源
|
|
self.plr_resources.clear()
|
|
self.plr_deck.children.clear()
|
|
|
|
# 从状态重新创建
|
|
resources_data = state.get("resources", {})
|
|
for resource_id, resource_data in resources_data.items():
|
|
try:
|
|
plr_resource = self.convert_from_unilab_format(resource_data)
|
|
self.plr_resources[resource_id] = plr_resource
|
|
self.plr_deck.assign_child_resource(plr_resource)
|
|
except Exception as e:
|
|
logger.error(f"重新创建资源失败 {resource_id}: {e}")
|
|
|
|
|
|
class CoinCellMaterialManagement(MaterialManagementBase):
|
|
"""纽扣电池物料管理类
|
|
|
|
从 button_battery_station 抽取的物料管理功能
|
|
"""
|
|
|
|
def _create_resource_by_type(
|
|
self,
|
|
resource_id: str,
|
|
resource_type: str,
|
|
config: Dict[str, Any],
|
|
data: Dict[str, Any],
|
|
location: PLRCoordinate
|
|
) -> Optional[PLRResource]:
|
|
"""根据类型创建纽扣电池相关资源"""
|
|
|
|
# 导入纽扣电池资源类
|
|
from unilabos.device_comms.button_battery_station import (
|
|
MaterialPlate, PlateSlot, ClipMagazine, BatteryPressSlot,
|
|
TipBox64, WasteTipBox, BottleRack, Battery, ElectrodeSheet
|
|
)
|
|
|
|
try:
|
|
if resource_type == "material_plate":
|
|
return self._create_material_plate(resource_id, config, data, location)
|
|
|
|
elif resource_type == "plate_slot":
|
|
return self._create_plate_slot(resource_id, config, data, location)
|
|
|
|
elif resource_type == "clip_magazine":
|
|
return self._create_clip_magazine(resource_id, config, data, location)
|
|
|
|
elif resource_type == "battery_press_slot":
|
|
return self._create_battery_press_slot(resource_id, config, data, location)
|
|
|
|
elif resource_type == "tip_box":
|
|
return self._create_tip_box(resource_id, config, data, location)
|
|
|
|
elif resource_type == "waste_tip_box":
|
|
return self._create_waste_tip_box(resource_id, config, data, location)
|
|
|
|
elif resource_type == "bottle_rack":
|
|
return self._create_bottle_rack(resource_id, config, data, location)
|
|
|
|
elif resource_type == "battery":
|
|
return self._create_battery(resource_id, config, data, location)
|
|
|
|
else:
|
|
logger.warning(f"未知的资源类型: {resource_type}")
|
|
return None
|
|
|
|
except Exception as e:
|
|
logger.error(f"创建资源失败 {resource_id} ({resource_type}): {e}")
|
|
return None
|
|
|
|
def _create_material_plate(self, resource_id: str, config: Dict[str, Any], data: Dict[str, Any], location: PLRCoordinate):
|
|
"""创建料板"""
|
|
from unilabos.device_comms.button_battery_station import MaterialPlate, ElectrodeSheet
|
|
|
|
plate = MaterialPlate(
|
|
name=resource_id,
|
|
size_x=config.get("size_x", 80.0),
|
|
size_y=config.get("size_y", 80.0),
|
|
size_z=config.get("size_z", 10.0),
|
|
hole_diameter=config.get("hole_diameter", 15.0),
|
|
hole_depth=config.get("hole_depth", 8.0),
|
|
hole_spacing_x=config.get("hole_spacing_x", 20.0),
|
|
hole_spacing_y=config.get("hole_spacing_y", 20.0),
|
|
number=data.get("number", "")
|
|
)
|
|
plate.location = location
|
|
|
|
# 如果有预填充的极片数据，创建极片
|
|
electrode_sheets = data.get("electrode_sheets", [])
|
|
for i, sheet_data in enumerate(electrode_sheets):
|
|
if i < len(plate.children): # 确保不超过洞位数量
|
|
hole = plate.children[i]
|
|
sheet = ElectrodeSheet(
|
|
name=f"{resource_id}_sheet_{i}",
|
|
diameter=sheet_data.get("diameter", 14.0),
|
|
thickness=sheet_data.get("thickness", 0.1),
|
|
mass=sheet_data.get("mass", 0.01),
|
|
material_type=sheet_data.get("material_type", "cathode"),
|
|
info=sheet_data.get("info", "")
|
|
)
|
|
hole.place_electrode_sheet(sheet)
|
|
|
|
return plate
|
|
|
|
def _create_plate_slot(self, resource_id: str, config: Dict[str, Any], data: Dict[str, Any], location: PLRCoordinate):
|
|
"""创建板槽位"""
|
|
from unilabos.device_comms.button_battery_station import PlateSlot
|
|
|
|
slot = PlateSlot(
|
|
name=resource_id,
|
|
max_plates=config.get("max_plates", 8)
|
|
)
|
|
slot.location = location
|
|
return slot
|
|
|
|
def _create_clip_magazine(self, resource_id: str, config: Dict[str, Any], data: Dict[str, Any], location: PLRCoordinate):
|
|
"""创建子弹夹"""
|
|
from unilabos.device_comms.button_battery_station import ClipMagazine
|
|
|
|
magazine = ClipMagazine(
|
|
name=resource_id,
|
|
size_x=config.get("size_x", 150.0),
|
|
size_y=config.get("size_y", 100.0),
|
|
size_z=config.get("size_z", 50.0),
|
|
hole_diameter=config.get("hole_diameter", 15.0),
|
|
hole_depth=config.get("hole_depth", 40.0),
|
|
hole_spacing=config.get("hole_spacing", 25.0),
|
|
max_sheets_per_hole=config.get("max_sheets_per_hole", 100)
|
|
)
|
|
magazine.location = location
|
|
return magazine
|
|
|
|
def _create_battery_press_slot(self, resource_id: str, config: Dict[str, Any], data: Dict[str, Any], location: PLRCoordinate):
|
|
"""创建电池压制槽"""
|
|
from unilabos.device_comms.button_battery_station import BatteryPressSlot
|
|
|
|
slot = BatteryPressSlot(
|
|
name=resource_id,
|
|
diameter=config.get("diameter", 20.0),
|
|
depth=config.get("depth", 15.0)
|
|
)
|
|
slot.location = location
|
|
return slot
|
|
|
|
def _create_tip_box(self, resource_id: str, config: Dict[str, Any], data: Dict[str, Any], location: PLRCoordinate):
|
|
"""创建枪头盒"""
|
|
from unilabos.device_comms.button_battery_station import TipBox64
|
|
|
|
tip_box = TipBox64(
|
|
name=resource_id,
|
|
size_x=config.get("size_x", 127.8),
|
|
size_y=config.get("size_y", 85.5),
|
|
size_z=config.get("size_z", 60.0),
|
|
with_tips=data.get("with_tips", True)
|
|
)
|
|
tip_box.location = location
|
|
return tip_box
|
|
|
|
def _create_waste_tip_box(self, resource_id: str, config: Dict[str, Any], data: Dict[str, Any], location: PLRCoordinate):
|
|
"""创建废枪头盒"""
|
|
from unilabos.device_comms.button_battery_station import WasteTipBox
|
|
|
|
waste_box = WasteTipBox(
|
|
name=resource_id,
|
|
size_x=config.get("size_x", 127.8),
|
|
size_y=config.get("size_y", 85.5),
|
|
size_z=config.get("size_z", 60.0),
|
|
max_tips=config.get("max_tips", 100)
|
|
)
|
|
waste_box.location = location
|
|
return waste_box
|
|
|
|
def _create_bottle_rack(self, resource_id: str, config: Dict[str, Any], data: Dict[str, Any], location: PLRCoordinate):
|
|
"""创建瓶架"""
|
|
from unilabos.device_comms.button_battery_station import BottleRack
|
|
|
|
rack = BottleRack(
|
|
name=resource_id,
|
|
size_x=config.get("size_x", 210.0),
|
|
size_y=config.get("size_y", 140.0),
|
|
size_z=config.get("size_z", 100.0),
|
|
bottle_diameter=config.get("bottle_diameter", 30.0),
|
|
bottle_height=config.get("bottle_height", 100.0),
|
|
position_spacing=config.get("position_spacing", 35.0)
|
|
)
|
|
rack.location = location
|
|
return rack
|
|
|
|
def _create_battery(self, resource_id: str, config: Dict[str, Any], data: Dict[str, Any], location: PLRCoordinate):
|
|
"""创建电池"""
|
|
from unilabos.device_comms.button_battery_station import Battery
|
|
|
|
battery = Battery(
|
|
name=resource_id,
|
|
diameter=config.get("diameter", 20.0),
|
|
height=config.get("height", 3.2),
|
|
max_volume=config.get("max_volume", 100.0),
|
|
barcode=data.get("barcode", "")
|
|
)
|
|
battery.location = location
|
|
return battery
|
|
|
|
# ============ 纽扣电池特定查找方法 ============
|
|
|
|
def find_material_plates(self):
|
|
"""查找所有料板"""
|
|
from unilabos.device_comms.button_battery_station import MaterialPlate
|
|
return self.resource_tracker.find_by_type(MaterialPlate)
|
|
|
|
def find_batteries(self):
|
|
"""查找所有电池"""
|
|
from unilabos.device_comms.button_battery_station import Battery
|
|
return self.resource_tracker.find_by_type(Battery)
|
|
|
|
def find_electrode_sheets(self):
|
|
"""查找所有极片"""
|
|
found = []
|
|
plates = self.find_material_plates()
|
|
for plate in plates:
|
|
for hole in plate.children:
|
|
if hasattr(hole, 'has_electrode_sheet') and hole.has_electrode_sheet():
|
|
found.append(hole._electrode_sheet)
|
|
return found
|
|
|
|
def find_plate_slots(self):
|
|
"""查找所有板槽位"""
|
|
from unilabos.device_comms.button_battery_station import PlateSlot
|
|
return self.resource_tracker.find_by_type(PlateSlot)
|
|
|
|
def find_clip_magazines(self):
|
|
"""查找所有子弹夹"""
|
|
from unilabos.device_comms.button_battery_station import ClipMagazine
|
|
return self.resource_tracker.find_by_type(ClipMagazine)
|
|
|
|
def find_press_slots(self):
|
|
"""查找所有压制槽"""
|
|
from unilabos.device_comms.button_battery_station import BatteryPressSlot
|
|
return self.resource_tracker.find_by_type(BatteryPressSlot)
|