Python Diagnostic Scripts for B&R CP1584: PVI and OPC-UA

Overview

This document covers building automated diagnostic, monitoring, and regression scripts for B&R CP1584 PLCs using Python. Two primary communication paths exist: OPC-UA (runs directly on the controller, no Windows middleware needed) and PVI (requires B&R PVI Manager on Windows). For a one-man automation engineer maintaining undocumented machines, Python scripts are force multipliers — they automate repetitive diagnostic tasks, capture trends, and generate reports.

See also: pvi-api.md, opcua.md, cp1584-forensics.md

1. Communication Path Comparison

Recommendation: If the OPC-UA server is enabled on the CP1584 (check port 4840 with nmap), use OPC-UA via asyncua. It requires no Windows middleware and no B&R PVI license. If OPC-UA is not enabled, use PVI from a Windows machine.

⚠️ PVI 6.x Breaking Change: INA2000 Removed

If your engineering workstation runs Automation Studio 6 (PVI 6.x), the legacy INA2000 protocol is completely removed — only ANSL and SNMP lines are supported. Any existing PVI-based scripts using pvipy with INA2000 connections will fail with missing DLL errors. You must update all connections to use ANSL (TCP port 1212, backward-compatible with all AR ≥ V4.08 including CP1584). See pvi-api.md §2 for the full migration checklist.

2. Prerequisites and Environment Setup

2.1 OPC-UA Path (Recommended)

pip install asyncua

## For data logging to CSV and reporting
pip install pandas matplotlib

## For scheduling automated scripts
pip install schedule

Verify OPC-UA is enabled on the PLC:

nmap -p 4840 <PLC_IP>
# PORT     STATE SERVICE
# 4840/tcp open  http-alt

If port 4840 is open, OPC-UA is running. Test anonymous access:

import asyncio
from asyncua import Client

async def test_connection():
    url = "opc.tcp://<PLC_IP>:4840"
    async with Client(url=url, timeout=5) as client:
        root = client.get_root_node()
        print(f"Connected! Root node: {root}")
        children = await root.get_children()
        print(f"Root has {len(children)} children")

asyncio.run(test_connection())

2.2 PVI Path (Windows Only)

Install B&R PVI Development Setup:

1. Download PVI Development Setup from B&R’s website (requires a B&R portal account)
2. Install PVI Development Setup on your Windows machine
3. Without a PVI license, PVI runs in trial mode (2-hour limit, then PVI Manager must be restarted)

PVI License: Order code 1TG0500.02 (+ TG Guard 0TG1000.02). Without it, the 2-hour trial cycle is a significant limitation for long-running monitoring scripts.

pip install pvipy

PVI Manager (PviMan.exe) must be running as a Windows service or process. It acts as the middleware between your Python code and the PLC.

See also: pvi-api.md for full PVI architecture details

3. OPC-UA: Browsing the Variable Namespace

When you have an undocumented CP1584, the first step is discovering what variables are available via OPC-UA.

3.1 Recursive Namespace Browser

import asyncio
from asyncua import Client

PLC_URL = "opc.tcp://192.168.1.100:4840"

async def browse_namespace(node, indent=0):
    children = await node.get_children()
    for child in children:
        name = await child.get_browse_name()
        nodeid = child.nodeid
        display = await child.get_display_name()
        dtype = await child.get_data_type()
        accessible = await child.read_attribute(0x1f)  # AccessLevel
        
        print(f"{'  ' * indent}{name.Name} [{nodeid}] - {display.Text}")
        
        if len(children) < 50:
            await browse_namespace(child, indent + 1)

async def main():
    async with Client(url=PLC_URL, timeout=10) as client:
        print("=== B&R OPC-UA Namespace Browser ===")
        root = client.get_root_node()
        print(f"Root: {await root.get_browse_name().Name}")
        
        objects = client.nodes.objects
        print(f"\nObjects node children:")
        await browse_namespace(objects)

asyncio.run(main())

3.2 Extract All Variable Names to CSV

For documentation purposes, dump the entire variable namespace to a CSV:

import asyncio
import csv
from asyncua import Client
from datetime import datetime

PLC_URL = "opc.tcp://<PLC_IP>:4840"
OUTPUT_FILE = "br_namespace_dump.csv"

