You need to enable JavaScript to run this app.
最新活动
大模型
产品
解决方案
定价
生态与合作
支持与服务
开发者
了解我们

如何使用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.

1. First, nail the core concepts (no jargon overload)

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 (like urn: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.
2. Pick your Python library (no overthinking needed)

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
3. Minimal working examples (get something running first)

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.")
4. Debugging tools to make your life easier

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.
5. Common pitfalls to avoid
  • 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_idx returned 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

火山引擎 最新活动