lpcmanager

Enable IEC60870 server data loading and starting on PLC
iec60870
2 months ago, Dino Kosic
d8b12828e6aa
Enable IEC60870 server data loading and starting on PLC
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from __future__ import absolute_import
import io
import os
import re
import traceback
from six import text_type
from six.moves import xrange
from ConfigTreeNode import ConfigTreeNode
from PLCControler import LOCATION_CONFNODE, LOCATION_VAR_INPUT, \
LOCATION_VAR_OUTPUT, LOCATION_VAR_MEMORY
import util.paths as paths
from iec60870.iec60870_utils import iec_iec_type_to_bind_kind
# (type_id, IEC_type, datasize, direction, size_code, description)
# direction: "I" for monitor (input), "Q" for control (output), "M" for memory
# size_code: "X" for bit, "B" for byte, "W" for word, "D" for dword
iec60870_asdu_types = {
"M_SP_NA_1 - Single point": (1, "BOOL", 1, "I", "X", "Single Point Information"),
"M_DP_NA_1 - Double point": (3, "BYTE", 8, "I", "B", "Double Point Information"),
"M_ST_NA_1 - Step position": (5, "INT", 16, "I", "W", "Step Position Information"),
"M_ME_NA_1 - Measured normalized": (9, "WORD", 16, "I", "W", "Measured Value Normalized"),
"M_ME_NB_1 - Measured scaled": (11, "INT", 16, "I", "W", "Measured Value Scaled"),
"M_ME_NC_1 - Measured float": (13, "REAL", 32, "I", "D", "Measured Value Short Float"),
"C_SC_NA_1 - Single command": (45, "BOOL", 1, "Q", "X", "Single Command"),
"C_DC_NA_1 - Double command": (46, "BYTE", 8, "Q", "B", "Double Command"),
"C_RC_NA_1 - Step command": (47, "BYTE", 8, "Q", "B", "Regulating Step Command"),
"C_SE_NA_1 - Setpoint normalized": (48, "WORD", 16, "Q", "W", "Set Point Normalized"),
"C_SE_NB_1 - Setpoint scaled": (49, "INT", 16, "Q", "W", "Set Point Scaled"),
"C_SE_NC_1 - Setpoint float": (50, "REAL", 32, "Q", "D", "Set Point Short Float"),
}
LOCATION_TYPES = {
"I": LOCATION_VAR_INPUT,
"Q": LOCATION_VAR_OUTPUT,
"M": LOCATION_VAR_MEMORY,
}
# XSD fragment for the shared IEC 60870-5-104 connection parameters
# (used in both server and client node XSD definitions)
_IEC60870_CONN_PARAMS_XSD = """\
<xsd:attribute name="APCI_k" use="optional" default="12">
<xsd:simpleType>
<xsd:restriction base="xsd:integer">
<xsd:minInclusive value="1"/>
<xsd:maxInclusive value="32767"/>
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="APCI_w" use="optional" default="8">
<xsd:simpleType>
<xsd:restriction base="xsd:integer">
<xsd:minInclusive value="1"/>
<xsd:maxInclusive value="32767"/>
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="Timeout_t0" use="optional" default="10">
<xsd:simpleType>
<xsd:restriction base="xsd:integer">
<xsd:minInclusive value="1"/>
<xsd:maxInclusive value="255"/>
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="Timeout_t1" use="optional" default="15">
<xsd:simpleType>
<xsd:restriction base="xsd:integer">
<xsd:minInclusive value="1"/>
<xsd:maxInclusive value="255"/>
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="Timeout_t2" use="optional" default="10">
<xsd:simpleType>
<xsd:restriction base="xsd:integer">
<xsd:minInclusive value="1"/>
<xsd:maxInclusive value="255"/>
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="Timeout_t3" use="optional" default="20">
<xsd:simpleType>
<xsd:restriction base="xsd:integer">
<xsd:minInclusive value="1"/>
<xsd:maxInclusive value="255"/>
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="Use_TLS" type="xsd:boolean" use="optional" default="false"/>
<xsd:attribute name="CA_Size" use="optional" default="2">
<xsd:simpleType>
<xsd:restriction base="xsd:integer">
<xsd:minInclusive value="1"/>
<xsd:maxInclusive value="2"/>
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="IOA_Size" use="optional" default="3">
<xsd:simpleType>
<xsd:restriction base="xsd:integer">
<xsd:minInclusive value="1"/>
<xsd:maxInclusive value="3"/>
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="COT_Has_OA" type="xsd:boolean" use="optional" default="true"/>
<xsd:attribute name="OA_Value" use="optional" default="10">
<xsd:simpleType>
<xsd:restriction base="xsd:integer">
<xsd:minInclusive value="0"/>
<xsd:maxInclusive value="255"/>
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="Use_Local_Timezone" type="xsd:boolean" use="optional" default="true"/>
"""
def _srv_attr_map(server_plug):
for element in server_plug.GetParamsAttributes():
if element["name"] == "IEC60870ServerNode":
return {c["name"]: c["value"] for c in element["children"]}
return {}
def _c_escape_str(s):
return s.replace("\\", "\\\\").replace("\"", "\\\"")
def _data_point_ct(child, index):
"""Attribute value from IEC60870DataPoint element by XSD order (0=ASDU, 1=IOA, 2=count)."""
for element in child.GetParamsAttributes():
if element["name"] == "IEC60870DataPoint":
return element["children"][index]["value"]
raise KeyError("IEC60870DataPoint")
def _location_tuple_from_tree_string(locstr):
"""
Map a plugin tree location string (e.g. D4.0.0, X4.0.2) to the numeric LOC tuple
in LOCATED_VARIABLES.h / ProjectController.GetLocations().
"""
if not locstr:
return None
i = 0
while i < len(locstr) and not locstr[i].isdigit():
i += 1
body = locstr[i:]
if not body:
return None
return tuple(int(x) for x in body.split("."))
def _find_name_for_loc_tuple(target_tuple, iterable_of_locdicts):
"""C symbol NAME for located var with LOC == target_tuple."""
if not target_tuple:
return None
want = tuple(target_tuple)
for locdic in iterable_of_locdicts:
loc = locdic.get("LOC")
if loc is None:
continue
if tuple(loc) == want:
return str(locdic["NAME"])
return None
def _resolve_diagnostics_by_suffix_tail(srv_loc, iterable_of_locdicts):
"""
Match %MD…0 / %MD…1 / %MX…2 when matiec LOC has a longer prefix than
GetCurrentLocation() but ends with the same address tail (e.g. extra
leading indices from configuration / resource).
"""
if not iterable_of_locdicts:
return None, None, None
srv_loc = tuple(srv_loc)
by_loc = {}
for locdic in iterable_of_locdicts:
loc = locdic.get("LOC")
if not loc:
continue
by_loc[tuple(loc)] = locdic
prefixes = set()
for locdic in iterable_of_locdicts:
loc = locdic.get("LOC")
if loc and len(loc) >= 2:
prefixes.add(tuple(loc[:-1]))
# Prefer longest P first so a resource-prefixed path wins over a shorter tail.
for P in sorted(prefixes, key=lambda x: -len(x)):
if len(P) < len(srv_loc):
continue
if P[-len(srv_loc):] != srv_loc:
continue
k0 = P + (0,)
k1 = P + (1,)
k2 = P + (2,)
d0 = by_loc.get(k0)
d1 = by_loc.get(k1)
d2 = by_loc.get(k2)
if not (d0 and d1 and d2):
continue
if (d0.get("IEC_TYPE") == "UDINT" and d1.get("IEC_TYPE") == "UDINT"
and d2.get("IEC_TYPE") == "BOOL"):
return str(d0["NAME"]), str(d1["NAME"]), str(d2["NAME"])
return None, None, None
def _safe_text(x):
"""Python 2/3: unified unicode/str for diag file content."""
if isinstance(x, text_type):
return x
if isinstance(x, bytes):
return x.decode("utf-8", "replace")
return text_type(x)
# Beremiz ProjectController.GetLocations() only accepts LOC without spaces
# (see LOC pattern [,0-9]*). Some matiec builds emit "4, 0, 0" and return
# zero entries while LOCATED_VARIABLES.h is non-empty. Parse those here.
_LOOSE_LOCATED_LINE = re.compile(
r"__LOCATED_VAR\s*\(\s*"
r"(?P<IEC_TYPE>[A-Z][A-Z0-9_]*)\s*,\s*"
r"(?P<NAME>[_A-Za-z0-9]+)\s*,\s*"
r"(?P<DIR>[QMI])\s*"
r"(?:,\s*(?P<SIZE>[XBWDL])\s*)?"
r",\s*(?P<LOC>[\d,\s]+)\)\s*"
r"(?:;)?\s*(?://.*)?"
)
def _parse_located_variables_h_loose(filepath):
"""
Same structure as ProjectController.GetLocations() resdicts, with
whitespace-tolerant LOC lists.
"""
locations = []
if not filepath or not os.path.isfile(filepath):
return locations
try:
with open(filepath, "r") as f:
lines = f.readlines()
except Exception:
return locations
for line in lines:
line = line.strip()
if "__LOCATED_VAR" not in line:
continue
if line.startswith("//") or line.startswith("/*"):
continue
m = _LOOSE_LOCATED_LINE.search(line)
if not m:
continue
resdict = m.groupdict()
loc_raw = resdict["LOC"]
loc_raw = loc_raw.replace(" ", "").replace("\t", "")
if not loc_raw:
continue
try:
resdict["LOC"] = tuple(map(int, loc_raw.split(",")))
except Exception:
continue
if not resdict.get("SIZE"):
resdict["SIZE"] = "X"
locations.append(resdict)
return locations
def _write_iec60870_diag_failure_file(
buildpath, srv, locations, project_locations, srv_get_locs):
"""Write build/iec60870_diag_failed.txt (UTF-8). Always leaves non-empty text."""
p = os.path.join(buildpath or ".", "iec60870_diag_failed.txt")
lines_u = []
try:
lines_u.append(_safe_text("IEC60870 - diagnostic variable resolution failed"))
lines_u.append(_safe_text(""))
lines_u.append(_safe_text("server_path (GetCurrentLocation): %s") % _safe_text(
".".join(map(str, srv.GetCurrentLocation()))))
root = srv.GetCTRoot()
lv_file = os.path.join(root._getBuildPath(), "LOCATED_VARIABLES.h")
lines_u.append(_safe_text("buildpath (this run): %s") % _safe_text(buildpath))
tree = srv.GetVariableLocationTree()
ch = tree.get("children") or []
lines_u.append(_safe_text("tree diagnostic location strings (first 3 children):"))
for i in range(min(3, len(ch))):
loc_raw = ch[i].get("location", "") if hasattr(ch[i], "get") else ""
lines_u.append(_safe_text(" [%s] %s") % (i, _safe_text(repr(loc_raw))))
t0 = _location_tuple_from_tree_string(
ch[0].get("location", "")) if len(ch) > 0 else None
t1 = _location_tuple_from_tree_string(
ch[1].get("location", "")) if len(ch) > 1 else None
t2 = _location_tuple_from_tree_string(
ch[2].get("location", "")) if len(ch) > 2 else None
lines_u.append(_safe_text("parsed tuples (matiec should match LOC): %s, %s, %s") % (
t0, t1, t2))
lines_u.append(_safe_text(""))
lines_u.append(_safe_text("LOCATED_VARIABLES.h: %s") % _safe_text(lv_file))
lines_u.append(_safe_text("file exists: %s") % _safe_text(os.path.isfile(lv_file)))
lines_u.append(_safe_text("len(locations pass to CTNGenerate_C): %s") % (
len(locations) if locations else 0,))
lines_u.append(_safe_text("len(GetCTRoot().GetLocations()): %s") % (
len(project_locations) if project_locations else 0,))
lines_u.append(_safe_text("len(srv.GetLocations()): %s") % len(srv_get_locs))
lv_bp = os.path.join(buildpath or ".", "LOCATED_VARIABLES.h")
loose_n = len(_parse_located_variables_h_loose(lv_bp))
lines_u.append(_safe_text("len(iec60870 loose LOCATED_VARIABLES parse): %s") % loose_n)
lines_u.append(_safe_text(""))
lines_u.append(_safe_text(
"Memory vars with LOC ending in last index 0, 1, or 2 (sample):"))
seen = set()
n = 0
for locdic in (project_locations or []) + (locations or []):
loc = locdic.get("LOC")
if not loc:
continue
if loc[-1] not in (0, 1, 2):
continue
if locdic.get("DIR") != "M":
continue
key = tuple(loc)
if key in seen:
continue
seen.add(key)
lines_u.append(_safe_text(
" NAME=%s IEC=%s SIZE=%s LOC=%s") % (
_safe_text(locdic.get("NAME")),
_safe_text(locdic.get("IEC_TYPE")),
_safe_text(locdic.get("SIZE")),
_safe_text(loc),
))
n += 1
if n >= 80:
break
if loose_n == 0 and os.path.isfile(lv_bp):
try:
with open(lv_bp, "rb") as rf:
head = rf.read(2500)
lines_u.append(_safe_text(
"--- LOCATED_VARIABLES.h (first 2500 bytes, repr) ---"))
lines_u.append(_safe_text(repr(head)))
if not head:
lines_u.append(_safe_text(
"NOTE: file is empty (0 bytes). matiec may not have "
"written it yet for this build step, or the toolchain "
"touched a stub. Plugin uses intrinsic static "
"diagnostic counters in IEC104_*.c (not IEC-visible)."))
except Exception:
pass
except Exception as exc:
lines_u = [
_safe_text("IEC60870 - diagnostic dump failed while collecting data"),
_safe_text(repr(exc)),
_safe_text(traceback.format_exc()),
]
body = _safe_text("\n").join(lines_u) + _safe_text("\n")
try:
with io.open(p, "w", encoding="utf-8") as out:
out.write(body)
except Exception as exc2:
try:
with io.open(p, "w", encoding="utf-8") as out:
out.write(_safe_text(
"Could not write UTF-8 diag file: %s\n%s"
% (repr(exc2), traceback.format_exc())))
except Exception:
pass
def _resolve_server_diagnostic_names(srv, locations, project_locations):
"""
Resolve read/write/connection to PLC symbol names, or request intrinsic storage.
Returns (rd, wr, cn, mode) with mode \"plc\" (bind to matiec globals) or
\"intrinsic\" (server-local static UDINT/IEC_BOOL in IEC104_*.c — used when
LOCATED_VARIABLES.h is empty or names cannot be resolved, to avoid bogus
externs like __MD4_0_0 that the linker does not provide).
"""
loc_sources = []
for src in (locations, project_locations):
if src and src not in loc_sources:
loc_sources.append(src)
tree = srv.GetVariableLocationTree()
children = tree.get("children") or []
if len(children) >= 3:
t0 = _location_tuple_from_tree_string(children[0].get("location", ""))
t1 = _location_tuple_from_tree_string(children[1].get("location", ""))
t2 = _location_tuple_from_tree_string(children[2].get("location", ""))
if t0 and t1 and t2:
for src in loc_sources:
if not src:
continue
rd = _find_name_for_loc_tuple(t0, src)
wr = _find_name_for_loc_tuple(t1, src)
cn = _find_name_for_loc_tuple(t2, src)
if rd and wr and cn:
return rd, wr, cn, "plc"
srv_loc = tuple(srv.GetCurrentLocation())
for src in loc_sources:
if not src:
continue
rd = wr = cn = None
for locdic in src:
loc = locdic.get("LOC")
if loc is None:
continue
loc = tuple(loc)
if len(loc) != len(srv_loc) + 1:
continue
if loc[:-1] != srv_loc:
continue
idx = loc[-1]
name = str(locdic["NAME"])
if idx == 0:
rd = name
elif idx == 1:
wr = name
elif idx == 2:
cn = name
if rd and wr and cn:
return rd, wr, cn, "plc"
merged = []
seen_loc = set()
for src in loc_sources:
if not src:
continue
for d in src:
loc = d.get("LOC")
if loc is None:
continue
t = tuple(loc)
if t in seen_loc:
continue
seen_loc.add(t)
merged.append(d)
rd, wr, cn = _resolve_diagnostics_by_suffix_tail(srv_loc, merged)
if rd and wr and cn:
return rd, wr, cn, "plc"
fallback = srv.GetLocations()
if fallback:
rd, wr, cn = _resolve_diagnostics_by_suffix_tail(srv_loc, fallback)
if rd and wr and cn:
return rd, wr, cn, "plc"
rd = wr = cn = None
for locdic in fallback:
loc = locdic.get("LOC")
if loc is None:
continue
loc = tuple(loc)
if len(loc) != len(srv_loc) + 1:
continue
if loc[:-1] != srv_loc:
continue
idx = loc[-1]
name = str(locdic["NAME"])
if idx == 0:
rd = name
elif idx == 1:
wr = name
elif idx == 2:
cn = name
if rd and wr and cn:
return rd, wr, cn, "plc"
return None, None, None, "intrinsic"
#
# D A T A P O I N T
#
class _DataPointPlug(object):
XSD = """<?xml version="1.0" encoding="ISO-8859-1" ?>
<xsd:schema xmlns:xsd="http://www.w3.org/2001/XMLSchema">
<xsd:element name="IEC60870DataPoint">
<xsd:complexType>
<xsd:attribute name="ASDU_Type" type="xsd:string" use="optional" default="M_SP_NA_1 - Single point"/>
<xsd:attribute name="IOA" use="optional" default="0">
<xsd:simpleType>
<xsd:restriction base="xsd:integer">
<xsd:minInclusive value="0"/>
<xsd:maxInclusive value="16777215"/>
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="Nr_of_Points" use="optional" default="1">
<xsd:simpleType>
<xsd:restriction base="xsd:integer">
<xsd:minInclusive value="1"/>
<xsd:maxInclusive value="65535"/>
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
</xsd:schema>
"""
def GetParamsAttributes(self, path=None):
infos = ConfigTreeNode.GetParamsAttributes(self, path=path)
for element in infos:
if element["name"] == "IEC60870DataPoint":
for child in element["children"]:
if child["name"] == "ASDU_Type":
_list = sorted(iec60870_asdu_types.keys())
child["type"] = _list
return infos
def GetVariableLocationTree(self):
current_location = self.GetCurrentLocation()
name = self.BaseParams.getName()
ioa = self.GetParamsAttributes()[0]["children"][1]["value"]
count = self.GetParamsAttributes()[0]["children"][2]["value"]
asdu_type_str = self.GetParamsAttributes()[0]["children"][0]["value"]
type_id, datatype, datasize, direction, size_code, desc = \
iec60870_asdu_types[asdu_type_str]
loc_type = LOCATION_TYPES[direction]
entries = []
for offset in range(ioa, ioa + count):
entries.append({
"name": desc + " IOA " + str(offset),
"type": loc_type,
"size": datasize,
"IEC_type": datatype,
"var_name": "IEC104_" + str(type_id) + "_" + str(offset),
"location": size_code + ".".join(
[str(i) for i in current_location]) + "." + str(offset),
"description": desc,
"children": []})
return {"name": name,
"type": LOCATION_CONFNODE,
"location": ".".join(
[str(i) for i in current_location]) + ".x",
"children": entries}
def CTNGenerate_C(self, buildpath, locations):
return [], "", False
#
# C O N T R O L L E D S T A T I O N (Server)
#
class _IEC60870ServerPlug(object):
XSD = ("""<?xml version="1.0" encoding="ISO-8859-1" ?>
<xsd:schema xmlns:xsd="http://www.w3.org/2001/XMLSchema">
<xsd:element name="IEC60870ServerNode">
<xsd:complexType>
<xsd:attribute name="Configuration_Name" type="xsd:string" use="optional" default=""/>
<xsd:attribute name="Local_IP_Address" type="xsd:string" use="optional" default="#ANY#"/>
<xsd:attribute name="Local_Port_Number" type="xsd:string" use="optional" default="2404"/>
<xsd:attribute name="Common_Address" use="optional" default="1">
<xsd:simpleType>
<xsd:restriction base="xsd:integer">
<xsd:minInclusive value="1"/>
<xsd:maxInclusive value="65534"/>
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
"""
+ _IEC60870_CONN_PARAMS_XSD +
"""
</xsd:complexType>
</xsd:element>
</xsd:schema>
""")
CTNChildrenTypes = [("IEC60870DataPoint", _DataPointPlug, "Data Point")]
PlugType = "IEC60870Server"
def __init__(self):
loc_str = ".".join(map(str, self.GetCurrentLocation()))
self.IEC60870ServerNode.setConfiguration_Name(
"IEC60870 Server " + loc_str)
def GetNodeCount(self):
return (1, 0)
def GetConfigName(self):
return self.IEC60870ServerNode.getConfiguration_Name()
def GetIPServerPortNumbers(self):
port = self.IEC60870ServerNode.getLocal_Port_Number()
addr = self.IEC60870ServerNode.getLocal_IP_Address()
return [(self.GetCurrentLocation(), addr, port)]
def GetVariableLocationTree(self):
current_location = self.GetCurrentLocation()
name = self.BaseParams.getName()
entries = []
entries.append({
"name": "Read Request Counter",
"type": LOCATION_VAR_MEMORY,
"size": 32,
"IEC_type": "UDINT",
"var_name": "var_name",
"location": "D" + ".".join(
[str(i) for i in current_location]) + ".0",
"description": "IEC60870 read request counter",
"children": []})
entries.append({
"name": "Write Request Counter",
"type": LOCATION_VAR_MEMORY,
"size": 32,
"IEC_type": "UDINT",
"var_name": "var_name",
"location": "D" + ".".join(
[str(i) for i in current_location]) + ".1",
"description": "IEC60870 write request counter",
"children": []})
entries.append({
"name": "Connection Active Flag",
"type": LOCATION_VAR_MEMORY,
"size": 1,
"IEC_type": "BOOL",
"var_name": "var_name",
"location": "X" + ".".join(
[str(i) for i in current_location]) + ".2",
"description": "IEC60870 connection active flag",
"children": []})
for child in self.IECSortedChildren():
entries.append(child.GetVariableLocationTree())
return {"name": name,
"type": LOCATION_CONFNODE,
"location": ".".join(
[str(i) for i in current_location]) + ".x",
"children": entries}
def CTNGenerate_C(self, buildpath, locations):
return [], "", False
#
# C O N T R O L L I N G S T A T I O N (Client)
#
class _IEC60870ClientPlug(object):
XSD = ("""<?xml version="1.0" encoding="ISO-8859-1" ?>
<xsd:schema xmlns:xsd="http://www.w3.org/2001/XMLSchema">
<xsd:element name="IEC60870ClientNode">
<xsd:complexType>
<xsd:attribute name="Configuration_Name" type="xsd:string" use="optional" default=""/>
<xsd:attribute name="Remote_IP_Address" type="xsd:string" use="optional" default="localhost"/>
<xsd:attribute name="Remote_Port_Number" type="xsd:string" use="optional" default="2404"/>
<xsd:attribute name="Common_Address" use="optional" default="1">
<xsd:simpleType>
<xsd:restriction base="xsd:integer">
<xsd:minInclusive value="1"/>
<xsd:maxInclusive value="65534"/>
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="Polling_Interval_ms" use="optional" default="1000">
<xsd:simpleType>
<xsd:restriction base="xsd:unsignedLong">
<xsd:minInclusive value="0"/>
<xsd:maxInclusive value="2147483647"/>
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
"""
+ _IEC60870_CONN_PARAMS_XSD +
"""
</xsd:complexType>
</xsd:element>
</xsd:schema>
""")
CTNChildrenTypes = [("IEC60870DataPoint", _DataPointPlug, "Data Point")]
PlugType = "IEC60870Client"
def __init__(self):
loc_str = ".".join(map(str, self.GetCurrentLocation()))
self.IEC60870ClientNode.setConfiguration_Name(
"IEC60870 Client " + loc_str)
def GetNodeCount(self):
return (1, 0)
def GetConfigName(self):
return self.IEC60870ClientNode.getConfiguration_Name()
def GetVariableLocationTree(self):
current_location = self.GetCurrentLocation()
name = self.BaseParams.getName()
entries = []
entries.append({
"name": "Connection Status",
"type": LOCATION_VAR_MEMORY,
"size": 8,
"IEC_type": "BYTE",
"var_name": "var_name",
"location": "B" + ".".join(
[str(i) for i in current_location]) + ".0",
"description": "Connection status (0=disconnected, 1=connected, "
"2=connecting, 3=error)",
"children": []})
entries.append({
"name": "Interrogation Trigger",
"type": LOCATION_VAR_MEMORY,
"size": 1,
"IEC_type": "BOOL",
"var_name": "var_name",
"location": "X" + ".".join(
[str(i) for i in current_location]) + ".1",
"description": "Trigger general interrogation",
"children": []})
for child in self.IECSortedChildren():
entries.append(child.GetVariableLocationTree())
return {"name": name,
"type": LOCATION_CONFNODE,
"location": ".".join(
[str(i) for i in current_location]) + ".x",
"children": entries}
def CTNGenerate_C(self, buildpath, locations):
return [], "", False
#
# R O O T C L A S S
#
def _lt_to_str(loctuple):
return '.'.join(map(str, loctuple))
class RootClass(object):
XSD = """<?xml version="1.0" encoding="ISO-8859-1" ?>
<xsd:schema xmlns:xsd="http://www.w3.org/2001/XMLSchema">
<xsd:element name="IEC60870Root">
<xsd:complexType>
<xsd:attribute name="MaxRemoteClients" use="optional" default="10">
<xsd:simpleType>
<xsd:restriction base="xsd:integer">
<xsd:minInclusive value="0"/>
<xsd:maxInclusive value="65535"/>
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
</xsd:schema>
"""
CTNChildrenTypes = [
("IEC60870Server", _IEC60870ServerPlug, "IEC 60870-5-104 Server"),
("IEC60870Client", _IEC60870ClientPlug, "IEC 60870-5-104 Client"),
]
def GetNodeCount(self):
max_remote_clients = self.GetParamsAttributes()[0]["children"][0]["value"]
total = (max_remote_clients, 0)
for child in self.IECSortedChildren():
total = tuple(
x1 + x2 for x1, x2 in zip(total, child.GetNodeCount()))
return total
def GetIPServerPortNumbers(self):
port_numbers = []
for child in self.IECSortedChildren():
if child.CTNType == "IEC60870Server":
port_numbers.extend(child.GetIPServerPortNumbers())
return port_numbers
def GetConfigNames(self):
names = []
for child in self.IECSortedChildren():
names.append(
(child.GetCurrentLocation(), child.GetConfigName()))
return names
def CTNGenerate_C(self, buildpath, locations):
node_config_names = []
for CTNInstance in self.GetCTRoot().IterChildren():
if CTNInstance.CTNType == "iec60870":
node_config_names.extend(CTNInstance.GetConfigNames())
for i in range(0, len(node_config_names) - 1):
for j in range(i + 1, len(node_config_names)):
if node_config_names[i][1] == node_config_names[j][1]:
error_message = _(
"Error: IEC60870 plugin nodes %%{a1}.x and %%{a2}.x "
"use the same Configuration_Name \"{a3}\".\n"
).format(
a1=_lt_to_str(node_config_names[i][0]),
a2=_lt_to_str(node_config_names[j][0]),
a3=node_config_names[j][1])
self.FatalError(error_message)
ip_ports = []
for CTNInstance in self.GetCTRoot().IterChildren():
if CTNInstance.CTNType == "iec60870":
ip_ports.extend(CTNInstance.GetIPServerPortNumbers())
i = 0
for loc1, addr1, port1 in ip_ports[:-1]:
i = i + 1
for loc2, addr2, port2 in ip_ports[i:]:
if (port1 == port2) and (
(addr1 == addr2)
or (addr1 in ("", "*", "#ANY#"))
or (addr2 in ("", "*", "#ANY#"))
):
error_message = _(
"Error: IEC60870 plugin nodes %%{a1}.x and %%{a2}.x "
"use same port number \"{a3}\" on the same "
"(or overlapping) network interfaces "
"\"{a4}\" and \"{a5}\".\n"
).format(
a1=_lt_to_str(loc1), a2=_lt_to_str(loc2),
a3=port1, a4=addr1, a5=addr2)
self.FatalError(error_message)
loc_prefix = "_".join(map(str, self.GetCurrentLocation()))
servers = [ch for ch in self.IECSortedChildren() if getattr(
ch, "PlugType", None) == "IEC60870Server"]
if not servers:
return [], "", False
IEC60870Path = paths.ThirdPartyPath("IEC60870")
for ch in servers:
tls = _srv_attr_map(ch).get("Use_TLS", False)
if str(tls).lower() in ("true", "1", "yes"):
self.FatalError(
_("IEC60870: TLS is not supported by the generated "
"CS104 runtime. Disable Use_TLS on server \"%s\".\n") %
ch.BaseParams.getName())
extern_lines = []
extern_seen = set()
bindings_rows = []
srv_rows = []
diag_static_lines = []
def add_extern(iec_type, name):
key = (iec_type, name)
if key in extern_seen:
return
extern_seen.add(key)
extern_lines.append(
"extern %(t)s %(n)s;" % {"t": iec_type, "n": name})
max_remote = int(self.GetParamsAttributes()[0]["children"][0]["value"])
project_locs = self.GetCTRoot().GetLocations()
lv_path = os.path.join(buildpath, "LOCATED_VARIABLES.h")
loose_locs = _parse_located_variables_h_loose(lv_path)
if len(loose_locs) > len(project_locs or []):
project_locs = loose_locs
for server_id, srv in enumerate(servers):
smap = _srv_attr_map(srv)
rd_name, wr_name, conn_name, diag_mode = _resolve_server_diagnostic_names(
srv, locations, project_locs)
if diag_mode != "intrinsic" and not (
rd_name and wr_name and conn_name):
_write_iec60870_diag_failure_file(
buildpath, srv, locations, project_locs,
srv.GetLocations())
dbg = os.path.join(buildpath, "iec60870_diag_failed.txt")
try:
self.GetCTRoot().logger.write_error(
_("IEC60870: diagnostic bind failed for server at %s. "
"Details: %s\n") % (
".".join(map(str, srv.GetCurrentLocation())),
dbg))
except Exception:
pass
self.FatalError(
_("IEC60870: Could not resolve server diagnostic variables "
"(read/write/connection) for server at %s.\n"
"See \"%s\" in the project build folder.\n") % (
".".join(map(str, srv.GetCurrentLocation())),
dbg))
if diag_mode == "intrinsic":
rd_name = "iec60870_diag_rd_%d" % server_id
wr_name = "iec60870_diag_wr_%d" % server_id
conn_name = "iec60870_diag_conn_%d" % server_id
diag_static_lines.append(
"static UDINT %(rd)s = 0U;\n"
"static UDINT %(wr)s = 0U;\n"
"static IEC_BOOL %(cn)s = (IEC_BOOL)0;\n" % {
"rd": rd_name,
"wr": wr_name,
"cn": conn_name,
})
else:
add_extern("UDINT", rd_name)
add_extern("UDINT", wr_name)
add_extern("BOOL", conn_name)
ip = smap.get("Local_IP_Address", "#ANY#")
if ip in ("", "*", "#ANY#"):
ip_c = "0.0.0.0"
else:
ip_c = _c_escape_str(str(ip))
port = int(smap.get("Local_Port_Number", 2404))
common = int(smap.get("Common_Address", 1))
ca_sz = int(smap.get("CA_Size", 2))
ioa_sz = int(smap.get("IOA_Size", 3))
cot_has_oa = str(smap.get("COT_Has_OA", True)).lower() in (
"true", "1", "yes")
oa = int(smap.get("OA_Value", 10))
apci_k = int(smap["APCI_k"])
apci_w = int(smap["APCI_w"])
t0 = int(smap["Timeout_t0"])
t1 = int(smap["Timeout_t1"])
t2 = int(smap["Timeout_t2"])
t3 = int(smap["Timeout_t3"])
loc_label = _c_escape_str(
".".join(map(str, srv.GetCurrentLocation())))
srv_rows.append(
"""
{ .loc_label = "%(loc)s",
.common_address = %(co)d,
.ip_str = "%(ipc)s",
.port = %(port)d,
.max_open = %(maxc)d,
.ca_sz = %(casz)d,
.ioa_sz = %(ioasz)d,
.cot_two_byte = %(cotoa)d,
.oa = %(oa)d,
.apci_k = %(apk)d, .apci_w = %(apw)d, .t0 = %(t0)d, .t1 = %(t1)d,
.t2 = %(t2)d, .t3 = %(t3)d,
.rd_ctr = &(%(rd)s),
.wr_ctr = &(%(wr)s),
.conn_bool = &(%(cn)s),
.slave = NULL,
.init_st = 0
},""" % {
"loc": loc_label,
"co": common,
"ipc": ip_c,
"port": port,
"maxc": max_remote,
"casz": ca_sz,
"ioasz": ioa_sz,
"cotoa": 1 if cot_has_oa else 0,
"oa": oa,
"apk": apci_k,
"apw": apci_w,
"t0": t0,
"t1": t1,
"t2": t2,
"t3": t3,
"rd": rd_name,
"wr": wr_name,
"cn": conn_name,
})
for dp in srv.IECSortedChildren():
asdu_str = _data_point_ct(dp, 0)
ioa0 = int(_data_point_ct(dp, 1))
npoints = int(_data_point_ct(dp, 2))
tid, _dt, _ds, direction, _sc, _d = iec60870_asdu_types[asdu_str]
is_cmd = 1 if direction == "Q" else 0
for iecvar in dp.GetLocations():
loc = iecvar["LOC"]
if len(loc) < 4:
continue
ioa_v = int(loc[-1])
if ioa_v < ioa0 or ioa_v >= ioa0 + npoints:
continue
nm = str(iecvar["NAME"])
iet = iecvar["IEC_TYPE"]
add_extern(iet, nm)
bk = iec_iec_type_to_bind_kind(iet)
bindings_rows.append(
" { %(sid)d, %(tid)d, %(ioa)d, %(isc)d, %(bk)d,"
" (void *)&(%(nm)s) }," % {
"sid": server_id,
"tid": tid,
"ioa": ioa_v,
"isc": is_cmd,
"bk": bk,
"nm": nm,
})
num_srv = len(servers)
if not bindings_rows:
bindings_rows.append(
" { -1, 0, 0, 0, 0, NULL }, /* placeholder */")
num_bind = len(bindings_rows)
extern_block = "\n".join(extern_lines)
bindings_body = "\n".join(bindings_rows)
srv_def = (
"static iec60870_srv_t iec60870_srv[IEC60870_NUM_SERVERS_%s] = {%s\n};"
% (loc_prefix, "".join(srv_rows)))
diag_static_block = "".join(diag_static_lines)
if diag_static_block:
diag_static_block = (
"/* Intrinsic diagnostic storage — not mapped to IEC %%MD/%%MX */\n"
+ diag_static_block)
tpl = {
"locstr": loc_prefix,
"diag_static_block": diag_static_block,
"extern_block": extern_block,
"bindings_rows": bindings_body,
"srv_def": srv_def,
"num_servers": str(num_srv),
"num_bindings": str(num_bind),
"max_remote_clients": str(max_remote),
}
c_src = os.path.join(os.path.split(__file__)[0], "iec60870_runtime.c")
h_src = os.path.join(os.path.split(__file__)[0], "iec60870_runtime.h")
gen_c = os.path.join(buildpath, "IEC104_%s.c" % loc_prefix)
gen_h = os.path.join(buildpath, "IEC104_%s.h" % loc_prefix)
with open(h_src, "r") as f:
h_text = f.read() % tpl
with open(gen_h, "w") as fh:
fh.write(h_text)
with open(c_src, "r") as f:
c_text = f.read() % tpl
with open(gen_c, "w") as fc:
fc.write(c_text)
LDFLAGS = []
LDFLAGS.append(' "-L' + IEC60870Path + '"')
LDFLAGS.append(' "' + os.path.join(IEC60870Path, "liblib60870.a") + '"')
LDFLAGS.append(' "-lpthread"')
cflags = ' -I"' + IEC60870Path + '"'
return (
[(gen_c, cflags)],
LDFLAGS,
True,
)