async def scan_node(node, writer, depth=0, max_depth=6):
    if depth > max_depth:
        return
    
    try:
        children = await node.get_children()
    except Exception:
        return
    
    for child in children:
        try:
            browse_name = await child.get_browse_name()
            display_name = await child.get_display_name()
            node_class = await child.read_node_class()
            
            row = {
                "timestamp": datetime.now().isoformat(),
                "nodeid": str(child.nodeid),
                "browse_name": browse_name.Name,
                "display_name": display_name.Text,
                "node_class": str(node_class),
                "depth": depth,
            }
            
            if node_class.name == "Variable":
                try:
                    val = await child.get_value()
                    row["value"] = str(val)
                except Exception:
                    row["value"] = "N/A"
                
                try:
                    dtype = await child.get_data_type()
                    row["data_type"] = str(dtype)
                except Exception:
                    row["data_type"] = "Unknown"
            
            writer.writerow(row)
            await scan_node(child, writer, depth + 1, max_depth)
        except Exception as e:
            pass

async def main():
    async with Client(url=PLC_URL, timeout=10) as client:
        objects = client.nodes.objects
        with open(OUTPUT_FILE, "w", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=[
                "timestamp", "nodeid", "browse_name", "display_name",
                "node_class", "depth", "value", "data_type"
            ])
            writer.writeheader()
            await scan_node(objects, writer)

asyncio.run(main())
print(f"Namespace dumped to {OUTPUT_FILE}")

3.3 B&R Variable Naming Conventions

B&R OPC-UA variable names follow these patterns:

The namespace index (ns=X;) varies depending on the OPC-UA configuration. Common indices:

• ns=0 — OPC-UA standard
• ns=1 or ns=2 — B&R default for application variables
• Higher indices for specific OPC-UA views or mapp components

4. OPC-UA: Reading and Writing Variables

4.1 Read Single Variable

import asyncio
from asyncua import Client

PLC_URL = "opc.tcp://<PLC_IP>:4840"

async def read_variable(node_id):
    async with Client(url=PLC_URL, timeout=5) as client:
        node = client.get_node(node_id)
        value = await node.get_value()
        name = await node.get_browse_name()
        print(f"{name.Name} = {value} (type: {type(value).__name__})")
        return value

asyncio.run(read_variable("ns=2;s=::AsGlobalPV:MachineSpeed"))

4.2 Read Multiple Variables (Batch)

import asyncio
from asyncua import Client

PLC_URL = "opc.tcp://<PLC_IP>:4840"

VARIABLES = {
    "MachineSpeed": "ns=2;s=::AsGlobalPV:MachineSpeed",
    "EmergencyStop": "ns=2;s=::MainProgram:EmergencyStop",
    "MotorRunning": "ns=2;s=::MainProgram:MotorControl.RunFlag",
    "CycleTime_ms": "ns=2;s=::MainProgram:CycleTime",
    "PartCount": "ns=2;s=::AsGlobalPV:PartCounter",
}

async def read_all():
    async with Client(url=PLC_URL, timeout=5) as client:
        print(f"{'Variable':<25} {'Value':<20} {'Type'}")
        print("-" * 60)
        for name, node_id in VARIABLES.items():
            try:
                node = client.get_node(node_id)
                value = await node.get_value()
                print(f"{name:<25} {str(value):<20} {type(value).__name__}")
            except Exception as e:
                print(f"{name:<25} ERROR: {e}")

asyncio.run(read_all())

4.3 Write Variable (B&R-Specific Method)

B&R OPC-UA servers require a specific write pattern. The standard node.write_value() may return BadWriteNotSupported even when the variable is writable in UAExpert. Use this workaround:

import asyncio
from asyncua import Client, ua
from asyncua import Node

PLC_URL = "opc.tcp://<PLC_IP>:4840"

async def write_br_variable(node_id, value):
    async with Client(url=PLC_URL, timeout=5) as client:
        node = client.get_node(node_id)
        
        attr = ua.WriteValue()
        attr.NodeId = node.nodeid
        attr.AttributeId = ua.AttributeIds.Value
        attr.Value = ua.DataValue(value)
        
        params = ua.WriteParameters()
        params.NodesToWrite = [attr]
        
        result = await node.write_params(params)
        
        if result[0].is_good():
            print(f"Write successful: {value}")
        else:
            print(f"Write failed: {result[0]}")

asyncio.run(write_br_variable("ns=2;s=::AsGlobalPV:ManualOverride", True))

Common B&R OPC-UA write errors:

4.4 Force Outputs via OPC-UA

Forcing outputs lets you override physical I/O for diagnostic purposes:

import asyncio
from asyncua import Client, ua

PLC_URL = "opc.tcp://<PLC_IP>:4840"

