bioyond-HR

unilabos/device_comms/opcua_client/node/uniopcua.py

@@ -43,7 +43,7 @@ def __init__(self, client: Client, name: str, node_id: str, typ: NodeType, data_
         self._type = typ
         self._data_type = data_type
         self._node: Optional[Node] = None
-        
+
     def _get_node(self) -> Node:
         if self._node is None:
             try:
@@ -66,7 +66,7 @@ def _get_node(self) -> Node:
                 # 直接以字符串形式处理
                 if isinstance(nid, str):
                     nid = nid.strip()
-                    
+
                     # 处理包含类名的格式，如 'StringNodeId(ns=4;s=...)' 或 'NumericNodeId(ns=2;i=...)'
                     # 提取括号内的内容
                     match_wrapped = re.match(r'(String|Numeric|Byte|Guid|TwoByteNode|FourByteNode)NodeId\((.*)\)', nid)
@@ -116,16 +116,16 @@ def _get_node(self) -> Node:
     def read(self) -> Tuple[Any, bool]:
         """读取节点值，返回(值, 是否出错)"""
         pass
-    
+
     @abstractmethod
     def write(self, value: Any) -> bool:
         """写入节点值，返回是否出错"""
         pass
-    
+
     @property
     def type(self) -> NodeType:
         return self._type
-    
+
     @property
     def node_id(self) -> str:
         return self._node_id
@@ -210,15 +210,15 @@ def __init__(self, client: Client, name: str, node_id: str, parent_node_id: str,
         super().__init__(client, name, node_id, NodeType.METHOD, data_type)
         self._parent_node_id = parent_node_id
         self._parent_node = None
-        
+
     def _get_parent_node(self) -> Node:
         if self._parent_node is None:
             try:
                 # 处理父节点ID，使用与_get_node相同的解析逻辑
                 import re
-                
+
                 nid = self._parent_node_id
-                
+
                 # 如果已经是 NodeId 对象，直接使用
                 try:
                     from opcua.ua import NodeId as UaNodeId
@@ -227,16 +227,16 @@ def _get_parent_node(self) -> Node:
                         return self._parent_node
                 except Exception:
                     pass
-                
+
                 # 字符串处理
                 if isinstance(nid, str):
                     nid = nid.strip()
-                    
+
                     # 处理包含类名的格式
                     match_wrapped = re.match(r'(String|Numeric|Byte|Guid|TwoByteNode|FourByteNode)NodeId\((.*)\)', nid)
                     if match_wrapped:
                         nid = match_wrapped.group(2).strip()
-                    
+
                     # 常见短格式
                     if re.match(r'^ns=\d+;[is]=', nid):
                         self._parent_node = self._client.get_node(nid)
@@ -271,7 +271,7 @@ def read(self) -> Tuple[Any, bool]:
     def write(self, value: Any) -> bool:
         """方法节点不支持写入操作"""
         return True
-        
+
     def call(self, *args) -> Tuple[Any, bool]:
         """调用方法，返回(返回值, 是否出错)"""
         try:
@@ -285,20 +285,20 @@ def call(self, *args) -> Tuple[Any, bool]:
 class Object(Base):
     def __init__(self, client: Client, name: str, node_id: str):
         super().__init__(client, name, node_id, NodeType.OBJECT, None)
-        
+
     def read(self) -> Tuple[Any, bool]:
         """对象节点不支持直接读取操作"""
         return None, True
 
     def write(self, value: Any) -> bool:
         """对象节点不支持直接写入操作"""
         return True
-    
+
     def get_children(self) -> Tuple[List[Node], bool]:
         """获取子节点列表，返回(子节点列表, 是否出错)"""
         try:
             children = self._get_node().get_children()
             return children, False
         except Exception as e:
             print(f"获取对象 {self._name} 的子节点失败: {e}")
-            return [], True 
+            return [], True

unilabos/devices/workstation/bioyond_studio/bioyond_rpc.py

@@ -176,7 +176,24 @@ def add_material(self, params: dict) -> dict:
             return {}
 
         print(f"add material data: {response['data']}")
-        return response.get("data", {})
+
+        # 自动更新缓存
+        data = response.get("data", {})
+        if data:
+            if isinstance(data, str):
+                # 如果返回的是字符串，通常是ID
+                mat_id = data
+                name = params.get("name")
+            else:
+                # 如果返回的是字典，尝试获取name和id
+                name = data.get("name") or params.get("name")
+                mat_id = data.get("id")
+
+            if name and mat_id:
+                self.material_cache[name] = mat_id
+                print(f"已自动更新缓存: {name} -> {mat_id}")
+
+        return data
 
     def query_matial_type_id(self, data) -> list:
         """查找物料typeid"""
@@ -203,7 +220,7 @@ def material_type_list(self) -> list:
             params={
                 "apiKey": self.api_key,
                 "requestTime": self.get_current_time_iso8601(),
-                "data": {},
+                "data": 0,
             })
         if not response or response['code'] != 1:
             return []
@@ -273,6 +290,14 @@ def delete_material(self, material_id: str) -> dict:
 
         if not response or response['code'] != 1:
             return {}
+
+        # 自动更新缓存 - 移除被删除的物料
+        for name, mid in list(self.material_cache.items()):
+            if mid == material_id:
+                del self.material_cache[name]
+                print(f"已从缓存移除物料: {name}")
+                break
+
         return response.get("data", {})
 
     def material_outbound(self, material_id: str, location_name: str, quantity: int) -> dict: