OPC UA with Python: asyncua Quick Start for BAS

What Is OPC UA in Building Automation

OPC UA (Open Platform Communications Unified Architecture) is a platform-independent, service-oriented communication protocol standardized under IEC 62541. In building automation, it serves as a vendor-neutral bridge between supervisory platforms, controllers, and analytics software. While BACnet and Modbus remain dominant at the field-device level, OPC UA is increasingly used at the supervisory and integration tiers for several reasons:

• Unified namespace — OPC UA organizes all data points into a hierarchical node tree with standardized data types, making it straightforward to map HVAC objects (zone temperatures, damper commands, schedule overrides) into a single browsable structure
• Built-in security — OPC UA supports encrypted transport (TLS-based), certificate authentication, and user/password login, addressing a major gap in legacy BAS protocols that transmit data in cleartext
• Cross-platform interoperability — platforms such as Tridium Niagara, Siemens Desigo CC, Johnson Controls Metasys, Honeywell EBI, and Schneider EcoStruxure all expose OPC UA server endpoints, allowing a single Python client to pull data from multiple vendor systems
• Publish/subscribe model — instead of polling every point on a timer, OPC UA lets clients subscribe to nodes and receive push notifications only when values change, reducing network traffic on large campus networks

Python scripting with asyncua fills the gap between clicking through a vendor's engineering tool and building a full middleware layer. Common BAS use cases include bulk point verification during commissioning, overnight trend captures across multiple controllers, automated setpoint adjustments, and feeding live building data into analytics or machine learning pipelines.

Installing asyncua

asyncua requires Python 3.10 or later. Install from PyPI:

pip install asyncua

For BAS scripting workflows that involve data export or analysis, install alongside common data tools:

pip install asyncua pandas openpyxl

Virtual environment recommended: asyncua depends on cryptography, aiofiles, sortedcontainers, and several other packages. Isolate your environment to avoid conflicts:

python -m venv opcua-env
source opcua-env/bin/activate   # Linux/macOS
opcua-env\Scripts\activate      # Windows
pip install asyncua

Verify the install by running python -c "import asyncua; print(asyncua.__version__)". If a version string prints with no errors, you are ready to connect.

Connecting to an OPC UA Server

Every OPC UA server exposes an endpoint URL in the format opc.tcp://host:port/path. In building automation, your Niagara station, Desigo CC server, or integration gateway will publish this URL in its OPC UA configuration page. The asyncua Client class manages the TCP connection, session creation, and security negotiation.

Basic Connection with Async Context Manager

The cleanest pattern uses async with, which automatically disconnects when the block exits:

import asyncio
from asyncua import Client

async def main():
    url = "opc.tcp://192.168.1.50:4840"
    async with Client(url=url) as client:
        print("Connected to OPC UA server")
        # Read, write, browse, or subscribe here

asyncio.run(main())

Connection with Authentication

Many BAS servers require username and password authentication. Set credentials before connecting:

async def main():
    url = "opc.tcp://192.168.1.50:4840"
    client = Client(url=url)
    client.set_user("bas_operator")
    client.set_password("securepassword")

    async with client:
        print("Authenticated and connected")
        # Your operations here

asyncio.run(main())

Connection with Security Policy

For encrypted connections (recommended in production BAS environments), configure a security policy and provide client certificates:

from asyncua.crypto.security_policies import SecurityPolicyBasic256Sha256

async def main():
    url = "opc.tcp://192.168.1.50:4840"
    client = Client(url=url)
    await client.set_security(
        SecurityPolicyBasic256Sha256,
        certificate="client_cert.pem",
        private_key="client_key.pem",
        server_certificate="server_cert.pem"
    )
    async with client:
        print("Secure connection established")

asyncio.run(main())

Browsing the Node Tree

OPC UA organizes all data into a hierarchical tree of nodes. Every node has a unique NodeId (e.g., ns=2;s=Building1/Floor3/VAV301/ZoneTemp) and a human-readable BrowseName. Browsing the tree is essential when you are commissioning a new system and need to discover what points are available.

from asyncua import Client

async def browse_tree(node, indent=0):
    """Recursively browse and print the OPC UA node tree."""
    name = await node.read_browse_name()
    print(f"{'  ' * indent}{name.Name} (NodeId: {node.nodeid})")
    children = await node.get_children()
    for child in children:
        await browse_tree(child, indent + 1)

