> For the complete documentation index, see [llms.txt](https://wiki.codeandcompile.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://wiki.codeandcompile.com/product-reviews/smart-platforms/virtual-plcs/autonomy-edge-openplc-redefined/openplc-with-python-fb.md). # OpenPLC with Python FB ### OpenPLC with Python FB- Sending PLC Data to the Cloud {% embed url="" %} *** ### 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 {% embed url="" %} *** ### 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 | {% hint style="warning" %} 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. {% endhint %} *** ### 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` ```python 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') ``` {% hint style="info" %} `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. {% endhint %} *** ### 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. ```python # 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 ``` {% hint style="info" %} 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. {% endhint %} *** ### 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** ```rst VAR simulated_temp: real; angle: real; my_inst_1: example_1; END_VAR ``` #### **Main ST Program** ```pascal (* 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 ```pascal 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** ```python VAR_INPUT temp_in_1 : real := 0.0; END_VAR ``` #### **Python Code** ```python # ================================================================ # 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** ```python 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](https://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** ```python # ============================================================ # 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** ```rst VAR simulated_temp : real; angle : real; my_inst_1 : example_1; my_inst_2 : example_2; sent_status : int; END_VAR ``` **Main ST Program** ```pascal (* 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. ```python # 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()`: ```python 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 {% embed url="" %} ### 🔗 Resources * 🌐 Open PLC editor and Runtime: [autonomylogic.com](https://autonomylogic.com) * 📖 [Python FB Documentation](https://edge.autonomylogic.com/docs/openplc-editor/custom-languages/python-blocks/python-basics) * 📊 [ThingsBoard Cloud](https://thingsboard.cloud) * 🧑‍💻 [Code Compile](https://codeandcompile.com) *** ## ♥️ 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: or drop me a message on [LinkedIn](https://www.linkedin.com/in/singhrajvir/)