lpcmanager

Improve naming in IEC60870 extension
py2compat tip
13 days ago, dino
2ac146c1b16c
Parents a03f3c87d03c
Children
Improve naming in IEC60870 extension
1 files changed, 95 insertions(+), 12 deletions(-)
  • +95 -12
    iec60870/iec60870.py
    • --- a/iec60870/iec60870.py Tue Jun 02 09:02:31 2026 +0200
    +++ b/iec60870/iec60870.py Thu Jun 04 09:32:25 2026 +0200
    @@ -269,6 +269,67 @@
    # Client node still uses the combined fragment (hidden from IDE)
    _IEC60870_CONN_PARAMS_XSD = _IEC60870_APCI_PARAMS_XSD + _IEC60870_APP_LAYER_PARAMS_XSD
    +# Short confnode type names (folder paths: Name@Type, e.g. CA_0@CommonAddress)
    +CTN_SERVER = "Server"
    +CTN_COMMON_ADDRESS = "CommonAddress"
    +CTN_DATA_POINT = "DataPoint"
    +
    +_LEGACY_CTN_TYPE_NAMES = frozenset([
    + "IEC60870Server",
    + "IEC60870CommonAddress",
    + "IEC60870DataPoint",
    +])
    +
    +# Default instance prefix when adding from IDE (AddConfNode uses Type_0)
    +_SHORT_INSTANCE_PREFIX = {
    + CTN_COMMON_ADDRESS: "CA",
    + CTN_DATA_POINT: "DP",
    +}
    +
    +
    +def _iec60870_ctn_add_child(self, CTNName, CTNType, IEC_Channel=0):
    + """Support legacy folder types; use CA_0 / DP_0 default instance names."""
    + if CTNType not in _LEGACY_CTN_TYPE_NAMES:
    + short = _SHORT_INSTANCE_PREFIX.get(CTNType)
    + if short and CTNName == "%s_0" % CTNType:
    + CTNName = "%s_0" % short
    + saved = self.CTNChildrenTypes
    + try:
    + self.CTNChildrenTypes = saved + _LEGACY_CTN_CHILDREN_TYPES
    + return ConfigTreeNode.CTNAddChild(self, CTNName, CTNType, IEC_Channel)
    + finally:
    + self.CTNChildrenTypes = saved
    +
    +
    +def _iec60870_get_contextual_menu_items(confnode):
    + app_frame = confnode.GetCTRoot().AppFrame
    + items = []
    + for ctn_type, cls, helpstr in confnode.CTNChildrenTypes:
    + if ctn_type in _LEGACY_CTN_TYPE_NAMES:
    + continue
    + if getattr(cls, "CTNMaxCount", None) and confnode.Children.get(ctn_type):
    + if len(confnode.Children[ctn_type]) >= cls.CTNMaxCount:
    + continue
    + label = _("Add %s") % (helpstr or ctn_type)
    + items.append((
    + label,
    + helpstr or "",
    + app_frame.GetAddConfNodeFunction(ctn_type, confnode)))
    + return items
    +
    +
    +def _is_server_plug(node):
    + return getattr(node, "PlugType", None) in (CTN_SERVER, "IEC60870Server")
    +
    +
    +def _is_common_address_plug(node):
    + return getattr(node, "PlugType", None) in (
    + CTN_COMMON_ADDRESS, "IEC60870CommonAddress")
    +
    +
    +def _is_data_point_ctn_type(ctn_type):
    + return ctn_type in (CTN_DATA_POINT, "IEC60870DataPoint")
    +
    def _srv_attr_map(server_plug):
    for element in server_plug.GetParamsAttributes():
    @@ -696,7 +757,7 @@
    asdu_type_str = self.GetParamsAttributes()[0]["children"][0]["value"]
    common_label = ""
    parent = getattr(self, "CTNParent", None)
    - if parent is not None and getattr(parent, "PlugType", None) == "IEC60870CommonAddress":
    + if parent is not None and _is_common_address_plug(parent):
    common_label = str(parent.IEC60870CommonAddressNode.getCommon_Address())
    type_id, datatype, datasize, direction, size_code, desc = \
    @@ -708,7 +769,7 @@
    for offset in range(ioa, ioa + count):
    ioa_name = desc + " IOA " + str(offset)
    if common_label:
    - ioa_name = "CA " + common_label + " — " + ioa_name
    + ioa_name = "CA %s - %s" % (common_label, ioa_name)
    entries.append({
    "name": ioa_name,
    "type": loc_type,
    @@ -757,8 +818,13 @@
    </xsd:schema>
    """)
    - CTNChildrenTypes = [("IEC60870DataPoint", _DataPointPlug, "Data Point")]
    - PlugType = "IEC60870CommonAddress"
    + CTNChildrenTypes = [(CTN_DATA_POINT, _DataPointPlug, "DataPoint")]
    + PlugType = CTN_COMMON_ADDRESS
    +
    + CTNAddChild = _iec60870_ctn_add_child
    +
    + def GetContextualMenuItems(self):
    + return _iec60870_get_contextual_menu_items(self)
    def __init__(self):
    loc_str = ".".join(map(str, self.GetCurrentLocation()))
    @@ -812,9 +878,13 @@
    </xsd:schema>
    """)
    - CTNChildrenTypes = [
    - ("IEC60870CommonAddress", _CommonAddressPlug, "Common Address")]
    - PlugType = "IEC60870Server"
    + CTNChildrenTypes = [(CTN_COMMON_ADDRESS, _CommonAddressPlug, "CommonAddress")]
    + PlugType = CTN_SERVER
    +
    + CTNAddChild = _iec60870_ctn_add_child
    +
    + def GetContextualMenuItems(self):
    + return _iec60870_get_contextual_menu_items(self)
    def __init__(self):
    loc_str = ".".join(map(str, self.GetCurrentLocation()))
    @@ -973,6 +1043,14 @@
    return [], "", False
    +# Legacy confnode folder types (load only; hidden from Add menu)
    +_LEGACY_CTN_CHILDREN_TYPES = [
    + ("IEC60870Server", _IEC60870ServerPlug, ""),
    + ("IEC60870CommonAddress", _CommonAddressPlug, ""),
    + ("IEC60870DataPoint", _DataPointPlug, ""),
    +]
    +
    +
    #
    # R O O T C L A S S
    #
    @@ -999,10 +1077,15 @@
    </xsd:schema>
    """
    CTNChildrenTypes = [
    - ("IEC60870Server", _IEC60870ServerPlug, "IEC 60870-5-104 Server"),
    + (CTN_SERVER, _IEC60870ServerPlug, "Server"),
    # ("IEC60870Client", _IEC60870ClientPlug, "IEC 60870-5-104 Client"),
    ]
    + CTNAddChild = _iec60870_ctn_add_child
    +
    + def GetContextualMenuItems(self):
    + return _iec60870_get_contextual_menu_items(self)
    +
    def GetNodeCount(self):
    max_remote_clients = self.GetParamsAttributes()[0]["children"][0]["value"]
    total = (max_remote_clients, 0)
    @@ -1014,7 +1097,7 @@
    def GetIPServerPortNumbers(self):
    port_numbers = []
    for child in self.IECSortedChildren():
    - if child.CTNType == "IEC60870Server":
    + if child.CTNType in (CTN_SERVER, "IEC60870Server"):
    port_numbers.extend(child.GetIPServerPortNumbers())
    return port_numbers
    @@ -1068,8 +1151,7 @@
    self.FatalError(error_message)
    loc_prefix = "_".join(map(str, self.GetCurrentLocation()))
    - servers = [ch for ch in self.IECSortedChildren() if getattr(
    - ch, "PlugType", None) == "IEC60870Server"]
    + servers = [ch for ch in self.IECSortedChildren() if _is_server_plug(ch)]
    if not servers:
    return [], "", False
    @@ -1140,7 +1222,8 @@
    smap = _srv_attr_map(srv)
    ca_nodes = srv.IECSortedChildren()
    legacy_dps = [
    - c for c in ca_nodes if getattr(c, "CTNType", None) == "IEC60870DataPoint"]
    + c for c in ca_nodes if _is_data_point_ctn_type(
    + getattr(c, "CTNType", None))]
    if legacy_dps:
    self.FatalError(
    _("Error: IEC60870 server %%{a1}.x has data points attached "