变频器集成以及增加点动控制(0209)

opc/opcua_client_test.py

@@ -52,20 +52,235 @@ class OPCUAClientTest:
             objects = self.client.get_objects_node()
             print(f"对象节点: {objects}")
             
-            upper_device = objects.get_child("2:upper")
-            print(f"\n上料斗对象: {upper_device}")
-            
-            lower_device = objects.get_child("2:lower")
-            print(f"下料斗对象: {lower_device}")
-            
-            print("\n=== 当前对象属性===")
-            self.read_object_properties(upper_device, lower_device)
+            # 浏览Objects下的所有子节点
+            print("\n=== Objects 节点下的子节点 ===")
+            for child in objects.get_children():
+                browse_name = child.get_browse_name()
+                print(f"  节点: {browse_name} (nodeId: {child.nodeid})")
+                
+                # 如果是上料斗或下料斗对象，继续浏览它们的子节点
+                if "upper" in str(browse_name).lower() or "lower" in str(browse_name).lower():
+                    try:
+                        for sub_child in child.get_children():
+                            sub_browse_name = sub_child.get_browse_name()
+                            print(f"    └─ {sub_browse_name} (nodeId: {sub_child.nodeid})")
+                    except:
+                        pass
             
+            # 尝试获取设备对象
+            print("\n=== 尝试获取设备对象 ===")
+            try:
+                upper_device = objects.get_child("2:upper")
+                print(f"上料斗对象: {upper_device}")
+            except Exception as e:
+                print(f"获取上料斗对象失败: {e}")
+                # 尝试其他可能的路径
+                try:
+                    upper_device = objects.get_child(["2:upper"])
+                    print(f"上料斗对象(列表方式): {upper_device}")
+                except Exception as e2:
+                    print(f"  也无法通过列表方式获取: {e2}")
+                    
+            try:
+                lower_device = objects.get_child("2:lower")
+                print(f"下料斗对象: {lower_device}")
+            except Exception as e:
+                print(f"获取下料斗对象失败: {e}")
+                
         except Exception as e:
             print(f"浏览节点时出错: {e}")
             