async def force_output(node_id, value):
    async with Client(url=PLC_URL, timeout=5) as client:
        node = client.get_node(node_id)
        
        attr = ua.WriteValue()
        attr.NodeId = node.nodeid
        attr.AttributeId = ua.AttributeIds.Value
        attr.Value = ua.DataValue(value)
        
        params = ua.WriteParameters()
        params.NodesToWrite = [attr]
        
        result = await node.write_params(params)
        print(f"Force {node_id} = {value}: {'OK' if result[0].is_good() else str(result[0])}")

asyncio.run(force_output("ns=2;s=::MainProgram:Output1", True))

See also: online-changes.md, io-sniffing.md

5. OPC-UA: Subscription-Based Monitoring

Subscriptions are the OPC-UA mechanism for real-time variable monitoring. The server pushes data changes to the client instead of the client polling.

5.1 Basic Subscription

import asyncio
from asyncua import Client

PLC_URL = "opc.tcp://<PLC_IP>:4840"

async def monitor_variables():
    async with Client(url=PLC_URL, timeout=10) as client:
        idx = await client.get_namespace_index("http://br-automation.com")
        print(f"B&R namespace index: {idx}")
        
        sub = await client.create_subscription(500, None)
        
        variables = [
            "ns=2;s=::AsGlobalPV:MachineSpeed",
            "ns=2;s=::MainProgram:CycleTime",
            "ns=2;s=::MainProgram:EmergencyStop",
        ]
        
        handlers = []
        for var_path in variables:
            node = client.get_node(var_path)
            name = await node.get_browse_name()
            
            def make_callback(n=name.Name):
                def datachange_notification(node, val, data):
                    print(f"[{asyncio.get_event_loop().time():.2f}] {n} = {val}")
                return datachange_notification
            
            handle = await sub.subscribe_data_change(
                var_path, make_callback()
            )
            handlers.append(handle)
        
        print("Monitoring... Press Ctrl+C to stop")
        
        try:
            while True:
                await asyncio.sleep(1)
        except KeyboardInterrupt:
            print("\nStopping...")
            await sub.delete()

asyncio.run(monitor_variables())

5.2 Subscription with Data Logging to CSV

import asyncio
import csv
from datetime import datetime
from asyncua import Client

PLC_URL = "opc.tcp://<PLC_IP>:4840"
CSV_FILE = "plc_trend_log.csv"

VARIABLES = {
    "MachineSpeed": "ns=2;s=::AsGlobalPV:MachineSpeed",
    "MotorTemp": "ns=2;s=::MainProgram:MotorTemp",
    "CycleTime": "ns=2;s=::MainProgram:CycleTime",
    "PartCount": "ns=2;s=::AsGlobalPV:PartCounter",
}

async def trend_logger():
    async with Client(url=PLC_URL, timeout=10) as client:
        sub = await client.create_subscription(200, None)
        
        with open(CSV_FILE, "w", newline="") as f:
            writer = csv.writer(f)
            writer.writerow(["timestamp", "variable", "value"])
            
            buffer = []
            flush_interval = 60  # seconds
            
            def make_callback(var_name):
                def callback(node, val, data):
                    ts = datetime.now().isoformat()
                    buffer.append([ts, var_name, str(val)])
                return callback
            
            handles = []
            for name, node_id in VARIABLES.items():
                handle = await sub.subscribe_data_change(
                    node_id, make_callback(name)
                )
                handles.append(handle)
            
            print(f"Logging {len(VARIABLES)} variables to {CSV_FILE}")
            
            try:
                while True:
                    await asyncio.sleep(flush_interval)
                    if buffer:
                        writer.writerows(buffer)
                        f.flush()
                        print(f"Flushed {len(buffer)} entries")
                        buffer.clear()
            except KeyboardInterrupt:
                if buffer:
                    writer.writerows(buffer)
                    f.flush()
                print(f"\nStopped. Total entries written to {CSV_FILE}")
                await sub.delete()

asyncio.run(trend_logger())

6. PVI: Python Access via pvipy

6.1 Installation

pip install pvipy

Pvipy requires the B&R PVI Development Setup to be installed on the Windows machine. The PVI Manager service must be running.

6.2 Connecting to a CP1584

from pvi import *

pviConnection = Connection()

line = Line(pviConnection.root, 'LNANSL', CD='LNANSL')
device = Device(line, 'TCP', CD='/IF=TcpIp')
cpu = Cpu(device, 'CP1584', CD='/IP=<PLC_IP>')

