97 lines
2.7 KiB
Python
97 lines
2.7 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""
|
||
|
|
Test script to debug OPC UA client connection issues
|
||
|
|
"""
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
import structlog
|
||
|
|
|
||
|
|
logger = structlog.get_logger()
|
||
|
|
|
||
|
|
|
||
|
|
async def test_opcua_connection():
|
||
|
|
"""Test OPC UA client connection with detailed debugging."""
|
||
|
|
try:
|
||
|
|
from asyncua import Client
|
||
|
|
|
||
|
|
endpoint = "opc.tcp://localhost:4840"
|
||
|
|
print(f"Testing OPC UA connection to: {endpoint}")
|
||
|
|
|
||
|
|
# Create client
|
||
|
|
client = Client(url=endpoint)
|
||
|
|
|
||
|
|
# Set timeout
|
||
|
|
client.secure_channel_timeout = 10000
|
||
|
|
client.session_timeout = 60000
|
||
|
|
|
||
|
|
print("Attempting to connect...")
|
||
|
|
|
||
|
|
# Try to connect
|
||
|
|
await client.connect()
|
||
|
|
|
||
|
|
print("✓ Connection successful!")
|
||
|
|
|
||
|
|
# Get server info
|
||
|
|
server_info = await client.get_server_node().get_child(["Server", "ServerStatus"])
|
||
|
|
print(f"Server status: {server_info}")
|
||
|
|
|
||
|
|
# Try to read a node
|
||
|
|
try:
|
||
|
|
# Try to read a known node
|
||
|
|
node = client.get_node("ns=2;s=Station_STATION_001.Pump_PUMP_001.Setpoint_Hz")
|
||
|
|
value = await node.read_value()
|
||
|
|
print(f"✓ Successfully read node value: {value}")
|
||
|
|
except Exception as e:
|
||
|
|
print(f"✗ Failed to read node: {e}")
|
||
|
|
|
||
|
|
# Disconnect
|
||
|
|
await client.disconnect()
|
||
|
|
print("✓ Disconnected successfully")
|
||
|
|
|
||
|
|
except asyncio.TimeoutError:
|
||
|
|
print("✗ Connection timeout")
|
||
|
|
except Exception as e:
|
||
|
|
print(f"✗ Connection failed: {e}")
|
||
|
|
import traceback
|
||
|
|
traceback.print_exc()
|
||
|
|
|
||
|
|
|
||
|
|
async def test_opcua_with_security_policy():
|
||
|
|
"""Test OPC UA connection with explicit security policy."""
|
||
|
|
try:
|
||
|
|
from asyncua import Client
|
||
|
|
from asyncua.crypto.security_policies import SecurityPolicyBasic256Sha256
|
||
|
|
|
||
|
|
endpoint = "opc.tcp://localhost:4840"
|
||
|
|
print(f"\nTesting OPC UA connection with explicit security policy: {endpoint}")
|
||
|
|
|
||
|
|
# Create client
|
||
|
|
client = Client(url=endpoint)
|
||
|
|
|
||
|
|
# Try with None security policy explicitly
|
||
|
|
print("Attempting to connect with None security policy...")
|
||
|
|
await client.set_security_string("None")
|
||
|
|
|
||
|
|
await client.connect()
|
||
|
|
print("✓ Connection successful with None security policy!")
|
||
|
|
|
||
|
|
await client.disconnect()
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print(f"✗ Connection failed: {e}")
|
||
|
|
import traceback
|
||
|
|
traceback.print_exc()
|
||
|
|
|
||
|
|
|
||
|
|
async def main():
|
||
|
|
"""Run all tests."""
|
||
|
|
print("=" * 60)
|
||
|
|
print("OPC UA Client Connection Debug")
|
||
|
|
print("=" * 60)
|
||
|
|
|
||
|
|
await test_opcua_connection()
|
||
|
|
await test_opcua_with_security_policy()
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
asyncio.run(main())
|