+    def get_node_path(self, obj_path: str) -> str:
+        """
+        获取节点路径 - 尝试多种格式
+        
+        Args:
+            obj_path: 对象名称（如 "upper", "upper_weight"）
+            
+        Returns:
+            str: 节点路径，如果找不到返回 None
+        """
+        if not self.connected:
+            return None
+            
+        try:
+            objects = self.client.get_objects_node()
+            
+            # 尝试多种节点路径格式
+            path_formats = [
+                f"2:{obj_path}",
+                f"2:upper/2:{obj_path}",
+                f"2:lower/2:{obj_path}",
+                f"ns=2;{obj_path}",
+                obj_path
+            ]
+            
+            for path in path_formats:
+                try:
+                    node = objects.get_child(path)
+                    print(f"  找到节点: {path} -> {node}")
+                    return path
+                except:
+                    continue
+                    
+            return None
+            
+        except Exception as e:
+            print(f"查找节点路径时出错: {e}")
+            return None
+            
+    def write_data(self, node_path: str, value, data_type: str = "auto") -> bool:
+        """
+        向OPC UA节点写入数据
+        
+        Args:
+            node_path: 节点路径（如 "2:upper/2:upper_weight"）
+            value: 要写入的值
+            data_type: 数据类型（"int", "float", "bool", "string", "auto"）
+            
+        Returns:
+            bool: 写入成功返回True，失败返回False
+        """
+        if not self.connected:
+            print("请先连接到服务器")
+            return False
+            
+        try:
+            # 获取对象节点
+            objects = self.client.get_objects_node()
+            
+            # 尝试多种方式获取节点
+            node = None
+            error_msg = None
+            
+            # 方式1: 直接使用路径
+            try:
+                node = objects.get_child(node_path)
+            except Exception as e:
+                error_msg = e
+                
+            # 方式2: 分解路径
+            if node is None and "/" in node_path:
+                try:
+                    parts = node_path.split("/")
+                    node = objects
+                    for part in parts:
+                        node = node.get_child(part)
+                except:
+                    pass
+                    
+            # 方式3: 尝试用数字索引
+            if node is None:
+                try:
+                    node = objects.get_child([node_path])
+                except:
+                    pass
+            
+            if node is None:
+                print(f"写入数据失败 {node_path}: 找不到节点")
+                print(f"  详细错误: {error_msg}")
+                print(f"  提示: 请先运行 browse_nodes() 方法查看可用的节点路径")
+                return False
+            
+            # 根据数据类型转换值
+            if data_type == "int":
+                value = int(value)
+            elif data_type == "float":
+                value = float(value)
+            elif data_type == "bool":
+                value = bool(value)
+            elif data_type == "string":
+                value = str(value)
+            # "auto" 模式下自动推断类型
+            
+            # 写入数据
+            node.set_value(value)
+            
+            # 获取节点名称用于显示
+            try:
+                node_name = node.get_browse_name()
+            except:
+                node_name = node_path
+                
+            print(f"✓ 成功写入 {node_name}: {value}")
+            return True
+            
+        except Exception as e:
+            print(f"✗ 写入数据失败 {node_path}: {e}")
+            return False
+    
+    def write_weights_directly(self, upper_value, lower_value) -> bool:
+        """
+        直接写入上下料斗重量（自动检测节点路径）
+        
+        Args:
+            upper_value: 上料斗重量值
+            lower_value: 下料斗重量值
+            
+        Returns:
+            bool: 写入成功返回True，失败返回False
+        """
+        if not self.connected:
+            print("请先连接到服务器")
+            return False
+            
+        success = True
+        
+        try:
+            objects = self.client.get_objects_node()
+            
+            # 查找上料斗重量节点
+            upper_weight_node = None
+            lower_weight_node = None
+            
+            # 遍历Objects下的所有节点
+            for child in objects.get_children():
+                browse_name = str(child.get_browse_name())
+                print(browse_name)
+                if "upper" in browse_name.lower():
+                    upper_weight_node = child
+                    print(f"找到上料斗重量节点: {browse_name}")
+                    break
+                    
+            for child in objects.get_children():
+                browse_name = str(child.get_browse_name())
+                if "lower" in browse_name.lower():
+                    lower_weight_node = child
+                    print(f"找到下料斗重量节点: {browse_name}")
+                    break
+            
+            # 写入上料斗重量
+            if upper_weight_node:
+                try:
+                    upper_weight_node.set_value(upper_value)
+                    print(f"✓ 成功写入上料斗重量: {upper_value}")
+                except Exception as e:
+                    print(f"✗ 写入上料斗重量失败: {e}")
+                    success = False
+            else:
+                print("✗ 未找到上料斗重量节点")
+                success = False
+                
+            # 写入下料斗重量
+            if lower_weight_node:
+                try:
+                    lower_weight_node.set_value(lower_value)
+                    print(f"✓ 成功写入下料斗重量: {lower_value}")
+                except Exception as e:
+                    print(f"✗ 写入下料斗重量失败: {e}")
+                    success = False
+            else:
+                print("✗ 未找到下料斗重量节点")
+                success = False
+                
+            return success
+            
+        except Exception as e:
+            print(f"写入重量数据时出错: {e}")
+            return False
+    
     def read_object_properties(self, upper_device, lower_device):
-        """读取重量数值"""
+        """读取重量数值（需要外部传入device对象）"""
         try:
             # 读取重量
             upper_weight = upper_device.get_child("2:upper_weight").get_value()
@@ -76,7 +291,79 @@ class OPCUAClientTest:
             
         except Exception as e:
             print(f"读取数据时出错: {e}")
+    
+    def read_weights(self) -> tuple:
+        """
+        直接读取上料斗和下料斗重量（无需先获取device对象）
+        
+        Returns:
+            tuple: (上料斗重量, 下料斗重量)，读取失败返回 (None, None)
+        """
+        if not self.connected:
+            print("请先连接到服务器")
+            return None, None
             