def cpuErrorChanged(error: int):
    if error == 11020:
        print("Unable to establish connection")
        pviConnection.stop()
    elif error != 0:
        raise PviError(error)
    else:
        print("Connected to CP1584 via PVI/ANSL")

cpu.errorChanged = cpuErrorChanged
pviConnection.start()

6.3 Reading Variables

variables = ["MainProgram.StartButton", "MainProgram.CycleCounter", "AsGlobalPV:MachineSpeed"]

for var_name in variables:
    var = cpu.create_variable(var_name, var_name)
    value = var.read()
    print(f"{var_name} = {value}")

6.4 Browsing Variables Without Project Files

PVI can discover variables on the controller even without a project file:

from pvi import *

pviConnection = Connection()
line = Line(pviConnection.root, 'LNANSL', CD='LNANSL')
device = Device(line, 'TCP', CD='/IF=TcpIp')
cpu = Cpu(device, 'CP1584', CD='/IP=<PLC_IP>')

def cpuErrorChanged(error: int):
    if error == 0:
        variables = cpu.list_variables()
        print(f"Found {len(variables)} variables:")
        for var in sorted(variables):
            print(f"  {var}")
        pviConnection.stop()

cpu.errorChanged = cpuErrorChanged
pviConnection.start()

6.5 Writing Variables

var = cpu.create_variable("AsGlobalPV:ManualOverride", "AsGlobalPV:ManualOverride")
var.write(True)
print("ManualOverride set to True")

6.6 SNMP Line — Network Configuration

PVI’s SNMP line can read and modify the PLC’s network settings:

from pvi import *

pviConnection = Connection()

line = Line(pviConnection.root, 'LNSNMP', CD='LNSNMP')
device = Device(line, 'SNMP', CD='/IP=<PLC_IP>')

cpu = Cpu(device, 'CP1584')
cpu.errorChanged = lambda error: print(f"SNMP connected, error={error}")
pviConnection.start()

7. Diagnostic Scripts: Practical Examples

7.1 All-Digital-IO Snapshot

import asyncio
from asyncua import Client
import json
from datetime import datetime

PLC_URL = "opc.tcp://<PLC_IP>:4840"

DIGITAL_INPUTS = [
    "ns=2;s=::MainProgram:DI[0]", "ns=2;s=::MainProgram:DI[1]",
    "ns=2;s=::MainProgram:DI[2]", "ns=2;s=::MainProgram:DI[3]",
    "ns=2;s=::MainProgram:DI[4]", "ns=2;s=::MainProgram:DI[5]",
    "ns=2;s=::MainProgram:DI[6]", "ns=2;s=::MainProgram:DI[7]",
]

DIGITAL_OUTPUTS = [
    "ns=2;s=::MainProgram:DO[0]", "ns=2;s=::MainProgram:DO[1]",
    "ns=2;s=::MainProgram:DO[2]", "ns=2;s=::MainProgram:DO[3]",
    "ns=2;s=::MainProgram:DO[4]", "ns=2;s=::MainProgram:DO[5]",
    "ns=2;s=::MainProgram:DO[6]", "ns=2;s=::MainProgram:DO[7]",
]

async def io_snapshot():
    async with Client(url=PLC_URL, timeout=5) as client:
        result = {
            "timestamp": datetime.now().isoformat(),
            "inputs": {},
            "outputs": {},
        }
        
        for i, node_id in enumerate(DIGITAL_INPUTS):
            try:
                val = await client.get_node(node_id).get_value()
                result["inputs"][f"DI[{i}]"] = bool(val)
            except Exception:
                result["inputs"][f"DI[{i}]"] = None
        
        for i, node_id in enumerate(DIGITAL_OUTPUTS):
            try:
                val = await client.get_node(node_id).get_value()
                result["outputs"][f"DO[{i}]"] = bool(val)
            except Exception:
                result["outputs"][f"DO[{i}]"] = None
        
        print(json.dumps(result, indent=2))
        return result

asyncio.run(io_snapshot())

7.2 Watchdog and Cycle Time Monitor

import asyncio
from asyncua import Client

PLC_URL = "opc.tcp://<PLC_IP>:4840"

