# 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/) --- # Agent Instructions: Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter: ``` GET https://wiki.codeandcompile.com/product-reviews/smart-platforms/virtual-plcs/autonomy-edge-openplc-redefined/openplc-with-python-fb.md?ask= ``` The question should be specific, self-contained, and written in natural language. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.