async def main():
    url = "opc.tcp://192.168.1.50:4840"
    async with Client(url=url) as client:
        # Start from the Objects folder (standard root for user data)
        objects = client.nodes.objects
        print("--- OPC UA Node Tree ---")
        await browse_tree(objects)

asyncio.run(main())

On a building automation server, the Objects folder typically contains folders organized by building, floor, and equipment. For example, a Niagara station might expose Objects/Building1/AHU-01/SupplyAirTemp as a path to a supply air temperature sensor.

Tip: On large BAS servers with thousands of points, a full recursive browse can take a long time and generate heavy traffic. Limit your browse depth or target a specific subfolder when you already know the general path structure.

Reading Values

Once you know a node's NodeId or its browse path, reading its current value requires just a few lines. The read_value() method returns the value as a native Python type (float, int, string, boolean), while read_data_value() returns the full OPC UA DataValue wrapper including timestamps and status codes.

from asyncua import Client

async def main():
    url = "opc.tcp://192.168.1.50:4840"
    async with Client(url=url) as client:
        # Option 1: Access a node by its NodeId string
        node = client.get_node("ns=2;s=Building1/AHU-01/SupplyAirTemp")
        value = await node.read_value()
        print(f"Supply Air Temp: {value}")

        # Option 2: Navigate by browse path from Objects root
        node = await client.nodes.objects.get_child(
            ["2:Building1", "2:AHU-01", "2:ZoneTemp"]
        )
        value = await node.read_value()
        print(f"Zone Temp: {value}")

        # Option 3: Read full DataValue (includes timestamp and status)
        dv = await node.read_data_value()
        print(f"Value: {dv.Value.Value}")
        print(f"Source Timestamp: {dv.SourceTimestamp}")
        print(f"Status: {dv.StatusCode}")

asyncio.run(main())

Bulk reads: If you need to read many points at once, gather node references first and read them in a loop or use asyncio.gather() for concurrent reads. This is significantly faster than sequential reads when pulling data from dozens of VAV boxes or AHU points during commissioning verification.

async def read_all_zones(client, zone_node_ids):
    """Read multiple zone temperatures concurrently."""
    nodes = [client.get_node(nid) for nid in zone_node_ids]
    tasks = [node.read_value() for node in nodes]
    values = await asyncio.gather(*tasks)
    return dict(zip(zone_node_ids, values))

Writing Values

Writing values to OPC UA nodes is how you push setpoint changes, override damper positions, or adjust schedule parameters from a Python script. The write_value() method accepts a Python value and optionally an explicit OPC UA variant type.

from asyncua import Client, ua

async def main():
    url = "opc.tcp://192.168.1.50:4840"
    async with Client(url=url) as client:
        # Write a float value (e.g., cooling setpoint)
        node = client.get_node("ns=2;s=Building1/VAV-301/ClgSetpoint")
        await node.write_value(72.0)
        print("Setpoint updated to 72.0")

        # Write with explicit type (required for some servers)
        await node.write_value(
            ua.Variant(72.0, ua.VariantType.Double)
        )

        # Write a boolean (e.g., enable/disable a schedule)
        schedule_node = client.get_node(
            "ns=2;s=Building1/AHU-01/OccupiedMode"
        )
        await schedule_node.write_value(True)

        # Batch write: update setpoints for multiple VAVs
        vav_setpoints = {
            "ns=2;s=Building1/VAV-301/ClgSetpoint": 72.0,
            "ns=2;s=Building1/VAV-302/ClgSetpoint": 73.0,
            "ns=2;s=Building1/VAV-303/ClgSetpoint": 71.5,
        }
        for node_id, sp in vav_setpoints.items():
            node = client.get_node(node_id)
            await node.write_value(ua.Variant(sp, ua.VariantType.Double))
        print(f"Updated {len(vav_setpoints)} setpoints")

asyncio.run(main())

Write permissions matter. Most BAS servers restrict write access based on the authenticated user's role. If your writes fail silently or return a BadNotWritable status code, verify that the account you are connecting with has operator or engineer-level privileges in the BAS platform's user management.

Subscribing to Data Changes

Polling every point on a timer wastes bandwidth and misses transient events. OPC UA subscriptions let the server push updates to your client only when a value actually changes. This is the preferred approach for real-time dashboards, alarm monitoring, and logging scripts that run for hours or days during commissioning.

import asyncio
from asyncua import Client