async def monitor_cycle_times():
    async with Client(url=PLC_URL, timeout=10) as client:
        sub = await client.create_subscription(1000, None)
        
        cycle_vars = [
            ("Cyclic1_ms", "ns=2;s=::System:TaskClass1.CycleTime"),
            ("Cyclic2_ms", "ns=2;s=::System:TaskClass2.CycleTime"),
            ("Watchdog", "ns=2;s=::System:WatchdogStatus"),
        ]
        
        for name, node_id in cycle_vars:
            try:
                def cb(n=name):
                    def handler(node, val, data):
                        if "Watchdog" in n:
                            print(f"*** WATCHDOG: {val} ***")
                        else:
                            if val > 50:
                                print(f"*** WARNING: {n} = {val} ms (HIGH) ***")
                            else:
                                print(f"{n} = {val} ms")
                    return handler
                await sub.subscribe_data_change(node_id, cb())
            except Exception as e:
                print(f"Cannot subscribe to {name}: {e}")
        
        print("Monitoring cycle times... (Ctrl+C to stop)")
        try:
            while True:
                await asyncio.sleep(1)
        except KeyboardInterrupt:
            await sub.delete()

asyncio.run(monitor_cycle_times())

7.3 Automated IO Stress Test

import asyncio
import time
from asyncua import Client, ua

PLC_URL = "opc.tcp://<PLC_IP>:4840"

OUTPUT_NODES = [
    "ns=2;s=::MainProgram:DO[0]",
    "ns=2;s=::MainProgram:DO[1]",
    "ns=2;s=::MainProgram:DO[2]",
    "ns=2;s=::MainProgram:DO[3]",
]

async def stress_test():
    async with Client(url=PLC_URL, timeout=5) as client:
        print("IO Stress Test — toggling all outputs at 1 Hz for 60 seconds")
        
        for i, node_id in enumerate(OUTPUT_NODES):
            node = client.get_node(node_id)
            attr = ua.WriteValue()
            attr.NodeId = node.nodeid
            attr.AttributeId = ua.AttributeIds.Value
            attr.Value = ua.DataValue(True)
            params = ua.WriteParameters()
            params.NodesToWrite = [attr]
            await node.write_params(params)
        
        time.sleep(0.5)
        
        cycles = 0
        start = time.time()
        
        while time.time() - start < 60:
            state = cycles % 2 == 0
            for node_id in OUTPUT_NODES:
                node = client.get_node(node_id)
                attr = ua.WriteValue()
                attr.NodeId = node.nodeid
                attr.AttributeId = ua.AttributeIds.Value
                attr.Value = ua.DataValue(state)
                params = ua.WriteParameters()
                params.NodesToWrite = [attr]
                await node.write_params(params)
            cycles += 1
            await asyncio.sleep(0.5)
        
        for node_id in OUTPUT_NODES:
            node = client.get_node(node_id)
            attr = ua.WriteValue()
            attr.NodeId = node.nodeid
            attr.AttributeId = ua.AttributeIds.Value
            attr.Value = ua.DataValue(False)
            params = ua.WriteParameters()
            params.NodesToWrite = [attr]
            await node.write_params(params)
        
        print(f"Completed {cycles} toggle cycles in 60 seconds")

asyncio.run(stress_test())

8. Automated Regression and Diagnostic Framework

8.1 Expected vs Actual Comparison

import asyncio
from asyncua import Client
import json
from datetime import datetime

PLC_URL = "opc.tcp://<PLC_IP>:4840"

EXPECTED_STATES = {
    "ns=2;s=::MainProgram:EmergencyStop": False,
    "ns=2;s=::MainProgram:DoorInterlock": True,
    "ns=2;s=::MainProgram:MainSwitch": True,
    "ns=2;s=::MainProgram:LightCurtain": False,
    "ns=2;s=::MainProgram:OvertempAlarm": False,
}

async def regression_check():
    async with Client(url=PLC_URL, timeout=5) as client:
        report = {
            "timestamp": datetime.now().isoformat(),
            "total": len(EXPECTED_STATES),
            "passed": 0,
            "failed": 0,
            "errors": 0,
            "details": [],
        }
        
        for node_id, expected in EXPECTED_STATES.items():
            try:
                actual = await client.get_node(node_id).get_value()
                passed = (actual == expected)
                
                if passed:
                    report["passed"] += 1
                else:
                    report["failed"] += 1
                
                report["details"].append({
                    "node": node_id,
                    "expected": expected,
                    "actual": actual,
                    "passed": passed,
                })
                
                status = "PASS" if passed else "FAIL"
                print(f"  [{status}] {node_id}: expected={expected}, actual={actual}")
                
            except Exception as e:
                report["errors"] += 1
                report["details"].append({
                    "node": node_id,
                    "expected": expected,
                    "actual": f"ERROR: {e}",
                    "passed": False,
                })
                print(f"  [ERROR] {node_id}: {e}")
        
        print(f"\nResults: {report['passed']}/{report['total']} passed, "
              f"{report['failed']} failed, {report['errors']} errors")
        return report

