← Files RungstoneARCHIVED FILE
skills/rungstone-plc/scripts/st_to_61499.py
12 KB · Nov 3, 2026 · 06:12 UTC
"""
st_to_61499.py - wrap a scan-style IEC 61131-3 ST block as IEC 61499 function blocks for Eclipse 4diac (3.x).
Input: one PROGRAM or FUNCTION_BLOCK in ST, written for this target (no calls to timer/edge/counter or
other FB instances inside the body: timers are written with a CYCLE time input; see
references/iec61499.md). Output, in the 4diac 3.x .fbt format:
<Name>.fbt Basic FB. Events INIT (EInit) and REQ in, INITO and CNF out, each carrying all data
inputs/outputs (With). ECC: START -INIT-> INIT (initialize, INITO) -1-> START;
START -REQ-> REQ (REQ algorithm = the scan body, CNF) -1-> START.
VAR_INPUT and AT %I.. variables become data inputs; VAR_OUTPUT and AT %Q.. become data
outputs; other variables (incl. AT %M..) become internal variables. A CYCLE : TIME
input carries the scan period for timer arithmetic.
<Name>_App.fbt Composite FB: an E_CYCLE (iec61499::events) fires REQ every CYCLE once INIT is
confirmed; STOP stops it. Drop it into an application and connect the I/O.
Usage:
python3 st_to_61499.py FILE.st [--entry NAME] [--cycle T#10ms] [-o OUTDIR]
"""
import argparse
import datetime
import json
import os
import re
import sys
from xml.sax.saxutils import quoteattr
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from st_to_plcopen import read_sections # noqa: E402
TYPE_DEFAULT = {"BOOL": "FALSE", "REAL": "0.0", "LREAL": "0.0", "TIME": "T#0s", "STRING": "''", "WSTRING": '""'}
def find_pou(text, entry):
rx = re.compile(r"^\s*(PROGRAM|FUNCTION_BLOCK)\s+(\w+)(.*?)^\s*END_\1\b", re.S | re.M | re.I)
pous = [(m.group(1).upper(), m.group(2), m.group(3)) for m in rx.finditer(text)]
if not pous:
raise SystemExit("No PROGRAM or FUNCTION_BLOCK found.")
if entry:
pous = [p for p in pous if p[1].lower() == entry.lower()]
if not pous:
raise SystemExit("%s not found." % entry)
elif len(pous) > 1:
raise SystemExit("Several POUs (%s); choose one with --entry." % ", ".join(p[1] for p in pous))
return pous[0]
def split_vars(inner, warnings, name):
lines = inner.splitlines()
last_end = max([i for i, l in enumerate(lines) if re.match(r"^\s*END_VAR\b", l, re.I)], default=-1)
sections = read_sections(lines[:last_end + 1], warnings, name)
body = "\n".join(lines[last_end + 1:]).strip("\n")
ins, outs, internal = [], [], []
for sec in sections:
for v in sec["vars"]:
addr = (v.get("address") or "").upper()
doc = v.get("doc") or ""
if addr:
doc = ("was %s. " % addr) + doc
v = dict(v, doc=doc.strip())
if sec["kind"] == "VAR_INPUT" or addr.startswith("%I"):
ins.append(v)
elif sec["kind"] == "VAR_OUTPUT" or addr.startswith("%Q"):
outs.append(v)
elif addr.startswith("%M"):
warnings.append("%s was %s: internal here. If the HMI/SCADA writes it (a setpoint), make it a data input "
"fed by a communication FB (OPC UA, SUBSCRIBE, SERVER); if the HMI reads it, an output "
"published the same way." % (v["name"], addr))
internal.append(v)
elif sec["kind"] in ("VAR", "VAR_TEMP"):
if sec.get("retain"):
warnings.append("%s: RETAIN has no IEC 61499 equivalent; 4diac keeps internal values while the "
"resource runs. Persist it explicitly (e.g. a storage FB) if it must survive a restart." % v["name"])
internal.append(v)
else:
warnings.append("%s: %s variables are not supported in the wrapper; declared as internal." % (v["name"], sec["kind"]))
internal.append(v)
dedent = min([len(l) - len(l.lstrip()) for l in body.splitlines() if l.strip()] or [0])
body = "\n".join(l[dedent:] for l in body.splitlines())
return ins, outs, internal, body
def var_decl(v, indent):
a = 'Name=%s Type=%s' % (quoteattr(v["name"]), quoteattr(v["type"]))
if v.get("init"):
a += " InitialValue=%s" % quoteattr(v["init"])
if v.get("doc"):
a += " Comment=%s" % quoteattr(v["doc"])
return "%s<VarDeclaration %s/>" % (indent, a)
def with_list(vs, indent):
return "".join('\n%s<With Var=%s/>' % (indent, quoteattr(v["name"])) for v in vs)
def header(name, comment, src):
today = datetime.date.today().isoformat()
return ['<?xml version="1.0" encoding="UTF-8"?>',
'<FBType Name=%s Comment=%s>' % (quoteattr(name), quoteattr(comment)),
'\t<Identification Standard="61499-2" Description=%s/>' % quoteattr(
"Generated by Rungstone from %s. DRAFT FOR ENGINEER REVIEW - NOT APPROVED. Not for use on a production "
"system without review and testing by a qualified engineer." % src),
'\t<VersionInfo Organization="Rungstone" Version="0.1" Author="Rungstone (draft)" Date="%s" '
'Remarks="Generated draft"/>' % today]
def interface(ins, outs, indent="\t"):
x = ["%s<InterfaceList>" % indent, "%s\t<EventInputs>" % indent,
'%s\t\t<Event Name="INIT" Type="EInit" Comment="Initialization request">%s\n%s\t\t</Event>' % (
indent, with_list(ins, indent + "\t\t\t"), indent),
'%s\t\t<Event Name="REQ" Type="Event" Comment="Run one scan">%s\n%s\t\t</Event>' % (
indent, with_list(ins, indent + "\t\t\t"), indent),
"%s\t</EventInputs>" % indent, "%s\t<EventOutputs>" % indent,
'%s\t\t<Event Name="INITO" Type="EInit" Comment="Initialization confirm">%s\n%s\t\t</Event>' % (
indent, with_list(outs, indent + "\t\t\t"), indent),
'%s\t\t<Event Name="CNF" Type="Event" Comment="Scan done">%s\n%s\t\t</Event>' % (
indent, with_list(outs, indent + "\t\t\t"), indent),
"%s\t</EventOutputs>" % indent]
if ins:
x += ["%s\t<InputVars>" % indent] + [var_decl(v, indent + "\t\t") for v in ins] + ["%s\t</InputVars>" % indent]
if outs:
x += ["%s\t<OutputVars>" % indent] + [var_decl(v, indent + "\t\t") for v in outs] + ["%s\t</OutputVars>" % indent]
x.append("%s</InterfaceList>" % indent)
return x
def algorithm(name, body, comment):
text = "ALGORITHM %s\n%s\nEND_ALGORITHM" % (name, body)
return ['\t\t<Algorithm Name=%s Comment=%s>' % (quoteattr(name), quoteattr(comment)),
"\t\t\t<ST><![CDATA[%s]]></ST>" % text.replace("]]>", "]] >"), "\t\t</Algorithm>"]
def basic_fb(name, ins, outs, internal, body, src):
init_lines = ["%s := %s;" % (v["name"], v.get("init") or TYPE_DEFAULT.get(v["type"].upper().split("[")[0], "0"))
for v in outs]
x = header(name, "Scan logic translated from %s, run once per REQ (DRAFT)" % src, src)
x += interface(ins, outs)
x += ["\t<BasicFB>"]
if internal:
x += ["\t\t<InternalVars>"] + [var_decl(v, "\t\t\t") for v in internal] + ["\t\t</InternalVars>"]
x += ["\t\t<ECC>",
'\t\t\t<ECState Name="START" Comment="Initial state" x="550" y="425"/>',
'\t\t\t<ECState Name="INIT" Comment="Initialization" x="920" y="140">',
'\t\t\t\t<ECAction Algorithm="initialize" Output="INITO"/>', "\t\t\t</ECState>",
'\t\t\t<ECState Name="REQ" Comment="One scan" x="920" y="710">',
'\t\t\t\t<ECAction Algorithm="REQ" Output="CNF"/>', "\t\t\t</ECState>",
'\t\t\t<ECTransition Source="START" Destination="INIT" Condition="INIT" x="760" y="270"/>',
'\t\t\t<ECTransition Source="INIT" Destination="START" Condition="1" x="700" y="350"/>',
'\t\t\t<ECTransition Source="START" Destination="REQ" Condition="REQ" x="760" y="560"/>',
'\t\t\t<ECTransition Source="REQ" Destination="START" Condition="1" x="700" y="640"/>',
"\t\t</ECC>"]
x += algorithm("initialize", "\n".join(init_lines) or ";", "Outputs to their start values (a PLC clears non-retained outputs at start-up)")
x += algorithm("REQ", body, "One scan of the translated logic")
x += ["\t</BasicFB>", "</FBType>"]
return "\n".join(x) + "\n"
def app_fb(name, core, ins, outs, src):
x = header(name, "%s run every CYCLE by an E_CYCLE once initialized; STOP stops it (DRAFT)" % core, src)
x += ["\t<InterfaceList>", "\t\t<EventInputs>",
'\t\t\t<Event Name="INIT" Type="EInit" Comment="Initialize, then start cycling">%s\n\t\t\t</Event>' % with_list(ins, "\t\t\t\t"),
'\t\t\t<Event Name="STOP" Type="Event" Comment="Stop cycling"/>', "\t\t</EventInputs>", "\t\t<EventOutputs>",
'\t\t\t<Event Name="INITO" Type="EInit" Comment="Initialized">%s\n\t\t\t</Event>' % with_list(outs, "\t\t\t\t"),
'\t\t\t<Event Name="CNF" Type="Event" Comment="Scan done">%s\n\t\t\t</Event>' % with_list(outs, "\t\t\t\t"),
"\t\t</EventOutputs>"]
if ins:
x += ["\t\t<InputVars>"] + [var_decl(v, "\t\t\t") for v in ins] + ["\t\t</InputVars>"]
if outs:
x += ["\t\t<OutputVars>"] + [var_decl(v, "\t\t\t") for v in outs] + ["\t\t</OutputVars>"]
x += ["\t</InterfaceList>", "\t<FBNetwork>",
'\t\t<FB Name="Cycle" Type="iec61499::events::E_CYCLE" x="400" y="200"/>',
'\t\t<FB Name="Logic" Type=%s x="1400" y="200"/>' % quoteattr(core),
"\t\t<EventConnections>",
'\t\t\t<Connection Source="INIT" Destination="Logic.INIT"/>',
'\t\t\t<Connection Source="Logic.INITO" Destination="INITO"/>',
'\t\t\t<Connection Source="Logic.INITO" Destination="Cycle.START"/>',
'\t\t\t<Connection Source="STOP" Destination="Cycle.STOP"/>',
'\t\t\t<Connection Source="Cycle.EO" Destination="Logic.REQ"/>',
'\t\t\t<Connection Source="Logic.CNF" Destination="CNF"/>',
"\t\t</EventConnections>", "\t\t<DataConnections>",
'\t\t\t<Connection Source="CYCLE" Destination="Cycle.DT"/>']
x += ['\t\t\t<Connection Source=%s Destination=%s/>' % (quoteattr(v["name"]), quoteattr("Logic." + v["name"])) for v in ins]
x += ['\t\t\t<Connection Source=%s Destination=%s/>' % (quoteattr("Logic." + v["name"]), quoteattr(v["name"])) for v in outs]
x += ["\t\t</DataConnections>", "\t</FBNetwork>", "</FBType>"]
return "\n".join(x) + "\n"
def main(argv=None):
ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
ap.add_argument("file")
ap.add_argument("--entry")
ap.add_argument("--cycle", default="T#10ms", help="scan period carried on the CYCLE input (default T#10ms)")
ap.add_argument("-o", "--outdir", default=".")
a = ap.parse_args(argv)
with open(a.file, encoding="utf-8") as f:
text = f.read()
kind, name, inner = find_pou(text, a.entry)
warnings = []
ins, outs, internal, body = split_vars(inner, warnings, name)
if not any(v["name"].upper() == "CYCLE" for v in ins):
ins.append({"name": "CYCLE", "type": "TIME", "init": a.cycle, "doc": "Scan period: must equal the E_CYCLE DT"})
for v in internal:
if v["type"].upper().split("(")[0] in ("TON", "TOF", "TP", "R_TRIG", "F_TRIG", "CTU", "CTD", "CTUD", "RS", "SR"):
warnings.append("%s : %s is an FB instance; a 61499 Basic FB algorithm can't call it. Rewrite it with "
"CYCLE arithmetic (references/iec61499.md) before wrapping." % (v["name"], v["type"]))
src = os.path.basename(a.file)
os.makedirs(a.outdir, exist_ok=True)
core = name if not name.lower().startswith("prog_") else name[5:]
p1 = os.path.join(a.outdir, "%s.fbt" % core)
p2 = os.path.join(a.outdir, "%s_App.fbt" % core)
with open(p1, "w", encoding="utf-8") as f:
f.write(basic_fb(core, ins, outs, internal, body, src))
with open(p2, "w", encoding="utf-8") as f:
f.write(app_fb(core + "_App", core, ins, outs, src))
print(json.dumps({"written": [p1, p2], "fb": core, "inputs": [v["name"] for v in ins],
"outputs": [v["name"] for v in outs], "internal": [v["name"] for v in internal],
"warnings": warnings}, indent=1))
if __name__ == "__main__":
main()
SHA-256: 13a0da7b0b8edc81d7fc26e15e8d694292226411ddaacc10b5d24725bcd48621