class BASDataHandler:
    """Handler that receives push notifications on data changes."""

    def datachange_notification(self, node, val, data):
        """Called by asyncua when a subscribed node value changes."""
        print(
            f"Data change: {node} = {val} "
            f"(source time: {data.monitored_item.Value.SourceTimestamp})"
        )

async def main():
    url = "opc.tcp://192.168.1.50:4840"
    async with Client(url=url) as client:
        handler = BASDataHandler()

        # Create a subscription with 1000ms publishing interval
        subscription = await client.create_subscription(
            period=1000,
            handler=handler
        )

        # Subscribe to specific BAS nodes
        nodes = [
            client.get_node("ns=2;s=Building1/AHU-01/SupplyAirTemp"),
            client.get_node("ns=2;s=Building1/AHU-01/ReturnAirTemp"),
            client.get_node("ns=2;s=Building1/AHU-01/DamperPosition"),
        ]
        handles = await subscription.subscribe_data_change(nodes)

        # Keep running to receive notifications
        print("Listening for data changes (Ctrl+C to stop)...")
        try:
            while True:
                await asyncio.sleep(1)
        except KeyboardInterrupt:
            pass

        # Clean up
        await subscription.unsubscribe(handles)
        await subscription.delete()

asyncio.run(main())

The period parameter (in milliseconds) controls how often the server bundles and sends notifications. A 1000ms interval works well for HVAC trending. For fast-moving values like variable frequency drive speeds, you can lower it to 250ms or less, but be mindful of network load on constrained BAS networks.

Common Mistakes

1. Forgetting await on async calls. asyncua is built entirely on asyncio. Calling node.read_value() without await returns a coroutine object instead of the actual sensor reading. If your output shows <coroutine object Node.read_value ...>, add the missing await.
2. Using the wrong NodeId namespace index. OPC UA servers assign namespace indices dynamically. A node that is ns=2 on one server restart might become ns=3 after a configuration change. Always resolve the namespace index at runtime using idx = await client.get_namespace_index("http://your.namespace.uri") instead of hardcoding the integer.
3. Certificate URI mismatch when using security. When configuring encrypted connections, the application URI in your client certificate must match the URI you set in the asyncua client (client.application_uri). A mismatch triggers BadCertificateUriInvalid errors that can be difficult to diagnose because some servers reject silently while others log vague warnings.
4. Not handling session timeouts on long-running scripts. OPC UA sessions expire if the client does not send keep-alive requests within the server's timeout window. asyncua handles this automatically, but if your script blocks the asyncio event loop with synchronous code (e.g., a long time.sleep() instead of await asyncio.sleep()), the keep-alive coroutine cannot run and the server drops the session.
5. Installing the deprecated opcua package instead of asyncua. The older pip install opcua installs the unmaintained python-opcua library, which lacks Python 3.12+ support and no longer receives bug fixes. The actively maintained library is pip install asyncua. If you see import errors referencing opcua.ua in online examples, translate them to asyncua.ua.

Platform Compatibility

BAS platform compatibility: asyncua works with any OPC UA server that conforms to the IEC 62541 standard. In building automation, this includes Tridium Niagara 4 (with its OPC UA module), Siemens Desigo CC, Johnson Controls Metasys, Honeywell EBI, Schneider EcoStruxure, and dedicated OPC UA gateways like Kepware KEPServerEX and Matrikon OPC UA. The server must have its OPC UA endpoint enabled and accessible on the network—consult your platform's documentation for the specific configuration steps.

Firewall considerations: OPC UA uses TCP (default port 4840). Unlike BACnet's UDP-based transport, OPC UA connections are stateful TCP sessions, which pass through most firewalls more reliably. However, you still need to ensure that port 4840 (or whichever port your server uses) is open between your Python client and the BAS server.

Source Attribution

This tutorial draws from the following sources:

• FreeOpcUa/opcua-asyncio GitHub Repository — source code, examples, and issue tracker (LGPL-3.0 license, 1,100+ stars)
• asyncua ReadTheDocs — official API documentation, minimal client and server examples, and usage guides
• Implementing Industrial OPC UA Communication with Python and Asyncio — practical tutorial covering client/server patterns with asyncua
• Industrial Automation with Python and OPC UA (Medium) — comprehensive guide covering subscriptions and data change handling

asyncua is developed by the FreeOpcUa community and licensed under the GNU Lesser General Public License v3.0 (LGPL-3.0). You may use it freely in commercial and proprietary projects as long as modifications to the asyncua library itself remain open-source per the LGPL terms.