asyncio.run(regression_check())

8.2 Scheduled Diagnostic Runner

import asyncio
import schedule
import time
import json
from datetime import datetime
from asyncua import Client

PLC_URL = "opc.tcp://<PLC_IP>:4840"
LOG_FILE = "regression_log.jsonl"

async def run_check():
    async with Client(url=PLC_URL, timeout=10) as client:
        result = {
            "timestamp": datetime.now().isoformat(),
            "status": "unknown",
            "errors": [],
        }
        
        try:
            estop = await client.get_node("ns=2;s=::MainProgram:EmergencyStop").get_value()
            result["estop"] = estop
            
            try:
                speed = await client.get_node("ns=2;s=::AsGlobalPV:MachineSpeed").get_value()
                result["speed"] = speed
            except Exception:
                pass
            
            result["status"] = "ok" if not estop else "estop_active"
            
        except Exception as e:
            result["status"] = "connection_error"
            result["errors"].append(str(e))
        
        with open(LOG_FILE, "a") as f:
            f.write(json.dumps(result) + "\n")
        
        print(f"[{result['timestamp']}] Status: {result['status']}")
        return result

def sync_wrapper():
    asyncio.run(run_check())

schedule.every(5).minutes.do(sync_wrapper)
schedule.every().hour.do(lambda: print("--- Hourly summary ---"))

if __name__ == "__main__":
    print("Starting scheduled diagnostics...")
    asyncio.run(run_check())  # Immediate first run
    while True:
        schedule.run_pending()
        time.sleep(1)

8.3 Diagnostic Report Generator

import asyncio
import json
from datetime import datetime
from asyncua import Client

PLC_URL = "opc.tcp://<PLC_IP>:4840"
REPORT_FILE = "diagnostic_report.html"

async def generate_report():
    report_data = {
        "timestamp": datetime.now().isoformat(),
        "connection": "unknown",
        "variables": [],
        "errors": [],
    }
    
    try:
        async with Client(url=PLC_URL, timeout=10) as client:
            report_data["connection"] = "ok"
            objects = client.nodes.objects
            children = await objects.get_children()
            report_data["root_children"] = len(children)
            
            sample_vars = [
                ("MachineSpeed", "ns=2;s=::AsGlobalPV:MachineSpeed"),
                ("PartCounter", "ns=2;s=::AsGlobalPV:PartCounter"),
                ("CycleTime", "ns=2;s=::MainProgram:CycleTime"),
            ]
            
            for name, node_id in sample_vars:
                try:
                    val = await client.get_node(node_id).get_value()
                    report_data["variables"].append({
                        "name": name,
                        "value": str(val),
                        "type": type(val).__name__,
                    })
                except Exception as e:
                    report_data["errors"].append(f"{name}: {e}")
                    
    except Exception as e:
        report_data["connection"] = f"failed: {e}"
    
    html = f"""<!DOCTYPE html>
<html><head><title>CP1584 Diagnostic Report</title></head>
<body>
<h1>CP1584 Diagnostic Report</h1>
<p>Generated: {report_data['timestamp']}</p>
<p>Connection: {report_data['connection']}</p>
<h2>Variable Status</h2>
<table border="1">
<tr><th>Variable</th><th>Value</th><th>Type</th></tr>"""
    
    for v in report_data["variables"]:
        html += f"""<tr><td>{v['name']}</td><td>{v['value']}</td><td>{v['type']}</td></tr>"""
    
    html += "</table>"
    
    if report_data["errors"]:
        html += "<h2>Errors</h2><ul>"
        for e in report_data["errors"]:
            html += f"<li>{e}</li>"
        html += "</ul>"
    
    html += "</body></html>"
    
    with open(REPORT_FILE, "w") as f:
        f.write(html)
    
    print(f"Report saved to {REPORT_FILE}")

asyncio.run(generate_report())

9. Integrating with IIoT and Monitoring Systems

9.1 OPC-UA to MQTT Bridge

For forwarding PLC data to MQTT-based monitoring (Grafana, Node-RED, Home Assistant):

import asyncio
from asyncua import Client
import paho.mqtt.client as mqtt

PLC_URL = "opc.tcp://<PLC_IP>:4840"
MQTT_BROKER = "localhost"
MQTT_PORT = 1883
MQTT_TOPIC = "br/plc/"

