OpenPLC with Python FB

[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

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')

# 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

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

(* 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);

angle := angle + 0.05;

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

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

VAR_INPUT
	temp_in_1 : real := 0.0;
END_VAR

# ================================================================
# 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')

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

VAR_INPUT
	temp_in_2 : real := 0.0;
END_VAR

VAR_OUTPUT
	status: int;
END_VAR

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

# ============================================================
# 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

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

(* 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;

# 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:

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