Commit 334e084e authored by Philipp Reuber's avatar Philipp Reuber
Browse files

add documentation for utils, add utils examples, use pathlib for paths

parent c3309d6e
Pipeline #257215 passed with stages
in 1 minute and 31 seconds
......@@ -3,3 +3,5 @@ from cimpy.cimimport import cim_import
import cimpy.utils
from cimpy.examples import import_example
from cimpy.examples import export_example
from cimpy.examples import addExternalNetworkInjection_example
from cimpy.examples import convertToBusBranch_example
......@@ -72,7 +72,7 @@ def _get_reference_uuid(attr_dict, version, topology, mRID, urls):
UUID = '%' + _search_mRID(attr_dict[key], topology)
if UUID == '%':
logger.warning('Object of type {} not found as reference for object with UUID {}.'.format(
elem.__class__.__name__, mRID))
attr_dict[key].__class__.__name__, mRID))
else:
UUID = '%' + attr_dict[key].mRID
attributes['value'] = UUID
......
......@@ -10,4 +10,16 @@ def import_example():
def export_example():
base = Path(__file__).resolve().parent.parent
example = base / 'examples' / 'quickstart' / 'exportCIGREMV.py'
exec(open(example).read())
\ No newline at end of file
exec(open(example).read())
def convertToBusBranch_example():
base = Path(__file__).resolve().parent.parent
example = base / 'examples' / 'quickstart' / 'convertToBusBranch.py'
exec(open(example).read())
def addExternalNetworkInjection_example():
base = Path(__file__).resolve().parent.parent
example = base / 'examples' / 'quickstart' / 'addExternalNetworkInjection.py'
exec(open(example).read())
import importlib
import cimpy
def node_breaker_to_bus_branch(res):
def node_breaker_to_bus_branch(import_result):
res = import_result['topology']
breaker_list = []
terminals_list = []
operational_limit_set_list = []
......@@ -37,72 +39,40 @@ def node_breaker_to_bus_branch(res):
del_terminals_list = []
for terminal in terminals_list:
cond_eq = res[terminal].ConductingEquipment
if isinstance(cond_eq, list):
if len(cond_eq) == 1:
if cond_eq[0].mRID in open_breakers:
del_terminals_list.append(terminal)
else:
res[terminal].ConnectivityNode = None
else:
print("List longer than 1")
if cond_eq.mRID in open_breakers:
del_terminals_list.append(terminal)
else:
print("No List!")
res[terminal].ConnectivityNode = None
# check for OperationalLimitSet with references to deleted Terminals
del_operationallimitset = []
for operational_limit in operational_limit_set_list:
keep_op_limit = []
for item in res[operational_limit].Terminal:
if item.mRID not in del_terminals_list:
keep_op_limit.append(item)
if len(keep_op_limit) == 0:
if res[operational_limit].Terminal.mRID in del_terminals_list:
del_operationallimitset.append(operational_limit)
else:
res[operational_limit].Terminal = keep_op_limit
del_voltage_limit = []
for voltage in voltage_limit_list:
keep_voltage = []
for item in res[voltage].OperationalLimitSet:
if item.mRID not in del_operationallimitset:
keep_voltage.append(item)
if len(keep_voltage) == 0:
if res[voltage].OperationalLimitSet.mRID in del_operationallimitset:
del_voltage_limit.append(voltage)
else:
res[voltage].OperationalLimitSet = keep_voltage
del_diagram_object = []
for diagram_object in diagram_objects_list:
keep_diagram = []
if res[diagram_object].IdentifiedObject is None:
continue
for item in res[diagram_object].IdentifiedObject:
if 'ConnectivityNode' in str(type(item)):
if item.TopologicalNode is not None:
if isinstance(item.TopologicalNode, list):
for elem in item.TopologicalNode:
keep_diagram.append(elem)
else:
keep_diagram.append(item.TopologicalNode)
elif item.mRID not in (open_breakers + del_terminals_list):
keep_diagram.append(item)
if len(keep_diagram) == 0:
if 'ConnectivityNode' in str(type(res[diagram_object].IdentifiedObject)):
if res[diagram_object].IdentifiedObject.TopologicalNode is not None:
keep_diagram.append(res[diagram_object].IdentifiedObject.TopologicalNode)
elif res[diagram_object].IdentifiedObject.mRID in (open_breakers + del_terminals_list):
del_diagram_object.append(diagram_object)
else:
res[diagram_object].IdentifiedObject = keep_diagram
del_diagram_object_points = []
for diagram_point in diagram_object_points_list:
keep_point = []
if res[diagram_point].DiagramObject is None:
continue
for item in res[diagram_point].DiagramObject:
if item.mRID not in del_diagram_object:
keep_point.append(item)
if len(keep_point) == 0:
if res[diagram_point].DiagramObject.mRID in del_diagram_object:
del_diagram_object_points.append(diagram_point)
else:
res[diagram_point].DiagramObject = keep_point
del_list = open_breakers + del_diagram_object_points + del_diagram_object + del_voltage_limit + \
del_operationallimitset + del_terminals_list + connect_nodes
......@@ -110,10 +80,13 @@ def node_breaker_to_bus_branch(res):
for key in del_list:
del res[key]
return res
import_result['topology'] = res
return import_result
def add_external_network_injection(res, version, mRID, voltage_set_point):
def add_external_network_injection(import_result, version, mRID, voltage_set_point):
res = import_result['topology']
TopologicalNode = ''
if mRID in res:
if 'TopologicalNode' in str(type(res[mRID])):
......@@ -158,4 +131,6 @@ def add_external_network_injection(res, version, mRID, voltage_set_point):
else:
print('No Terminal with mRID ', mRID, ' found in object list!')
return res
import_result['topology'] = res
return import_result
This diff is collapsed.
......@@ -37,4 +37,36 @@ This example uses the import result of the import example and exports them back
.. toctree::
:glob:
Export
\ No newline at end of file
Export
Utils
-----
Currently there are two functions in utils.
.. toctree::
:glob:
utils
Add External Network injection
______________________________
The function add_external_network_injection adds an external network injection to an existing node. An example for this function can be run with:
.. code-block:: python
import cimpy
cimpy.addExternalNetworkInjection_example()
Convert Node-Breaker to Bus-Branch
__________________________________
This function converts grid data in Node-Breaker topology into Bus-Branch topology.
.. code-block:: python
import cimpy
cimpy.convertToBusBranch_example()
......@@ -7,6 +7,8 @@ The processing of grid data is based on CIM compatible Python classes. The codeb
The focus of CIMpy is on the support of the Common Grid Model Exchange Standard (CGMES) specified by the European Network of Transmission System Operators for Electricity (ENTSO-E). However, the CIMpy package can readily support further as well as new CIM versions if required.
.. image:: CIMpy.svg
Installation
-------------
For the installation of CIMpy take a look at the `installation instructions <https://acs.pages.rwth-aachen.de/public/cim/cimpy/Install.html>`__.
......
Utils
"""""
This part of the package provides two basic functions to modify the imported CIM data.
Convert Node-Breaker to Bus-Branch
----------------------------------
This function converts as the name suggests a Node-Breaker topology into a Bus-Branch topology.
First we have to import the grid data. For more information see the documentation of the import function.
.. code-block:: python
import logging
import cimpy
from pathlib import Path
logging.basicConfig(filename='importCIGREMV.log', level=logging.INFO, filemode='w')
example = Path('.').resolve()
sample_folder = example / 'examples' / 'sampledata' / 'Sample_Grid_Switches' / 'Bus-Branch'
sample_files = sample_folder.glob('*.xml')
xml_files = []
for file in sample_folder.glob('*.xml'):
xml_files.append(str(file.absolute()))
import_result = cimpy.cim_import(xml_files, "cgmes_v2_4_15")
Now we can convert the imported topology into a Bus-Branch topology.
.. code-block:: python
import_result = cimpy.utils.node_breaker_to_bus_branch(import_result)
The converted grid data is stored in import_result['topology']
Add external network injection
------------------------------
This function adds an external network injection to the imported grid data. First we have to import some grid data. Here we use the same data as the import example. The imported data is stored in import_example.
To add the external network injection we need the mRID of either a Topological Node or a Connectivity Node.
.. code-block:: python
import_result = cimpy.cim_import(xml_files, "cgmes_v2_4_15")
import_result = cimpy.utils.add_external_network_injection(import_result, "cgmes_v2_4_15", "N1", 20.0)
Here the the external network injection is added to the node with the mRID "N1" with a voltage set point of 20.0. For this function the CIM version must also be given to the function.
After the injection is added the modified grid data can be exported with cim_export.
.. code-block:: python
activeProfileList = ['DI', 'EQ', 'SV', 'TP']
cimpy.cim_export(import_result, 'ExternalInjection', 'cgmes_v2_4_15', activeProfileList)
import logging
import cimpy
from pathlib import Path
logging.basicConfig(filename='importCIGREMV.log', level=logging.INFO, filemode='w')
example = Path(__file__).resolve().parent.parent
# called as cimpy.examples.addExternalNetworkInjection() or file run from quickstart directory?
if 'examples.py' in str(__file__):
sample_folder = example / 'examples' / 'sampledata' / 'CIGRE_MV'
else:
sample_folder = example / 'sampledata' / 'CIGRE_MV'
sample_files = sample_folder.glob('*.xml')
xml_files = []
for file in sample_folder.glob('*.xml'):
xml_files.append(str(file.absolute()))
import_result = cimpy.cim_import(xml_files, "cgmes_v2_4_15")
import_result = cimpy.utils.add_external_network_injection(import_result, "cgmes_v2_4_15", "N1", 20.0)
activeProfileList = ['DI', 'EQ', 'SV', 'TP']
cimpy.cim_export(import_result, 'ExternalInjection', 'cgmes_v2_4_15', activeProfileList)
import logging
import os
import cimpy
from pathlib import Path
logging.basicConfig(filename='Convert_to_Bus_Branch.log', level=logging.INFO, filemode='w')
xml_files = [r"..\sampledata\Sample_Grid_Switches\Node-Breaker\20191030T0924Z_XX_YYY_DL_.xml",
r"..\sampledata\Sample_Grid_Switches\Node-Breaker\20191030T0924Z_XX_YYY_GL_.xml",
r"..\sampledata\Sample_Grid_Switches\Node-Breaker\20191030T0924Z_XX_YYY_SSH_.xml",
r"..\sampledata\Sample_Grid_Switches\Node-Breaker\20191030T0924Z_XX_YYY_SV_.xml",
r"..\sampledata\Sample_Grid_Switches\Node-Breaker\20191030T0924Z_XX_YYY_TP_.xml",
r"..\sampledata\Sample_Grid_Switches\Node-Breaker\20191030T0924Z_YYY_EQ_.xml",]
example = Path(__file__).resolve().parent.parent
# called as cimpy.examples.convertBusBranch() or file run from quickstart directory?
if 'examples.py' in str(__file__):
sample_folder = example / 'examples' / 'sampledata' / 'Sample_Grid_Switches' / 'Node-Breaker'
else:
sample_folder = example / 'sampledata' / 'Sample_Grid_Switches' / 'Node-Breaker'
sample_files = sample_folder.glob('*.xml')
xml_files = []
for file in sample_folder.glob('*.xml'):
xml_files.append(str(file.absolute()))
xml_files_abs = []
for file in xml_files:
xml_files_abs.append(os.path.abspath(file))
import_result = cimpy.cim_import(xml_files, "cgmes_v2_4_15")
# res = cimpy.cimread(xml_files)
res, namespaces = cimpy.cim_import(xml_files_abs, "cgmes_v2_4_15")
import_result = cimpy.utils.node_breaker_to_bus_branch(import_result)
bus_branch = cimpy.utils.node_breaker_to_bus_branch(res)
activeProfileList = ['DI', 'EQ', 'SV', 'TP']
# dicts = cimpy.get_class_attributes_dict(res)
cimpy.cim_export(bus_branch, namespaces, 'Bus_Branch_Converted', 'cgmes_v2_4_15')
\ No newline at end of file
cimpy.cim_export(import_result, 'Bus_Branch_Converted', 'cgmes_v2_4_15', activeProfileList)
......@@ -3,8 +3,13 @@ from pathlib import Path
import cimpy
logging.basicConfig(filename='exportCIGREMV.log', level=logging.INFO, filemode='w')
example = Path('.').resolve()
sample_folder = example / 'examples' / 'sampledata' / 'CIGRE_MV'
example = Path(__file__).resolve().parent.parent
# called as cimpy.examples.import_example() or file run from quickstart directory?
if 'examples.py' in str(__file__):
sample_folder = example / 'examples' / 'sampledata' / 'CIGRE_MV'
else:
sample_folder = example / 'sampledata' / 'CIGRE_MV'
sample_files = sample_folder.glob('*.xml')
......
......@@ -4,8 +4,13 @@ from pathlib import Path
logging.basicConfig(filename='importCIGREMV.log', level=logging.INFO, filemode='w')
example = Path('.').resolve()
sample_folder = example / 'examples' / 'sampledata' / 'CIGRE_MV'
example = Path(__file__).resolve().parent.parent
# called as cimpy.examples.import_example() or file run from quickstart directory?
if 'examples.py' in str(__file__):
sample_folder = example / 'examples' / 'sampledata' / 'CIGRE_MV'
else:
sample_folder = example / 'sampledata' / 'CIGRE_MV'
sample_files = sample_folder.glob('*.xml')
......
......@@ -3,6 +3,7 @@ import cimpy
import xmltodict
import os
import pytest_check as check
from pathlib import Path
logging.basicConfig(filename='Test_export_with_exported_files.log', level=logging.INFO, filemode='w')
......@@ -16,18 +17,16 @@ short_profile_name = {
"Topology": "TP"
}
example_path = os.path.join('..',
os.path.join('examples',
os.path.join('sampledata', 'CIGRE_MV')))
tests = Path('.').resolve().parent
example_path = tests / 'examples' / 'sampledata' / 'CIGRE_MV'
# This test tests the export functionality of this package by first importing the CIGRE_MV_Rudion_With_LoadFlow_Results
# example and exporting them. The exported files are compared with previously exported files which were checked manually
def test_export_with_exported_files():
import_files = [os.path.join(example_path, 'Rootnet_FULL_NE_24J13h_DI.xml'),
os.path.join(example_path, 'Rootnet_FULL_NE_24J13h_EQ.xml'),
os.path.join(example_path, 'Rootnet_FULL_NE_24J13h_SV.xml'),
os.path.join(example_path, 'Rootnet_FULL_NE_24J13h_TP.xml'), ]
import_files = []
for file in example_path.glob('*.xml'):
import_files.append(str(file.absolute()))
activeProfileList = ['DI', 'EQ', 'SV', 'TP']
......
......@@ -2,6 +2,7 @@ import logging
import cimpy
import xmltodict
import os
from pathlib import Path
import pytest_check as check
logging.basicConfig(filename='Test_export_with_imported_files.log', level=logging.INFO, filemode='w')
......@@ -16,19 +17,17 @@ short_profile_name = {
"Topology": "TP"
}
example_path = os.path.join('..',
os.path.join('examples',
os.path.join('sampledata', 'CIGRE_MV')))
tests = Path('.').resolve().parent
example_path = tests / 'examples' / 'sampledata' / 'CIGRE_MV'
# This test function tests the export functionality by comparing files before the import and export procedure with the
# exported files. Since cyclic attributes are not resolved in this package, the imported files only need to be a subset
# of the exported files.
def test_export_with_imported_files():
import_files = [os.path.join(example_path, 'Rootnet_FULL_NE_24J13h_DI.xml'),
os.path.join(example_path, 'Rootnet_FULL_NE_24J13h_EQ.xml'),
os.path.join(example_path, 'Rootnet_FULL_NE_24J13h_SV.xml'),
os.path.join(example_path, 'Rootnet_FULL_NE_24J13h_TP.xml'), ]
import_files = []
for file in example_path.glob('*.xml'):
import_files.append(str(file.absolute()))
activeProfileList = ['DI', 'EQ', 'SV', 'TP']
......
import logging
import cimpy
import os
import pytest_check as check
import pickle
from pathlib import Path
logging.basicConfig(filename='Test_import.log', level=logging.INFO, filemode='w')
......@@ -16,18 +16,16 @@ short_profile_name = {
"Topology": "TP"
}
example_path = os.path.join('..',
os.path.join('examples',
os.path.join('sampledata', 'CIGRE_MV')))
tests = Path('.').resolve().parent
example_path = tests / 'examples' / 'sampledata' / 'CIGRE_MV'
# This function tests the import functionality by importing files and comparing them to previously imported and pickled
# files.
def test_import():
test_files = [os.path.join(example_path, 'Rootnet_FULL_NE_24J13h_DI.xml'),
os.path.join(example_path, 'Rootnet_FULL_NE_24J13h_EQ.xml'),
os.path.join(example_path, 'Rootnet_FULL_NE_24J13h_SV.xml'),
os.path.join(example_path, 'Rootnet_FULL_NE_24J13h_TP.xml'), ]
test_files = []
for file in example_path.glob('*.xml'):
test_files.append(str(file.absolute()))
imported_result = cimpy.cim_import(test_files, 'cgmes_v2_4_15')
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment