cim2pp: extracted getting default classes, added generic setting data…

…types from CGMES XMI schema

CHANGELOG.rst

@@ -34,6 +34,7 @@ Change Log
 - [ADDED] option to use a second tap changer for the trafo element
 - [CHANGED] parameters of function merge_internal_net_and_equivalent_external_net()
 - [FIXED] :code:`convert_format.py`: update the attributes of the characteristic objects to match the new characteristic
+- [CHANGED] cim2pp: extracted getting default classes, added generic setting datatypes from CGMES XMI schema
 
 
 [2.13.1] - 2023-05-12

pandapower/converter/cim/cim2pp/from_cim.py

@@ -9,6 +9,7 @@
 from . import build_pp_net
 from .. import cim_classes
 from .. import interfaces
+from . import converter_classes as std_converter_classes
 
 logger = logging.getLogger('cim.cim2pp.from_cim')
 
@@ -38,8 +39,30 @@ def from_cim_dict(cim_parser: cim_classes.CimParser, log_debug=False, convert_li
     :param custom_converter_classes: Dict to inject classes for different functionality. Optional, default: None
     :return: The pandapower net.
     """
-    from . import converter_classes as std_converter_classes
-    converter_classes: dict = {
+    converter_classes = get_converter_classes()
+    if custom_converter_classes is not None:
+        for key in custom_converter_classes:
+            converter_classes[key] = custom_converter_classes.get(key)
+
+    # repair the input CIM data
+    if repair_cim is not None and repair_cim_class is not None:
+        repair_cim = repair_cim_class().deserialize(repair_cim, report_container=cim_parser.get_report_container())
+        repair_cim.repair(cim_parser.get_cim_dict(), report_container=cim_parser.get_report_container())
+
+    cim_converter = build_pp_net.CimConverter(cim_parser=cim_parser, converter_classes=converter_classes, **kwargs)
+    pp_net = cim_converter.convert_to_pp(convert_line_to_switch=convert_line_to_switch, line_r_limit=line_r_limit,
+                                         line_x_limit=line_x_limit, log_debug=log_debug, **kwargs)
+
+    # repair the output pandapower network
+    if repair_pp is not None and repair_pp_class is not None:
+        repair_pp = repair_pp_class().deserialize(repair_pp, report_container=cim_parser.get_report_container())
+        repair_pp.repair(pp_net, report_container=cim_parser.get_report_container())
+
+    return pp_net
+
+
+def get_converter_classes():
+    converter_classes: Dict[str,classmethod] = {
         'connectivityNodesCim16': std_converter_classes.connectivitynodes.connectivityNodesCim16,
         'externalNetworkInjectionsCim16': std_converter_classes.externalnetworks.externalNetworkInjectionsCim16,
         'acLineSegmentsCim16': std_converter_classes.lines.acLineSegmentsCim16,
@@ -63,25 +86,7 @@ def from_cim_dict(cim_parser: cim_classes.CimParser, log_debug=False, convert_li
         'geoCoordinatesFromGLCim16': std_converter_classes.coordinates.geoCoordinatesFromGLCim16,
         'coordinatesFromDLCim16': std_converter_classes.coordinates.coordinatesFromDLCim16,
     }
-    if custom_converter_classes is not None:
-        for key in custom_converter_classes:
-            converter_classes[key] = custom_converter_classes.get(key)
-
-    # repair the input CIM data
-    if repair_cim is not None and repair_cim_class is not None:
-        repair_cim = repair_cim_class().deserialize(repair_cim, report_container=cim_parser.get_report_container())
-        repair_cim.repair(cim_parser.get_cim_dict(), report_container=cim_parser.get_report_container())
-
-    cim_converter = build_pp_net.CimConverter(cim_parser=cim_parser, converter_classes=converter_classes, **kwargs)
-    pp_net = cim_converter.convert_to_pp(convert_line_to_switch=convert_line_to_switch, line_r_limit=line_r_limit,
-                                         line_x_limit=line_x_limit, log_debug=log_debug, **kwargs)
-
-    # repair the output pandapower network
-    if repair_pp is not None and repair_pp_class is not None:
-        repair_pp = repair_pp_class().deserialize(repair_pp, report_container=cim_parser.get_report_container())
-        repair_pp.repair(pp_net, report_container=cim_parser.get_report_container())
-
-    return pp_net
+    return converter_classes
 
 
 def from_cim(file_list: List[str] = None, encoding: str = 'utf-8', convert_line_to_switch: bool = False,

pandapower/converter/cim/cim_classes.py

@@ -14,6 +14,7 @@
 import xml.etree.ElementTree
 import xml.etree.cElementTree as xmlET
 from .other_classes import ReportContainer, Report, LogLevel, ReportCode
+from .cim_tools import get_cim16_schema
 
 
 class CimParser:
@@ -76,60 +77,39 @@ def set_cim_data_types(self) -> CimParser:
         float_type = float
         int_type = pd.Int64Dtype()
         bool_type = pd.BooleanDtype()
-        data_types = dict({'nominalVoltage': float_type, 'r': float_type, 'x': float_type, 'r2': float_type,
-                           'x2': float_type, 'bch': float_type, 'v': float_type, 'p': float_type, 'q': float_type,
-                           'ratedUdc': float_type, 'targetUpcc': float_type, 'droop': float_type,
-                           'droopCompensation': float_type, 'qShare': float_type, 'targetUdc': float_type,
-                           'targetPpcc': float_type, 'targetQpcc': float_type,
-                           'minP': float_type, 'maxP': float_type, 'minQ': float_type, 'maxQ': float_type,
-                           'ratedPowerFactor': float_type, 'minOperatingP': float_type, 'maxOperatingP': float_type,
-                           'nominalP': float_type, 'initialP': float_type, 'baseS': float_type,
-                           'maxInitialSymShCCurrent': float_type, 'minInitialSymShCCurrent': float_type,
-                           'maxR1ToX1Ratio': float_type, 'minR1ToX1Ratio': float_type, 'targetValue': float_type,
-                           'activePower': float_type, 'reactivePower': float_type, 'b': float_type,
-                           'ratedS': float_type, 'ratedU': float_type, 'stepVoltageIncrement': float_type,
-                           'stepPhaseShiftIncrement': float_type, 'step': float_type, 'length': float_type,
-                           'g': float_type, 'ratio': float_type, 'regulationTarget': float_type, 'value': float_type,
-                           'sensorAccuracy': float_type, 'neutralU': float_type, 'voltageStepIncrement': float_type,
-                           'xPosition': float_type, 'yPosition': float_type, 'r21': float_type, 'x21': float_type,
-                           'nomU': float_type, 'gPerSection': float_type, 'bPerSection': float_type,
-                           'angle': float_type, 'r0': float_type, 'x0': float_type, 'b0ch': float_type,
-                           'gch': float_type, 'g0ch': float_type, 'shortCircuitEndTemperature': float_type,
-                           'targetDeadband': float_type, 'netInterchange': float_type, 'iaIrRatio': float_type,
-                           'voltageRegulationRange': float_type, 'rxLockedRotorRatio': float_type,
-                           'maxR0ToX0Ratio': float_type, 'maxZ0ToZ1Ratio': float_type, 'xground': float_type,
-                           'efficiency': float_type, 'ratedMechanicalPower': float_type, 'voltageSetPoint': float_type,
-                           'zeroR12': float_type, 'zeroR21': float_type, 'zeroX12': float_type, 'zeroX21': float_type,
-                           'voltageAngle': float_type, 'voltageMagnitude': float_type,
-                           'controlEnabled': bool_type, 'connected': bool_type, 'open': bool_type,
-                           'regulationStatus': bool_type, 'positiveFlowIn': bool_type,
-                           'isPartOfGeneratorUnit': bool_type, 'ltcFlag': bool_type, 'discrete': bool_type,
-                           'enabled': bool_type, 'grounded': bool_type,
-                           'referencePriority': int_type, 'sectionNumber': int_type, 'sections': int_type,
-                           'maximumSections': int_type, 'endNumber': int_type, 'sequenceNumber': int_type,
-                           'neutralStep': int_type, 'lowStep': int_type, 'highStep': int_type, 'normalStep': int_type,
-                           'phaseAngleClock': int_type, 'position': int_type})
+        data_types_map = dict({'Float': float_type, 'Integer': int_type, 'Boolean': bool_type})
+        cim_16_schema = get_cim16_schema()
         for profile in self.cim.keys():
             for cim_element_type, item in self.cim[profile].items():
                 for col in item.columns:
-                    if col in data_types.keys():
+                    # skip elements which are not available in the schema like FullModel
+                    if cim_element_type not in cim_16_schema[profile]:
+                        self.logger.debug("Skipping CIM element type %s from profile %s." % (cim_element_type, profile))
+                        continue
+                    if col in cim_16_schema[profile][cim_element_type]['fields'].keys() and \
+                            'data_type_prim' in cim_16_schema[profile][cim_element_type]['fields'][col].keys():
+                        data_type_col_str = cim_16_schema[profile][cim_element_type]['fields'][col]['data_type_prim']
+                        if data_type_col_str in data_types_map.keys():
+                            data_type_col = data_types_map[data_type_col_str]
+                        else:
+                            continue
                         self.logger.debug("Setting data type of %s from CIM element %s as type %s" %
-                                          (col, cim_element_type, data_types[col]))
-                        if col in default_values.keys():
+                                          (col, cim_element_type, data_type_col_str))
+                        if col in default_values.keys():  # todo deprecated due to repair function?
                             self.cim[profile][cim_element_type][col].fillna(value=default_values[col], inplace=True)
-                        if data_types[col] == bool_type:
+                        if data_type_col == bool_type:
                             self.cim[profile][cim_element_type][col] = \
                                 self.cim[profile][cim_element_type][col].map(to_bool)
                         try:
                             # special fix for integers:
-                            if data_types[col] == int_type:
+                            if data_type_col == int_type:
                                 self.cim[profile][cim_element_type][col] = \
                                     self.cim[profile][cim_element_type][col].astype(float_type)
                             self.cim[profile][cim_element_type][col] = \
-                                self.cim[profile][cim_element_type][col].astype(data_types[col])
+                                self.cim[profile][cim_element_type][col].astype(data_type_col)
                         except Exception as e:
                             self.logger.warning("Couldn't set the datatype to %s for field %s at CIM type %s in "
-                                                "profile %s!" % (data_types[col], col, cim_element_type, profile))
+                                                "profile %s!" % (data_type_col_str, col, cim_element_type, profile))
                             self.logger.warning("This may be harmless if the data is not need by the converter. "
                                                 "Message: %s" % e)
         self.logger.info("Finished setting the cim data types.")

pandapower/converter/cim/cim_tools.py

@@ -1,13 +1,13 @@
 # -*- coding: utf-8 -*-
-
 # Copyright (c) 2016-2023 by University of Kassel and Fraunhofer Institute for Energy Economics
 # and Energy System Technology (IEE), Kassel. All rights reserved.
 import logging
+import os
+import json
 from typing import Dict, List
 import numpy as np
 from pandapower.auxiliary import pandapowerNet
 import pandas as pd
-from . import cim_classes
 
 logger = logging.getLogger(__name__)
 
@@ -129,5 +129,19 @@ def extend_pp_net_cim(net: pandapowerNet, override: bool = True) -> pandapowerNe
     return net
 
 
-def get_cim_data_structure() -> Dict[str, Dict[str, pd.DataFrame]]:
-    return cim_classes.CimParser().get_cim_data_structure()
+def get_cim16_schema() -> Dict[str, Dict[str, Dict[str, str or Dict[str, Dict[str, str]]]]]:
+    """
+    Parses the CIM 16 schema from the CIM 16 RDF schema files for the serializer from the CIM data structure used by
+    the cim2pp and pp2cim converters.
+    :return: The CIM 16 schema as dictionary.
+    """
+    path_with_serialized_schemas = os.path.dirname(__file__) + os.sep + 'serialized_schemas'
+    if not os.path.isdir(path_with_serialized_schemas):
+        os.mkdir(path_with_serialized_schemas)
+    for one_file in os.listdir(path_with_serialized_schemas):
+        if one_file.lower().endswith('_schema.json'):
+            path_to_schema = path_with_serialized_schemas + os.sep + one_file
+            logger.info("Parsing the schema from CIM 16 from disk: %s" % path_to_schema)
+            with open(path_to_schema, encoding='UTF-8', mode='r') as f:
+                cim16_schema = json.load(f)
+            return cim16_schema