如何使用Python基于OPC UA协议实现服务器与客户端通信?
Hey there! As someone who’s stumbled through my first OPC UA + Python project too, I totally get how fuzzy this requirement might feel right now. Let’s break this down into bite-sized pieces so you can wrap your head around it and start building.
Before writing code, let’s make sure you understand the basics of what you’re building:
- OPC UA Server: This is the "data provider"—it exposes variables, objects, or methods that clients can interact with (think: industrial sensor readings, system status values).
- OPC UA Client: This connects to the server to read data, write new values, or subscribe to real-time updates.
- Namespaces: OPC UA uses these to avoid naming conflicts. The default namespace is
0(for standard OPC UA objects), but you’ll want to create your own (likeurn:my-team:my-project) to keep your custom data organized. - Nodes: Everything in an OPC UA server is a node—variable nodes store data, object nodes act as containers, and method nodes are callable functions.
The most mature, beginner-friendly OPC UA library for Python is python-opcua (synchronous version). If you need async support later, you can switch to asyncua (its async sibling), but start with the sync one—it’s way easier to debug when you’re learning.
Install it with:
pip install python-opcua
Let’s write a super simple server and client to see how they talk. This will help you visualize the flow immediately.
Server code (server.py)
This creates a server that exposes a temperature variable and updates it every second:
from opcua import Server import time # Initialize the server server = Server() # Set the endpoint (use any free port; 4840 is the OPC UA default) server.set_endpoint("opc.tcp://0.0.0.0:4840/my-python-opcua-server/") # Register your custom namespace uri = "urn:my-team:temperature-monitor" namespace_idx = server.register_namespace(uri) # Get core nodes (OPC UA uses a tree structure) root_node = server.get_root_node() objects_node = server.get_objects_node() # Create a custom object to hold our variable temp_object = objects_node.add_object(namespace_idx, "TemperatureSensor") # Add a writable temperature variable (initial value: 0.0) temp_variable = temp_object.add_variable(namespace_idx, "CurrentTemp", 0.0) temp_variable.set_writable() # Allow clients to write to this # Start the server server.start() print(f"Server running at: opc.tcp://0.0.0.0:4840/my-python-opcua-server/") # Simulate temperature updates try: while True: new_temp = temp_variable.get_value() + 0.5 temp_variable.set_value(new_temp) print(f"Updated temp: {new_temp:.1f}°C") time.sleep(1) finally: # Clean up when done server.stop()
Client code (client.py)
This connects to the server, reads the temperature, writes a new value, and subscribes to real-time updates:
from opcua import Client # Connect to the server's endpoint client = Client("opc.tcp://localhost:4840/my-python-opcua-server/") try: client.connect() print("Connected to OPC UA Server!") # Get the temperature variable (two ways to find it) # Way 1: Traverse the node tree temp_var = client.get_root_node().get_child( ["0:Objects", "2:TemperatureSensor", "2:CurrentTemp"] ) # Way 2: Use the node ID (more efficient; find this via server logs or UA tools) # temp_var = client.get_node("ns=2;i=2") # Read the current value current_temp = temp_var.get_value() print(f"Current temperature: {current_temp:.1f}°C") # Write a new value (if the server allows it) temp_var.set_value(22.5) print(f"Set temperature to 22.5°C. New value: {temp_var.get_value():.1f}°C") # Subscribe to real-time updates def on_temp_change(node, value, data): print(f"🔔 Temperature updated: {value:.1f}°C") # Create a subscription (500ms refresh rate) subscription = client.create_subscription(500, on_temp_change) subscription.subscribe_data_change(temp_var) print("Subscribed to temperature updates. Press Enter to exit...") input() finally: client.disconnect() print("Disconnected from server.")
When you’re building, it helps to visualize the server’s node structure without writing code. Use these free tools:
- UA Expert: Siemens’ industry-standard tool—connect to your server, browse all nodes, and test read/write operations.
- UaViewer: A lightweight alternative for quick debugging.
- Port conflicts: If your server won’t start, check if port 4840 is in use—swap it for another free port (like 4841).
- Namespace index mix-ups: The
namespace_idxreturned when registering your namespace is critical—use it consistently when creating nodes, or you’ll get "node not found" errors. - Data type mismatches: OPC UA enforces strict data types. Don’t try to write a string to a float variable—double-check types on both server and client sides.
- Security (for production): The examples use anonymous access, but for real-world use, add user authentication or certificate-based security to your server.
内容的提问来源于stack exchange,提问作者Prakash Ganesan