variables_map = {
    "ns=2;s=::AsGlobalPV:MachineSpeed": "machine/speed",
    "ns=2;s=::MainProgram:CycleTime": "machine/cycle_time",
    "ns=2;s=::MainProgram:EmergencyStop": "machine/estop",
}

mqtt_client = mqtt.Client()
mqtt_client.connect(MQTT_BROKER, MQTT_PORT)

async def opcua_to_mqtt():
    async with Client(url=PLC_URL, timeout=10) as client:
        sub = await client.create_subscription(500, None)
        
        for node_id, mqtt_path in variables_map.items():
            def cb(topic=f"{MQTT_TOPIC}{mqtt_path}"):
                def handler(node, val, data):
                    mqtt_client.publish(topic, str(val))
                return handler
            await sub.subscribe_data_change(node_id, cb())
        
        print("OPC-UA to MQTT bridge running...")
        try:
            while True:
                await asyncio.sleep(1)
                mqtt_client.loop(0.01)
        except KeyboardInterrupt:
            await sub.delete()

asyncio.run(opcua_to_mqtt())

9.2 Prometheus Exporter

For integration with Grafana/Prometheus:

import asyncio
from aiohttp import web
from asyncua import Client

PLC_URL = "opc.tcp://<PLC_IP>:4840"
PORT = 8000

latest_values = {}

VARIABLES = {
    "br_machine_speed": "ns=2;s=::AsGlobalPV:MachineSpeed",
    "br_cycle_time_ms": "ns=2;s=::MainProgram:CycleTime",
    "br_part_count": "ns=2;s=::AsGlobalPV:PartCounter",
}

async def update_values():
    while True:
        try:
            async with Client(url=PLC_URL, timeout=5) as client:
                for metric, node_id in VARIABLES.items():
                    try:
                        val = await client.get_node(node_id).get_value()
                        latest_values[metric] = val
                    except Exception:
                        pass
        except Exception:
            pass
        await asyncio.sleep(2)

async def metrics_handler(request):
    output = ""
    for metric, value in latest_values.items():
        if isinstance(value, (int, float)):
            output += f"{metric} {value}\n"
    return web.Response(text=output, content_type="text/plain")

async def main():
    app = web.Application()
    app.router.add_get("/metrics", metrics_handler)
    
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, "0.0.0.0", PORT)
    await site.start()
    
    asyncio.create_task(update_values())
    print(f"Prometheus exporter on :{PORT}/metrics")
    
    while True:
        await asyncio.sleep(3600)

asyncio.run(main())

See also: iiot-retrofit.md

10. Common Pitfalls and Gotchas

10.1 B&R OPC-UA Write Issue

The standard asyncua node.write_value() method often returns BadWriteNotSupported on B&R OPC-UA servers, even when the variable is writable in UAExpert. Always use the write_params() method documented in Section 4.3.

10.2 Namespace Discovery

B&R OPC-UA namespaces are not always predictable. The namespace index for application variables depends on how the OPC-UA configuration was set up in Automation Studio. Always browse the namespace first rather than hardcoding ns=2 or ns=6.

10.3 PVI Trial License Limitation

Without a PVI license, PVI Manager runs for 2 hours before stopping all PVI-based programs. This is disruptive for long-running monitoring scripts. If you need continuous monitoring, use the OPC-UA path instead.

10.4 Variable Not Found

If a variable is not accessible via OPC-UA:

• The variable may not be mapped to the OPC-UA namespace (the OEM must explicitly export it)
• Try PVI, which has access to all variables in the PLC’s variable table
• Check if the variable is in a different namespace or has been renamed

10.5 Connection Timeout

B&R OPC-UA server startup can take several seconds after PLC boot. Add retry logic:

import asyncio
from asyncua import Client

async def connect_with_retry(url, max_retries=10, delay=3):
    for attempt in range(max_retries):
        try:
            client = Client(url=url, timeout=5)
            await client.connect()
            print(f"Connected on attempt {attempt + 1}")
            return client
        except Exception as e:
            print(f"Attempt {attempt + 1} failed: {e}")
            if attempt < max_retries - 1:
                await asyncio.sleep(delay)
    raise ConnectionError(f"Failed to connect after {max_retries} retries")

async def main():
    client = await connect_with_retry("opc.tcp://<PLC_IP>:4840")
    try:
        # ... work with client ...
        pass
    finally:
        await client.disconnect()

asyncio.run(main())

Community Tools for Python-Based Diagnostics

Beyond the core asyncua and pvipy libraries, several community tools extend Python-based diagnostics for B&R systems:

