By the end of this section, you will be able to:
Before diving into this chapter, you should be familiar with:
The Problem: In a typical factory, you might have: - PLCs from Siemens, Rockwell, Mitsubishi - Sensors from 10 different manufacturers - SCADA software from one vendor - MES software from another - Cloud analytics from yet another
Each speaks a different “language” (protocol). Getting them to talk to each other traditionally required custom integration for every pair—expensive and fragile.
OPC-UA’s Solution: One universal language that everyone speaks.
Analogy: OPC-UA is like English in international business: - Everyone agrees to communicate in English (OPC-UA) - Each company still has their native language internally (Modbus, PROFINET, etc.) - But when they need to share information, they use English (OPC-UA)
Key Benefits:
“Every factory machine speaks a different language!” complained Sammy the Sensor. “The PLC speaks Modbus, the robot speaks PROFINET, and the quality system speaks HTTP. How do they communicate?”
Max the Microcontroller introduced OPC-UA: “That’s exactly the problem OPC-UA solves! It’s like a universal translator that every device can speak. Instead of building custom adapters between each pair of systems, everything connects through OPC-UA. It defines a standard way to describe data – not just the value, but its type, unit, timestamp, and quality.”
“The cool part is the information model,” said Lila the LED. “OPC-UA doesn’t just say ‘temperature is 25.’ It says ‘this is a temperature measurement, in Celsius, from sensor #3, on production line 2, taken at 10:30 AM, with good quality.’ All that context is built into the protocol.”
Bella the Battery appreciated the security: “OPC-UA has built-in encryption and authentication from the ground up. Unlike Modbus, which was designed in 1979 with zero security, OPC-UA was built for the modern connected factory where cyber-attacks are a real threat. It’s the future of industrial IoT communication!”
This diagram shows OPC-UA’s information model structure: Objects contain Variables (data) and Methods (actions), connected by typed References that define relationships.
The traditional OPC-UA communication pattern:
Core Concept: The OPC-UA information model is a structured address space where all data is represented as typed nodes connected by semantic references, enabling machines to understand not just values but their meaning and relationships.
Why It Matters: Unlike flat register-based protocols (Modbus), OPC-UA’s information model allows devices to describe themselves. A client connecting to an unknown server can browse its address space, discover what data is available, understand data types and units, and call methods—all without prior configuration or documentation.
Key Takeaway: Think of nodes as objects in object-oriented programming: they have properties (attributes), can contain other nodes (composition), inherit from types (polymorphism), and expose callable functions (methods).
Everything in OPC-UA is a Node. There are eight node classes:
| Node Class | Purpose | Example |
|---|---|---|
| Object | Container for other nodes | Device, Folder |
| Variable | Data value with type | Temperature, Status |
| Method | Callable function | Calibrate(), Start() |
| ObjectType | Template for objects | DeviceType, SensorType |
| VariableType | Template for variables | AnalogItemType |
| ReferenceType | Relationship definition | HasComponent, Organizes |
| DataType | Data type definition | Int32, String, custom |
| View | Subset of address space | Production view |
Every node has a unique identifier (NodeId):
| Format | Example | Use Case |
|---|---|---|
| Numeric | i=2253 |
Standard OPC-UA nodes |
| String | s=MyDevice.Temperature |
Application nodes |
| GUID | g=09087e75-8e5e-499b-954f-... |
Unique identifiers |
| Opaque | b=M/RbKBsRVk... |
Binary data |
Namespace Index:
ns=0: OPC-UA standard namespacens=1: Server-specificns=2+: Application namespacesNodes are connected by typed references:
| Reference Type | Meaning | Example |
|---|---|---|
| HasComponent | Contains as part | Device → Sensor |
| HasProperty | Has metadata | Variable → EngineeringUnits |
| Organizes | Logical grouping | Folder → Items |
| HasTypeDefinition | Instance of type | Sensor → SensorType |
| HasSubtype | Type inheritance | SensorType → TemperatureSensorType |
# Python example using asyncua library
from asyncua import Client
async def discover_server():
# Get endpoints from discovery URL
client = Client("opc.tcp://localhost:4840")
endpoints = await client.get_endpoints()
for ep in endpoints:
print(f"Endpoint: {ep.EndpointUrl}")
print(f"Security: {ep.SecurityPolicyUri}")
print(f"Mode: {ep.SecurityMode}")async def read_write_example():
client = Client("opc.tcp://localhost:4840")
await client.connect()
# Read a variable
node = client.get_node("ns=2;s=Temperature")
value = await node.read_value()
print(f"Temperature: {value}")
# Read multiple attributes
attrs = await node.read_attributes([
ua.AttributeIds.Value,
ua.AttributeIds.DataType,
ua.AttributeIds.DisplayName
])
# Write a value
setpoint = client.get_node("ns=2;s=Setpoint")
await setpoint.write_value(75.0)
await client.disconnect()class DataChangeHandler:
def datachange_notification(self, node, val, data):
print(f"Data change: {node} = {val}")
async def subscribe_example():
client = Client("opc.tcp://localhost:4840")
await client.connect()
# Create subscription
handler = DataChangeHandler()
subscription = await client.create_subscription(500, handler) # 500ms interval
# Monitor nodes
temp_node = client.get_node("ns=2;s=Temperature")
await subscription.subscribe_data_change(temp_node)
# Keep running to receive notifications
await asyncio.sleep(3600)async def call_method():
client = Client("opc.tcp://localhost:4840")
await client.connect()
# Find the method
device = client.get_node("ns=2;s=Device1")
calibrate = client.get_node("ns=2;s=Device1.Calibrate")
# Call with arguments
result = await device.call_method(calibrate, 25.0, "high")
print(f"Calibration result: {result}")These worked examples demonstrate practical OPC-UA implementation decisions for real-world industrial scenarios.
How does OPC-UA’s security overhead affect message throughput? Consider a factory monitoring system polling 100 sensors every 100 ms with different security policies.
Scenario: Each sensor reading is 8 bytes (float32 + timestamp). OPC-UA wraps this in protocol headers and encryption.
No Security (Sign=None, Encrypt=None): - OPC-UA binary header: 24 bytes - Service header: 16 bytes - Total per reading: \(8 + 24 + 16 = 48\text{ bytes}\)
Sign Only (Basic128Rsa15): - Adds SHA1 signature: 20 bytes - Total: \(48 + 20 = 68\text{ bytes}\) (42% overhead)
Sign & Encrypt (Basic256Sha256): - AES-256-CBC encryption padding: ~16 bytes average - SHA-256 signature: 32 bytes - Total: \(48 + 16 + 32 = 96\text{ bytes}\) (100% overhead)
Network impact (100 sensors × 10 Hz): - No security: \(48 \times 100 \times 10 = 48\text{ kB/s}\) - Sign & encrypt: \(96 \times 100 \times 10 = 96\text{ kB/s}\)
On 100 Mbps Ethernet: 96 kB/s is only 0.077% utilization - negligible! But on Zigbee (250 kbps): 96 kB/s = 3,072 kbps needed = 12× over capacity. This is why lightweight protocols (MQTT, CoAP) dominate battery/wireless IoT while OPC-UA dominates wired industrial automation!
Scenario: A beverage bottling line has 20 machines (fillers, cappers, labelers, packers) that must be monitored with sub-second latency. The production manager needs a real-time dashboard showing machine states, current speeds, and fault conditions. Historical data must be logged for quality audits.
Given:
Steps:
Configure subscriptions for real-time monitoring:
from asyncua import Client, ua
async def setup_monitoring():
client = Client("opc.tcp://plc.factory.local:4840")
await client.connect()
# Create subscription with 100ms publishing interval
# This gives us 200ms worst-case latency (100ms sampling + 100ms publish)
subscription = await client.create_subscription(
period=100, # 100ms publishing interval
handler=DashboardHandler()
)
# Monitor all critical variables
machines = ['Filler01', 'Capper01', 'Labeler01', 'Packer01']
monitored_items = []
for machine in machines:
for var in ['Speed', 'State', 'FaultCode', 'TotalCount']:
node = client.get_node(f"ns=2;s=Line1.{machine}.{var}")
handle = await subscription.subscribe_data_change(
node,
queuesize=1, # Only keep latest value
sampling_interval=100 # 100ms at server
)
monitored_items.append(handle)
print(f"Monitoring {len(monitored_items)} variables at 100ms intervals")
return subscription
class DashboardHandler:
def datachange_notification(self, node, val, data):
# Called automatically when any monitored value changes
node_id = node.nodeid.Identifier
timestamp = data.monitored_item.Value.SourceTimestamp
# Update dashboard widget
dashboard.update(node_id, val, timestamp)
# Check for faults
if 'FaultCode' in node_id and val != 0:
send_operator_alert(node_id, val)Implement historical logging with batch writes:
import asyncio
from collections import deque
class HistorianLogger:
def __init__(self, batch_size=100, flush_interval=5):
self.buffer = deque(maxlen=10000)
self.batch_size = batch_size
self.flush_interval = flush_interval
def datachange_notification(self, node, val, data):
# Buffer values for batch writing
self.buffer.append({
'node_id': str(node.nodeid),
'value': val,
'timestamp': data.monitored_item.Value.SourceTimestamp,
'quality': data.monitored_item.Value.StatusCode.value
})
# Flush if batch is full
if len(self.buffer) >= self.batch_size:
asyncio.create_task(self.flush_to_database())
async def flush_to_database(self):
if not self.buffer:
return
# Batch insert to time-series database
batch = [self.buffer.popleft()
for _ in range(min(self.batch_size, len(self.buffer)))]
await timescaledb.insert_batch('opc_history', batch)Result: The dashboard receives updates within 200ms (100ms server sampling + 100ms publish interval). Fault codes trigger immediate operator alerts. Historical data is buffered and batch-written to TimescaleDB every 5 seconds or 100 records, ensuring efficient database writes without blocking real-time updates.
Key Insight: Separate real-time monitoring from historical logging using different handlers on the same subscription. Real-time dashboards need immediate processing with minimal buffering (queuesize=1), while historians should batch writes to avoid overwhelming the database. OPC-UA subscriptions handle both patterns elegantly by pushing data changes to registered handlers.
Scenario: A pharmaceutical manufacturing plant must send batch records to a cloud-based quality management system. The cloud connection is unreliable (occasional 5-minute outages). Regulatory requirements (FDA 21 CFR Part 11) mandate that no batch data is ever lost, and all data must include tamper-proof timestamps.
Given:
Steps:
Design edge gateway with local persistence:
from asyncua import Client
import sqlite3
import json
class PharmaBridge:
def __init__(self):
# Local SQLite for store-and-forward
self.db = sqlite3.connect('batch_buffer.db')
self.db.execute('''
CREATE TABLE IF NOT EXISTS pending_batches (
id INTEGER PRIMARY KEY,
reactor_id TEXT,
batch_id TEXT UNIQUE,
data JSON,
source_timestamp TEXT,
received_timestamp TEXT,
signature TEXT,
upload_status TEXT DEFAULT 'pending'
)
''')
async def on_batch_complete(self, node, val, data):
"""Called when BatchComplete event fires on any reactor"""
reactor_id = extract_reactor_id(node.nodeid)
batch_id = val['BatchID']
# Read complete batch record from OPC-UA server
batch_data = await self.read_batch_record(reactor_id, batch_id)
# Preserve original OPC-UA timestamps (FDA requirement)
source_timestamp = data.monitored_item.Value.SourceTimestamp
# Sign the record for tamper detection
signature = self.sign_record(batch_data, source_timestamp)
# Store locally first (guaranteed persistence)
self.db.execute('''
INSERT INTO pending_batches
(reactor_id, batch_id, data, source_timestamp,
received_timestamp, signature)
VALUES (?, ?, ?, ?, ?, ?)
''', (
reactor_id,
batch_id,
json.dumps(batch_data),
source_timestamp.isoformat(),
datetime.utcnow().isoformat(),
signature
))
self.db.commit()
# Attempt immediate upload
await self.upload_pending()Implement reliable cloud upload with retry:
async def upload_pending(self):
"""Upload pending batches to cloud QMS"""
pending = self.db.execute('''
SELECT id, reactor_id, batch_id, data, source_timestamp, signature
FROM pending_batches
WHERE upload_status = 'pending'
ORDER BY source_timestamp
''').fetchall()
for record in pending:
try:
# Upload to cloud with original timestamps
response = await cloud_qms.upload_batch({
'reactor_id': record[1],
'batch_id': record[2],
'data': json.loads(record[3]),
'source_timestamp': record[4], # Original OPC-UA time
'signature': record[5]
})
if response.status == 200:
# Mark as uploaded, keep for audit trail
self.db.execute('''
UPDATE pending_batches
SET upload_status = 'uploaded',
upload_timestamp = ?
WHERE id = ?
''', (datetime.utcnow().isoformat(), record[0]))
self.db.commit()
log.info(f"Batch {record[2]} uploaded successfully")
except ConnectionError:
log.warning(f"Cloud unavailable, batch {record[2]} queued")
# Will retry on next cycle
break # Stop trying others until connection restored
async def retry_loop(self):
"""Background task: retry pending uploads every 30 seconds"""
while True:
await asyncio.sleep(30)
await self.upload_pending()Configure OPC-UA subscription for batch events:
async def connect_reactors(self):
for reactor in ['Reactor1', 'Reactor2', 'Reactor3', 'Reactor4', 'Reactor5']:
client = Client(f"opc.tcp://{reactor}.pharma.local:4840")
# Use SignAndEncrypt for FDA compliance
await client.set_security(
security_policy=SecurityPolicy.Basic256Sha256,
mode=MessageSecurityMode.SignAndEncrypt,
certificate="gateway.der",
private_key="gateway.pem"
)
await client.connect()
# Subscribe to BatchComplete events
subscription = await client.create_subscription(
period=1000, # 1 second is fine for batch events
handler=self
)
event_node = client.get_node(f"ns=2;s={reactor}.BatchCompleteEvent")
await subscription.subscribe_events(event_node)Result: Batch records are stored locally immediately upon completion with cryptographic signatures preserving data integrity. The background retry loop uploads to cloud every 30 seconds. During a 5-minute cloud outage, up to 10 batch completions are buffered locally and uploaded automatically when connectivity resumes. Original OPC-UA timestamps are preserved for FDA audit trail requirements.
Key Insight: Store-and-forward is essential for reliable cloud connectivity. Always persist data locally before attempting cloud upload, especially for compliance-critical applications. OPC-UA’s built-in timestamps (SourceTimestamp) provide the authoritative record time for audit trails - preserve these rather than using gateway arrival time. The edge gateway acts as a reliability buffer between deterministic OT systems and best-effort IT networks.
| Mode | Encryption | Signing | Use Case |
|---|---|---|---|
| None | ❌ | ❌ | Development only |
| Sign | ❌ | ✅ | Integrity protection |
| SignAndEncrypt | ✅ | ✅ | Production systems |
| Policy | Key Size | Algorithm | Status |
|---|---|---|---|
| None | - | - | Deprecated |
| Basic128Rsa15 | 1024 | RSA, AES-128 | Deprecated |
| Basic256 | 2048 | RSA, AES-256 | Legacy |
| Basic256Sha256 | 2048 | RSA, SHA-256 | Recommended |
| Aes128_Sha256_RsaOaep | 2048 | AES-128, SHA-256 | Current |
| Aes256_Sha256_RsaPss | 4096 | AES-256, SHA-256 | High security |
For scalable, one-to-many communication:
| Aspect | Client-Server | Pub/Sub |
|---|---|---|
| Connection | Direct, stateful | Indirect via broker |
| Scaling | 1:1 connections | 1:N distribution |
| Latency | Lower | Higher (broker hop) |
| Reliability | Session-based | Message-based QoS |
| Firewall | Requires inbound | Outbound only |
| Use Case | Configuration, methods | Data distribution |
OPC-UA is extended through standardized companion specifications:
| Specification | Domain | Key Features |
|---|---|---|
| OPC 40001 | Devices (DI) | Device identification, diagnostics |
| OPC 40010 | Robotics | Robot programs, motion control |
| OPC 40100 | Injection Molding | Machine cycles, parameters |
| OPC 40083 | Weighing | Weighing instruments |
| OPC 30000 | ISA-95 | Manufacturing operations |
| OPC 40223 | Pumps | Pump monitoring, control |
from asyncua import Server, ua
import asyncio
async def main():
server = Server()
await server.init()
server.set_endpoint("opc.tcp://0.0.0.0:4840/freeopcua/server/")
idx = await server.register_namespace("http://example.org/sensor")
objects = server.get_objects_node()
device = await objects.add_object(idx, "TemperatureSensor")
# Add variables and methods to the address space
temperature = await device.add_variable(idx, "Temperature", 25.0)
await temperature.set_writable()
status = await device.add_variable(idx, "Status", "Running")
async def calibrate(parent, offset):
current = await temperature.read_value()
await temperature.write_value(current + offset)
return f"Calibrated with offset {offset}"
await device.add_method(idx, "Calibrate", calibrate,
[ua.VariantType.Float], [ua.VariantType.String])
async with server: # Start server and simulate readings
while True:
await asyncio.sleep(1)
current = await temperature.read_value()
await temperature.write_value(current + 0.1)
asyncio.run(main())
The Mistake: Using default browser WebSocket timeouts (30-60 seconds) for OPC-UA WebSocket transport, causing sessions to drop unexpectedly when operators leave dashboards open during shift breaks.
Why It Happens: OPC-UA over WebSocket (for browser-based HMI/SCADA) inherits HTTP/WebSocket timeout behavior. Developers assume the OPC-UA session keep-alive handles everything, but proxy servers, load balancers, and browser idle detection can close the underlying WebSocket independently of OPC-UA session state.
The Fix: Configure timeouts at all layers to match OPC-UA session lifetime:
// Browser client: Send WebSocket ping frames
const ws = new WebSocket('wss://opcua.factory.local/ua');
const PING_INTERVAL = 15000; // 15 seconds (less than typical 30s timeout)
setInterval(() => {
if (ws.readyState === WebSocket.OPEN) {
ws.send(new Uint8Array([0x89, 0x00])); // WebSocket ping frame
}
}, PING_INTERVAL);
// Server-side (nginx): Extend WebSocket timeouts
location /ua {
proxy_pass http://opcua_backend;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_read_timeout 3600s; # 1 hour for long-running dashboards
proxy_send_timeout 3600s;
}
// OPC-UA session: Match session timeout to WebSocket config
session.requestedSessionTimeout = 3600000; // 1 hour in millisecondsAlso implement automatic WebSocket reconnection with OPC-UA session reactivation to handle inevitable disconnects gracefully.
The Mistake: Exposing OPC-UA subscriptions through an HTTP REST gateway without rate limiting, allowing a single high-frequency subscription (e.g., 100ms updates) to overwhelm the REST API and block other clients.
Why It Happens: OPC-UA subscriptions can push updates at 10-100Hz for real-time monitoring. When bridged to REST (for mobile apps or web dashboards), each OPC-UA notification becomes an HTTP response. Without buffering, 100 updates/second from one PLC floods the gateway, exhausting connection pools and starving other API consumers.
The Fix: Implement subscription aggregation and rate limiting at the gateway:
# BAD: Direct 1:1 OPC-UA notification to HTTP response
def on_datachange(node, value):
# Called 100 times/second - floods HTTP clients!
broadcast_to_all_http_clients(node, value)
# GOOD: Aggregate and throttle with configurable intervals
class SubscriptionAggregator:
def __init__(self, min_interval_ms=100):
self.buffer = {}
self.min_interval = min_interval_ms / 1000
self.last_send = {}
def on_datachange(self, node, value):
node_id = str(node.nodeid)
self.buffer[node_id] = value # Always keep latest
async def publish_loop(self):
while True:
await asyncio.sleep(self.min_interval)
if self.buffer:
# Send batch update to HTTP clients
batch = dict(self.buffer)
self.buffer.clear()
await broadcast_batch(batch) # One HTTP response with all changes
# Also configure OPC-UA subscription parameters
subscription = await client.create_subscription(
period=500, # 500ms publishing interval (not faster)
handler=aggregator,
sampling_interval=100 # 100ms sampling, but batched publishes
)This reduces HTTP traffic by 5-10x while maintaining data freshness. Configure the minimum interval based on dashboard refresh requirements (typically 100-500ms is sufficient for human viewing).
Scenario: You’re implementing an OPC-UA solution for a pharmaceutical manufacturing line with: - 5 mixing tanks with temperature, pressure, level sensors - 2 packaging machines with speed, count, status - Requirement: All data must be encrypted - Cloud analytics needs access for batch reporting
Questions:
:
OPC-UA is the industrial interoperability standard—platform-independent, secure, extensible
Information Model: Everything is a Node with typed references forming an address space tree
Services: Discovery -> Secure Channel -> Session -> Read/Write/Subscribe/Call
Security is built-in: Certificates, encryption, authentication at protocol level
Pub/Sub enables scaling: MQTT/AMQP for one-to-many data distribution
OPC-UA and MQTT are complementary: OPC-UA for factory-floor depth, MQTT for cloud-scale breadth
Use OPC-UA as the IT/OT bridge—connect legacy protocols to modern cloud systems
OPC-UA represents the intersection of industrial control, IT/OT convergence, and semantic interoperability:
Information Model is the Foundation:
Objects/DeviceSet/TemperatureSensor1/Temperature with type AnalogItemType, engineering units metadata, and quality flags.Service Sets Build on Information Model:
Client-Server vs Pub/Sub Trade-off:
OPC-UA + MQTT Bridge Pattern:
Companion Specifications Extend Base Model:
InjectionMoldingMachineType with standardized Variables (cycle time, cavity pressure) and Methods (StartProduction, Abort). All vendors implement same semantic model - plug-and-play interoperability.Security Architecture Differences:
TSN Convergence (Future):
Key Insight: OPC-UA is NOT just “another protocol” - it’s an integration framework. The information model provides semantic interoperability (what does “temperature” mean?), services provide access patterns (browse/read/subscribe/call), security provides trust, and pub/sub provides scalability. Choosing OPC-UA means choosing standardized integration over custom parsers and proprietary APIs.
Protocol Comparisons:
spBv1.0/group_id/DDATA/edge_node_id/device_id (Sparkplug B convention). OPC-UA adds semantic model; MQTT adds pub/sub scalability.Information Modeling:
PackMLBaseStateType with specific state machine). Both needed for plug-and-play.Security Deep Dive:
Real-World Deployments:
Hands-On Exploration:
Advanced Topics:
Bridging IT and OT:
| Chapter | Focus | Why Read It |
|---|---|---|
| Modbus Protocol | Legacy register-based polling protocol | Assess when Modbus RTU/TCP is sufficient versus when OPC-UA’s richer model is required |
| PROFINET and Industrial Ethernet | Real-time deterministic Ethernet (PROFINET IRT, EtherCAT, TSN) | Understand how OPC-UA over TSN combines semantic integration with sub-millisecond control |
| MQTT Fundamentals | Lightweight pub/sub for IoT telemetry | Design the OPC-UA-to-MQTT bridge pattern that connects factory-floor OPC-UA servers to cloud platforms |
| Industrial Protocols Overview | Landscape of IT/OT convergence protocols | Place OPC-UA within the broader hierarchy of industrial communication standards |
| Edge and Fog Computing Overview | Edge gateway architectures | Implement OPC-UA aggregation servers and store-and-forward gateways at the network edge |
| Stream Processing | Real-time data pipelines in the cloud | Connect OPC-UA telemetry data to cloud analytics and time-series databases |