+        try:
+            # 直接获取节点并读取数据
+            objects = self.client.get_objects_node()
+            
+            # 使用列表格式访问节点（freeopcua推荐的方式）
+            upper_weight = objects.get_child(["2:upper", "2:upper_weight"]).get_value()
+            lower_weight = objects.get_child(["2:lower", "2:lower_weight"]).get_value()
+            
+            print(f"上料斗重量: {upper_weight}")
+            print(f"下料斗重量: {lower_weight}")
+            
+            return upper_weight, lower_weight
+            
+        except Exception as e:
+            print(f"读取重量数据时出错: {e}")
+            return None, None
+    
+    def write_multiple_values(self, values_dict: dict) -> dict:
+        """
+        批量写入多个节点
+        
+        Args:
+            values_dict: 字典，key为节点路径，value为要写入的值
+            
+        Returns:
+            dict: 写入结果，key为节点路径，value为成功/失败状态
+        """
+        results = {}
+        for node_path, value in values_dict.items():
+            results[node_path] = self.write_data(node_path, value)
+        return results
+    
+    def write_test_data(self):
+        """测试写入各种类型的数据"""
+        if not self.connected:
+            print("请先连接到服务器")
+            return
+            
+        print("\n=== 测试写入数据 ===")
+        
+        # 测试写入目标重量
+        self.write_data("2:upper/2:target_weight", 150.5, "float")
+        
+        # 测试写入开关量
+        self.write_data("2:upper/2:valve_on", True, "bool")
+        
+        # 测试写入整数
+        self.write_data("2:upper/2:cycle_count", 10, "int")
+        
+        # 测试批量写入
+        values = {
+            "2:upper/2:target_weight": 200.0,
+            "2:lower/2:target_weight": 100.0,
+        }
+        results = self.write_multiple_values(values)
+        
+        print("\n批量写入结果:")
+        for path, success in results.items():
+            status = "✓ 成功" if success else "✗ 失败"
+            print(f"  {path}: {status}")
+    
     def monitor_data(self, duration=30):
         """监控数据变化"""
         if not self.connected:
@@ -97,7 +384,7 @@ class OPCUAClientTest:
             start_time = time.time()
             while time.time() - start_time < duration:
                 print(f"\n--- {time.strftime('%H:%M:%S')} ---")
-                self.read_sensor_values(upper_device, lower_device)
+                self.read_object_properties(upper_device, lower_device)
                 time.sleep(5)  # 每5秒读取一次
                 
         except KeyboardInterrupt:
@@ -117,21 +404,61 @@ def main():
         if not client.connect():
             return
             
-        # 浏览节点结构
+        # 浏览节点结构（首先发现实际节点结构）
+        print("\n" + "="*60)
+        print("步骤1: 浏览服务器节点结构")
+        print("="*60)
         client.browse_nodes()
         
-        # 监控数据变化
-        client.monitor_data(duration=30)
+        # 尝试使用新方法写入数据
+        print("\n" + "="*60)
+        print("步骤2: 使用动态节点查找方法写入数据")
+        print("="*60)
         
-        # 测试写入数据
-        # client.write_test_data()
+        # 方法1: 使用write_weights_directly自动查找节点
+        print("\n尝试方法1: write_weights_directly (自动查找节点)")
+        # client.write_data("2:upper/2:upper_weight", 180, "int")
+        # client.write_data("2:lower/2:lower_weight", 120, "int")
+
+        values = {
+            "2:upper/2:upper_weight": 200,
+            "2:lower/2:lower_weight": 100,
+        }
+        client.write_multiple_values(values)
+        # success1 = client.write_weights_directly(150, 120)
+        time.sleep(2)
         
-        # 继续监控
-        print("\n继续监控数据...")
-        client.monitor_data(duration=15)
+        # if not success1:
+        #     # 方法2: 尝试可能的替代路径
+        #     print("\n尝试方法2: 尝试其他节点路径格式")
+            
+        #     # 列出可能的节点路径格式
+        #     possible_paths = [
+        #         "2:upper_weight",
+        #         "2:lower_weight", 
+        #         "2:upper/upper_weight",
+        #         "2:lower/lower_weight",
+        #         "ns=2;upper_weight",
+        #         "ns=2;lower_weight"
+        #     ]
+            
+        #     for path in possible_paths:
+        #         print(f"  尝试写入: {path}")
+        #         client.write_data(path, 150.5, "float")
+        #         time.sleep(0.5)
+        
+        print("\n" + "="*60)
+        print("步骤3: 读取验证数据")
+        print("="*60)
+        upper, lower = client.read_weights()
+        print(f"读取结果 - 上料斗: {upper}, 下料斗: {lower}")
         
     except KeyboardInterrupt:
         print("\n客户端被用户中断")
+    except Exception as e:
+        print(f"客户端运行错误: {e}")
+        import traceback
+        traceback.print_exc()
     finally:
         # 断开连接
         client.disconnect()