OpenPLC with Python FB

In this article, you will learn how to use Python FB in OpenPLC Editor and send the PLC data to the ThingsBOard's Cloud Dashboard.

OpenPLC with Python FB- Sending PLC Data to the Cloud

---

What This Project Does

In this project, we use OpenPLC Runtime v4 running on a laptop (or edge device) to:

• Generate a simulated sine wave temperature value using the built-in OpenPLC simulator

• Read it inside a Python Function Block via shared memory

• Send live telemetry to ThingsBoard Cloud via HTTP — using only Python's standard library

• Write a connection status back to the PLC so the ST program knows if the cloud is reachable

No hardware required. No third-party packages. No cost.

Watch Full Walkthrough with LIVE Demo

---

System Architecture

[OpenPLC Simulator]
        |
        |  Structured Text - sine wave generator
        |
[OpenPLC Runtime v4]
        |
        |  Python Function Block - direct variable access
        |
[Python Function Block - example_2.py]
        |
        |  HTTP POST - urllib (standard library only)
        |
[ThingsBoard Cloud - thingsboard.cloud]
        |
        -- Live temperature graph on dashboard

---

What is a Python Function Block?

Python Function Blocks let you write automation logic in Python while integrating seamlessly with your IEC 61131-3 program. They are ideal for tasks that are difficult or verbose in Structured Text:

• HTTP / REST API calls

• JSON formatting and parsing

• Statistical calculations

• Cloud and IoT communication

Aspect	Standard IEC Function Block	Python Function Block
Execution	Runs inside the PLC scan cycle	Runs as a separate process
Timing	Synchronized with scan cycle	Asynchronous ~100ms loop
State	Managed by the runtime	Managed by the Python process
Language	ST, LD, FBD, IL	Python 3
Libraries	IEC standard functions only	Python standard library

Python FBs are NOT synchronized with the PLC scan cycle. Use standard IEC languages for real-time, time-critical control logic. Use Python for non-time-critical tasks like cloud communication.

---

The Two Required Functions

Every Python Function Block defines exactly two functions called automatically by the runtime:

block_init() - Called once when the Python process starts. Use it for:

• Initializing global variables

• Setting up timers and data structures

• Printing startup messages

block_loop() - Called every ~100ms for the lifetime of the process. Use it for:

• Reading inputs from shm_in

• Processing data

• Writing outputs to shm_out

from multiprocessing import shared_memory
import struct
import time
import os

def block_init():
    print('Block was initialized')

def block_loop():
    print('Block has run the loop function')

shm_in and shm_out are injected automatically by the runtime — you never declare them yourself. Your IDE may show a warning — this is safe to ignore.

---

How Variables Work in Python FBs

Variables declared in the Variables Table in the OpenPLC Editor are accessible directly by name inside your Python code.

# If you declare temp_in as an Input variable in the table:
# You can read it directly in Python like this:
print(temp_in)           # reads the current PLC value

# If you declare status as an Output variable in the table:
# You can write to it directly like this:
global status           # it is mandatory to declare the variable to global
status = 1               # writes value back to PLC

Variables declared as Input are written by the PLC and read by Python. Variables declared as Output are written by Python and read by the PLC.

---

Example 1: Console Print (shm_in Only)

This example introduces shm_in and struct.unpack, reading a value from the PLC into Python and printing it to the PLC Logs.

Main ST Program: Example 1

Variable Table

VAR
	simulated_temp: real;
	angle: real;
	my_inst_1: example_1;
END_VAR

Main ST Program

(* Sine wave angle in radians - increments each scan cycle *)
angle := angle + 0.05;

(* Reset after full cycle - 6.283 = 2 x PI = 360 degrees *)
IF angle > 6.283 THEN
    angle := 0.0;
END_IF;

(* Generate simulated temperature - cycles between 20C and 80C *)
simulated_temp := 50.0 + (30.0 * SIN(angle));

(* Send temperature to Python FB via shared memory *)
my_inst_1(temp_in := simulated_temp);

