How to Integrate PLC with Python in a Production Line

A practical, code-first guide for industrial automation (OPC UA, data pipelines, and MES integration)

🔧 Introduction

Programmable Logic Controllers (PLCs) excel at deterministic, real-time control. They are reliable, predictable, and designed for harsh industrial environments.

But modern production lines demand more:

• real-time analytics
• cloud integration
• predictive maintenance
• MES / ERP connectivity

That’s where Python becomes extremely valuable.

Instead of replacing PLCs, Python acts as a bridge between OT (Operational Technology) and IT systems.

🧠 Core Architecture

The most common integration pattern looks like this:

7

Flow:

PLC → OPC UA → Python → Database / Cloud / MES

⚙️ Why Python?

Python is widely used in industrial environments because:

• fast prototyping
• huge ecosystem (AI, data, cloud)
• easy OPC UA integration
• strong community support

Typical libraries:

• opcua (PLC communication)
• pandas (data processing)
• numpy (analytics)
• sqlalchemy (database)
• requests (API integration)

🏭 Real Use Cases

1. Production Monitoring

• machine states (RUN / STOP / ERROR)
• cycle time tracking
• downtime analysis

2. OEE Calculation

• availability
• performance
• quality

3. Predictive Maintenance

• vibration data
• temperature trends
• anomaly detection

4. MES / ERP Integration

• production orders
• material tracking
• reporting

🔌 Step 1: Connecting to PLC via OPC UA

Most modern PLCs (e.g., Siemens, Beckhoff, Allen-Bradley) expose OPC UA servers.

Install dependencies

pip install opcua

pip install opcua

Python OPC UA Client Example

from opcua import Client

url = "opc.tcp://192.168.0.10:4840"
client = Client(url)

try:
    client.connect()

    temperature_node = client.get_node("ns=2;i=2")
    value = temperature_node.get_value()

    print(f"Temperature: {value}")

finally:
    client.disconnect()

from opcua import Client

url = "opc.tcp://192.168.0.10:4840"
client = Client(url)

try:
    client.connect()

    temperature_node = client.get_node("ns=2;i=2")
    value = temperature_node.get_value()

    print(f"Temperature: {value}")

finally:
    client.disconnect()

📡 Step 2: Subscribing to Real-Time Data

Polling is inefficient. Instead, use subscriptions.

from opcua import Client, ua

class SubscriptionHandler:
    def datachange_notification(self, node, val, data):
        print(f"Data changed: {node} -> {val}")

client = Client("opc.tcp://192.168.0.10:4840")
client.connect()

handler = SubscriptionHandler()
subscription = client.create_subscription(500, handler)

node = client.get_node("ns=2;i=2")
subscription.subscribe_data_change(node)

from opcua import Client, ua

class SubscriptionHandler:
    def datachange_notification(self, node, val, data):
        print(f"Data changed: {node} -> {val}")

client = Client("opc.tcp://192.168.0.10:4840")
client.connect()

handler = SubscriptionHandler()
subscription = client.create_subscription(500, handler)

node = client.get_node("ns=2;i=2")
subscription.subscribe_data_change(node)

👉 This enables event-driven processing.

🗄️ Step 3: Saving Data to Database

Example using SQLite:

import sqlite3
from datetime import datetime

conn = sqlite3.connect("production.db")
cursor = conn.cursor()

cursor.execute("""
CREATE TABLE IF NOT EXISTS telemetry (
    timestamp TEXT,
    value REAL
)
""")

def save_value(value):
    cursor.execute(
        "INSERT INTO telemetry VALUES (?, ?)",
        (datetime.utcnow().isoformat(), value)
    )
    conn.commit()

import sqlite3
from datetime import datetime

conn = sqlite3.connect("production.db")
cursor = conn.cursor()

cursor.execute("""
CREATE TABLE IF NOT EXISTS telemetry (
    timestamp TEXT,
    value REAL
)
""")

def save_value(value):
    cursor.execute(
        "INSERT INTO telemetry VALUES (?, ?)",
        (datetime.utcnow().isoformat(), value)
    )
    conn.commit()

☁️ Step 4: Sending Data to Cloud API