brwatch (Windows Service + Command Line)

brwatch is a Windows service tool for variable watch, change, logging, IP configuration, and CPU reboot — without requiring Automation Studio or the project file. While not Python-native, it integrates with Python workflows:

• CSV logging: brwatch can log variable values to CSV files that Python scripts then ingest for analysis
• Variable discovery: Use brwatch to enumerate all visible variables, then build Python monitoring scripts targeting specific variables
• IP configuration: Automate initial PLC network setup from the command line before any Python connectivity

import pandas as pd
import glob

latest_csv = sorted(glob.glob("brwatch_log_*.csv"))[-1]
df = pd.read_csv(latest_csv)
print(f"Loaded {len(df)} samples from {latest_csv}")
print(df.describe())

See access-recovery.md §12 for the complete brwatch guide.

Pvi.py (Python PVI Wrapper)

Pvi.py is a Python wrapper for B&R PVI that provides a cleaner API than pvipy for common operations:

from pvi import Pvi

pvi = Pvi()
plc = pvi.line("ANSL", "TCP", "<PLC_IP>").device("CP1584")

var = plc.variable("gMachineData.nTemperature")
print(f"Current temperature: {var.read()}")

var.write(25.0)
plc.restart()

Comparison: Pvi.py vs pvipy:

systemdump.py (CLI Dump Analyzer)

systemdump.py parses B&R SystemDump.xml files from the command line:

pip install systemdump.py
systemdump --input SystemDump.xml --summary
systemdump --input SystemDump.xml --logger --filter ERROR
systemdump --input SystemDump.xml --tasks --format json > task_stats.json

Combined with Python:

import subprocess, json

result = subprocess.run(
    ["systemdump", "--input", "SystemDump.xml", "--tasks", "--format", "json"],
    capture_output=True, text=True
)
task_data = json.loads(result.stdout)
for task in task_data.get("tasks", []):
    print(f"{task['name']}: cycle={task['cycle_ms']}ms, max={task['max_cycle_ms']}ms")

See ebpf-telemetry.md for post-mortem analysis workflows and diagnostics-sdm.md for system dump download procedures.

awesome-B-R Curated Tool List

The awesome-B-R repository maintains a comprehensive catalog of all community tools, libraries, and resources for B&R automation. Key entries relevant to Python diagnostics:

Key Findings

1. OPC-UA via asyncua is the recommended path for Python diagnostics on B&R CP1584 — no Windows middleware, no PVI license required, works from any OS.
2. B&R OPC-UA requires write_params() instead of write_value() for writing variables — a documented gotcha in the community.
3. The pvipy library provides clean Python access via PVI but requires Windows + PVI Development Setup + license.
4. Variable namespace must be discovered — B&R namespace indices are project-specific, not standardized.
5. Subscriptions enable real-time monitoring with CSV logging, Prometheus export, or MQTT bridging for IIoT integration.
6. Regression testing (expected vs actual) is practical for documenting and verifying machine behavior on undocumented systems.
7. Community tools extend Python diagnostics significantly — brwatch for no-project variable access and CSV logging, Pvi.py for a Pythonic PVI API, systemdump.py for automated dump analysis, and brsnmp for PLC discovery without Automation Studio. See awesome-B-R for the full catalog.

Sources

• B&R PVI (Process Visualization Interface) API documentation — C/C++ and .NET SDK references
• B&R OPC-UA Server Configuration Guide — namespace and endpoint configuration
• Python OPC-UA library (asyncua) documentation — https://github.com/FreeOpcUa/opcua-asyncio
• Python Modbus library (pymodbus) documentation — https://pymodbus.readthedocs.io
• Python PVI library (pvi.py) by hilch — https://github.com/hilch/Pvi.py
• B&R Community Forum (community.br-automation.com) — PVI and OPC-UA scripting discussions
• awesome-B-R GitHub repository — community tool listing for B&R automation

Related Documents

• pvi-api.md — PVI API reference for programmatic PLC access
• opcua.md — OPC-UA server configuration and namespace structure
• modbus-gateway.md — Modbus gateway configuration as alternative data source
• cp1584-forensics.md — Initial information extraction methodology
• iiot-retrofit.md — MQTT, SNMP, and dashboard integration from Python data
• diagnostics-sdm.md — SDM web interface for baseline diagnostic data
• custom-diagnostic-tools.md — Building diagnostic tools that run ON the PLC
• ebpf-telemetry.md — Alternative telemetry and monitoring approaches