Understanding the Sine Wave Generator

angle := angle + 0.05;

IF angle > 6.283 
    THEN angle := 0.0; 
END_IF;

simulated_temp := 50.0 + (30.0 * SIN(angle));

The SIN() function uses radians. One full circle = 2π = 6.283 radians. Resetting angle After 6.283, the wave repeats cleanly.

Each scan adds 0.05 radians — completing a full cycle in approximately 125 scans (~2.5 seconds at 20ms scan time).

The formula 50.0 + (30.0 * SIN(angle)) produces:

SIN value	Temperature
+1 (peak)	80°C
0 (midpoint)	50°C
-1 (trough)	20°C

Python FB Code: Example_1

Variable Table

VAR_INPUT
	temp_in_1 : real := 0.0;
END_VAR

Python Code

# ================================================================
# DISCLAIMER: Python Function Block Execution
#
# This block runs asynchronously from the main PLC runtime.
# ---------------------------------------------------------------
# - All variables are shared with the runtime through shared memory.
# - The block_init() function is called once when the block starts.
# - The block_loop() function is called periodically (~100ms).
# - IMPORTANT: This periodic call DOES NOT follow the PLC scan cycle.
#   It is NOT guaranteed that block_loop() will execute once per scan.
#
# Use this block for non-time-critical tasks. For logic that must
# match the PLC scan cycle, use standard IEC 61131-3 function blocks.
# ================================================================

from multiprocessing import shared_memory
import struct
import time
import os

def block_init():
    global last_print_time   # persist across block_loop() calls
    # Record the current time as the starting reference point
    last_print_time = time.time()
    print('Python FB started!')

def block_loop():
    global last_print_time   # must redeclare global in every function that uses it
    now = time.time()

    # Only print every 2 seconds - avoids flooding the PLC Logs
    if now - last_print_time >= 2.0:
        last_print_time = now  # reset the timer
        #print(f'Temperature: {round(temp_in_1, 2)} C')

PLC Logs Output

Python FB started!
Temperature: 62.35 C
Temperature: 71.18 C
Temperature: 78.43 C
Temperature: 79.98 C

---

Example 2: ThingsBoard Cloud Integration

This example extends Example 1 with two additions:

• Sends the temperature to ThingsBoard Cloud via HTTP every 5 seconds

• Write a connection status back to the PLC via shm_out - 1 for connected, 0 for disconnected

Variable Table

VAR_INPUT
	temp_in_2 : real := 0.0;
END_VAR

VAR_OUTPUT
	status: int;
END_VAR

ThingsBoard Cloud Setup

1. Sign up for free at thingsboard.cloud

2. Create a new device e.g. OpenPLC

3. Copy the url with the Access Token from the device credentials

4. Replace YOUR_TOKEN in the code below

The ThingsBoard HTTP telemetry API endpoint is:

'https://thingsboard.cloud/api/v1/YOUR_TOKEN/telemetry'

Python FB Code

# ============================================================
# Example 2: OpenPLC Python FB - ThingsBoard Cloud Telemetry
# Reads simulated temperature from PLC via shared memory
# Sends to ThingsBoard Cloud via HTTP every 5 seconds
# Writes connection status back to PLC via shared memory
# ============================================================

from multiprocessing import shared_memory
import struct
import time
import os

import urllib.request
import json

# ThingsBoard Cloud endpoint curl -v -X POST http://thingsboard.cloud/api/v1/K4GHZZMJY2zsnvesTDm8/telemetry --header Content-Type:application/json --data "{temperature:25}"
URL = 'https://thingsboard.cloud/api/v1/3JOfPb6SZUURbfIkVVId/telemetry'

#  HTTP Send Function 
def send_to_thingsboard(payload):
    try:
        req = urllib.request.Request(
            URL,
            data=payload.encode('utf-8'),
            headers={'Content-Type': 'application/json'}
        )
        urllib.request.urlopen(req, timeout=3)
        print('Sent: ' + payload)
        return True
    except Exception as e:
        print('Error: ' + str(e))
        return False