import requests

def send_to_cloud(value):
    payload = {
        "machine": "line_1",
        "temperature": value
    }

    requests.post(
        "https://api.example.com/telemetry",
        json=payload
    )

import requests

def send_to_cloud(value):
    payload = {
        "machine": "line_1",
        "temperature": value
    }

    requests.post(
        "https://api.example.com/telemetry",
        json=payload
    )

🔄 Step 5: Full Integration Example

Putting everything together:

from opcua import Client
import sqlite3
import requests
from datetime import datetime

class Handler:
    def datachange_notification(self, node, val, data):
        print(f"New value: {val}")
        save_value(val)
        send_to_cloud(val)

def save_value(value):
    conn = sqlite3.connect("production.db")
    cursor = conn.cursor()

    cursor.execute("""
    CREATE TABLE IF NOT EXISTS telemetry (
        timestamp TEXT,
        value REAL
    )
    """)

    cursor.execute(
        "INSERT INTO telemetry VALUES (?, ?)",
        (datetime.utcnow().isoformat(), value)
    )

    conn.commit()
    conn.close()

def send_to_cloud(value):
    requests.post(
        "https://api.example.com/data",
        json={"value": value}
    )

client = Client("opc.tcp://192.168.0.10:4840")
client.connect()

handler = Handler()
sub = client.create_subscription(1000, handler)

node = client.get_node("ns=2;i=2")
sub.subscribe_data_change(node)

try:
    while True:
        pass
except KeyboardInterrupt:
    client.disconnect()

from opcua import Client
import sqlite3
import requests
from datetime import datetime

class Handler:
    def datachange_notification(self, node, val, data):
        print(f"New value: {val}")
        save_value(val)
        send_to_cloud(val)

def save_value(value):
    conn = sqlite3.connect("production.db")
    cursor = conn.cursor()

    cursor.execute("""
    CREATE TABLE IF NOT EXISTS telemetry (
        timestamp TEXT,
        value REAL
    )
    """)

    cursor.execute(
        "INSERT INTO telemetry VALUES (?, ?)",
        (datetime.utcnow().isoformat(), value)
    )

    conn.commit()
    conn.close()

def send_to_cloud(value):
    requests.post(
        "https://api.example.com/data",
        json={"value": value}
    )

client = Client("opc.tcp://192.168.0.10:4840")
client.connect()

handler = Handler()
sub = client.create_subscription(1000, handler)

node = client.get_node("ns=2;i=2")
sub.subscribe_data_change(node)

try:
    while True:
        pass
except KeyboardInterrupt:
    client.disconnect()

🧩 Industrial Data Flow Explained

6

Key layers:

1. PLC Layer
   Real-time signals
2. OPC UA Layer
   Standardized communication
3. Python Layer
   • processing
   • enrichment
   • routing
4. IT Systems
   • database
   • MES
   • cloud

⚠️ Common Pitfalls

❌ Polling instead of subscriptions

👉 leads to performance issues

❌ No buffering

👉 data loss during network failures

❌ Blocking code

👉 delays real-time processing

❌ Tight coupling with PLC structure

👉 hard to maintain

🛠️ Production-Ready Improvements

For real-world deployments:

• use message brokers (Kafka / RabbitMQ)
• implement retry policies
• add data buffering (Redis / local queue)
• run as Docker container
• add monitoring (Prometheus / Grafana)

🔐 Security Considerations

• enable OPC UA security (certificates)
• avoid exposing PLC directly to internet
• use VPN or gateway
• validate incoming data

🧪 Testing Strategy

• simulate PLC using OPC UA simulator
• test network failures
• load test subscriptions
• validate data integrity

🚀 When to Use This Pattern

This architecture works best when:

• PLC logic should remain unchanged
• analytics is required
• integration with IT systems is needed
• scalability is important

🧾 Conclusion

Python does not replace PLCs.

👉 It extends their capabilities into the IT world.

References

https://akpolatcem.medium.com/getting-started-with-opc-ua-building-your-first-client-server-application-in-python-6107fb4cafec

More Info

We implement this in real production systems.
If you need help → contact us

Contact