479 lines
16 KiB
Python
479 lines
16 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
OPC UA 客户端测试脚本
|
||
用于连接和测试 OPC UA 服务器
|
||
"""
|
||
|
||
from opcua import Client
|
||
import time
|
||
import sys
|
||
|
||
class OPCUAClientTest:
|
||
def __init__(self, server_url="opc.tcp://localhost:4840/zjsh_feed/server/"):
|
||
"""
|
||
初始化 OPC UA 客户端
|
||
|
||
Args:
|
||
server_url: 服务器URL地址
|
||
"""
|
||
self.client = Client(server_url)
|
||
self.connected = False
|
||
|
||
def connect(self):
|
||
"""连接到服务器"""
|
||
try:
|
||
self.client.connect()
|
||
self.connected = True
|
||
print(f"成功连接到 OPC UA 服务器: {self.client.server_url}")
|
||
return True
|
||
except Exception as e:
|
||
print(f"连接服务器失败: {e}")
|
||
return False
|
||
|
||
def disconnect(self):
|
||
"""断开连接"""
|
||
if self.connected:
|
||
self.client.disconnect()
|
||
self.connected = False
|
||
print("已断开与 OPC UA 服务器的连接")
|
||
|
||
def browse_nodes(self):
|
||
"""浏览服务器节点结构"""
|
||
if not self.connected:
|
||
print("请先连接到服务器")
|
||
return
|
||
|
||
try:
|
||
# 获取根节点
|
||
root = self.client.get_root_node()
|
||
print(f"根节点: {root}")
|
||
|
||
# 获取对象节点
|
||
objects = self.client.get_objects_node()
|
||
print(f"对象节点: {objects}")
|
||
|
||
# 浏览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()
|
||
lower_weight = lower_device.get_child("2:lower_weight").get_value()
|
||
print(f"上料斗重量: {upper_weight}")
|
||
print(f"下料斗重量: {lower_weight}")
|
||
|
||
|
||
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:
|
||
print("请先连接到服务器")
|
||
return
|
||
|
||
print(f"\n开始监控数据变化，持续 {duration} 秒...")
|
||
|
||
try:
|
||
# 获取传感器节点
|
||
objects = self.client.get_objects_node()
|
||
upper_device = objects.get_child("2:upper")
|
||
lower_device = objects.get_child("2:lower")
|
||
|
||
|
||
|
||
|
||
start_time = time.time()
|
||
while time.time() - start_time < duration:
|
||
print(f"\n--- {time.strftime('%H:%M:%S')} ---")
|
||
self.read_object_properties(upper_device, lower_device)
|
||
time.sleep(5) # 每5秒读取一次
|
||
|
||
except KeyboardInterrupt:
|
||
print("\n监控被用户中断")
|
||
except Exception as e:
|
||
print(f"监控数据时出错: {e}")
|
||
|
||
|
||
|
||
def main():
|
||
"""主函数"""
|
||
# 创建客户端
|
||
client = OPCUAClientTest("opc.tcp://localhost:4840/zjsh_feed/server/")
|
||
|
||
try:
|
||
# 连接到服务器
|
||
if not client.connect():
|
||
return
|
||
|
||
# 浏览节点结构（首先发现实际节点结构）
|
||
print("\n" + "="*60)
|
||
print("步骤1: 浏览服务器节点结构")
|
||
print("="*60)
|
||
client.browse_nodes()
|
||
|
||
# 尝试使用新方法写入数据
|
||
print("\n" + "="*60)
|
||
print("步骤2: 使用动态节点查找方法写入数据")
|
||
print("="*60)
|
||
|
||
# 方法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)
|
||
|
||
# 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()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
if len(sys.argv) > 1:
|
||
# 支持自定义服务器地址
|
||
server_url = sys.argv[1]
|
||
client = OPCUAClientTest(server_url)
|
||
else:
|
||
client = OPCUAClientTest()
|
||
|
||
try:
|
||
main()
|
||
except Exception as e:
|
||
print(f"客户端运行错误: {e}")
|
||
sys.exit(1) |