#  Called ONCE on startup ============================================================
def block_init():
    global last_send_time, status

    # Wait 5 seconds before first send (let runtime stabilise)
    last_send_time = time.time() + 5
    status = 0
    print('Cloud FB started!')

#  Called EVERY ~100ms ============================================================
def block_loop():
    global last_send_time, status

    # Only send every 5 seconds
    if time.time() - last_send_time < 5.0:
        return
    last_send_time = time.time()

    payload = json.dumps({'temperature': round(float(temp_in_2), 2)})
    success = send_to_thingsboard(payload)

    # WRITE: status 
    status = 1 if success else 0

Main ST Program

Variable Table

VAR
	simulated_temp : real;
	angle : real;
	my_inst_1 : example_1;
	my_inst_2 : example_2;
	sent_status : int;
END_VAR

Main ST Program

(* Sine wave angle in radians - increments each scan cycle *)
angle := angle + 0.05;

(* Reset after full cycle - 6.283 = 2 x PI = 360 degrees *)
IF angle > 6.283 THEN
    angle := 0.0;
END_IF;

(* Generate simulated temperature - cycles between 20C and 80C *)
simulated_temp := 50.0 + (30.0 * SIN(angle));

my_inst_2(temp_in_2 := simulated_temp);
sent_status := my_inst_2.status;

---

Key Gotchas: Lessons Learned

Building this project revealed several non-obvious behaviors worth knowing before you start:

1. Use time.time() for intervals — not loop counters

Using loop_counter % 50 == 0 seems logical, but fires immediately on loop 0, before the runtime has stabilized. An HTTP call at that moment can stall the process. Use time.time() instead — it waits a true interval from a stable starting point.

# WRONG - triggers on loop 0 before runtime is stable
if loop_counter % 50 == 0:

# CORRECT - waits true 5 seconds from stable init
if time.time() - last_send_time >= 5.0:

2. Always import time

The OpenPLC runtime uses the time module internally to manage block_loop() timing. Even if your code does not use time directly — always include it, or you will get a NameError that crashes the process.

3. Declare ALL persistent variables as global in BOTH functions

Without global in block_loop() Python creates a silent local copy of the variable that is destroyed at the end of each call. Always declare in both block_init() and block_loop():

def block_init():
    global last_send_time, status   # declare and initialise here
    last_send_time = time.time() + 5
    status = 0

def block_loop():
    global last_send_time, status   # redeclare in every function that uses it
    status = 1 if success else 0

4. Avoid special characters in comments

Em dashes —, curly quotes ', and other Unicode characters cause C compiler errors when OpenPLC embeds your Python code. Use plain ASCII only — hyphens - and straight quotes '.

5. ThingsBoard Cloud requires HTTPS

Local ThingsBoard uses http:// on port 8080. ThingsBoard Cloud requires https:// on port 443.

---

When to Use Python vs ST vs C++

Task	Language
Cloud communication, HTTP, REST APIs	Python FB
JSON formatting, string processing	Python FB
Real-time control, timing, interlocks	Structured Text
Scan-cycle-critical logic	Structured Text
Direct hardware access, byte manipulation	C++ FB

---

Hardware and Software Used

Component	Details
PLC Runtime	OpenPLC Runtime v4
Editor	Autonomy Edge IDE (edge.autonomylogic.com)
Hardware	Any laptop or edge device
Cloud Dashboard	ThingsBoard Cloud (free tier)
Protocol	HTTP REST API via Python urllib
Languages	IEC 61131-3 Structured Text + Python 3

---

Keynotes

🔗 Resources

• 🌐 Open PLC editor and Runtime: autonomylogic.com

• 📖 Python FB Documentation

• 📊 ThingsBoard Cloud

• 🧑‍💻 Code Compile

---

♥️ Work With Me

I regularly test industrial automation and IIoT devices. If you’d like me to review your product or showcase it in my courses and YouTube channel:

📧 Email: [e-mail] or drop me a message on LinkedIn