Source code for dave_core.converter.create_mynts

# Copyright (c) 2022-2024 by Fraunhofer Institute for Energy Economics and Energy System Technology (IEE)
# Kassel and individual contributors (see AUTHORS file for details). All rights reserved.
# Use of this source code is governed by a BSD-style license that can be found in the LICENSE file.

from pathlib import Path

from dave_core.converter.converter import Converter
from dave_core.converter.converter import Strategy
from dave_core.converter.elements import Elements
from dave_core.progressbar import create_tqdm

# dictionaries for Mynts text properties and numeric properties;  # !!! todo complete list
# used to convert dave names to the corresponding Mynts properties

MyntsTextProps = {
    "dave_name": "name",
    "lat": "Y",
    "long": "X",
    "from_junction": "node1",
    "to_junction": "node2",
    "country_code": "Des",
    "scigrid_name": "Bez",
    "junction": "node",
}

MyntsNumProps = {
    "diameter_mm": "D",
    "length_km": "L",
    "height_m": "H",
    "roughness": "k",
    "max_pressure_bar": "POMAX",
}

MyntsReqNodeProps = [
    "H",
    "X",
    "Y",
    "Web",
    "KTyp",
    "nVNB",
    "aFNB",
    "EIC",
    "MG",
    "Zuornd",
    "Des",
    "GeoLat",
    "GeoLong",
    "AnLNum",
    "LNum",
    "Schemaplan",
    "UmstSchr",
    "UmstTag",
    "NEPID",
    "NEPID18",
    "UmstJ",
    "UmstBer",
    "Zone",
    "Teilnetz",
    "Messort",
]


MyntsReqPipeProps = [
    "node1",
    "node2",
    "L",
    "D",
    "k",
    "htc",
    "NurPlan",
    "Bez",
    "Des",
    "LNum",
    "LName",
    "PNPipe",
    "Eig",
    "NEPID",
    "NEPID18",
    "UmstJ",
    "ModVar",
    "IJahr",
    "UmstBer",
    "UmstSchr",
    "UmstTag",
    "Update",
]

MyntsReqValveProps = [
    "node1",
    "node2",
    "pimin",
    "pomax",
    "L",
    "D",
    "NurPlan",
    "Bez",
    "Des",
    "LNum",
    "LName",
    "Autom",
    "Eig",
    "NEPID",
    "NEPID18",
    "UmstJ",
    "ModVar",
    "IJahr",
    "UmstSchr",
    "Update",
]

MyntsReqCompressorProps = [
    "D",
    "pimin",
    "pomax",
    "node1",
    "node2",
    "kind",
    "NurPlan",
    "Bez",
    "Des",
    "Autom",
    "Eig",
    "NEPID",
    "NEPID18",
    "UmstJ",
    "ModVar",
    "IJahr",
    "UmstSchr",
    "Update",
]

MyntsReqDriveProps = [
    "node",
    "kind",
    "NurPlan",
    "Bez",
    "Des",
    "Autom",
    "Eig",
    "NEPID",
    "NEPID18",
    "UmstJ",
    "ModVar",
    "IJahr",
    "UmstSchr",
    "Update",
]

MyntsReqProps = {
    "n": MyntsReqNodeProps,
    "p": MyntsReqPipeProps,
    "v": MyntsReqValveProps,
    "c": MyntsReqCompressorProps,
    "d": MyntsReqDriveProps,
}


# convert prop value to Mynts internal unit                # !!! todo complete list
def convertPropValue2Mynts(prop, value) -> str:
    #    if prop=="roughness":
    #        print("roughness ="+str(value))
    if value == "":
        return "-1"
    fvalue = float(value)
    if prop == "diameter_mm":
        fvalue = fvalue / 1.0e3
    elif prop == "length_km":
        fvalue = fvalue * 1.0e3
    # elif prop.endswith("bar"): ok
    return str(fvalue)


# get Mynts name of DaVe property; returns original name if not in Mynts name dict
def myntsProp(prop) -> str:
    if prop in MyntsTextProps:
        return MyntsTextProps[prop]
    elif prop in MyntsNumProps:
        return MyntsNumProps[prop]
    return prop


def daveProp(prop) -> str:
    # reversed MyntsProp dict lists
    daveProps = {y: x for x, y in MyntsTextProps.items()}
    daveNumProps = {y: x for x, y in MyntsNumProps.items()}
    if prop in daveProps:
        return daveProps[prop]
    if prop in daveNumProps:
        return daveNumProps[prop]
    return prop


class MyntsWriter:  # Output file strategy class for Mynts
    elements = Elements()
    # MyntsProps = {}

    def __init__(self, file=None):
        self.file = file

    def setFile(self, file):
        self.file = file

    # get the elements from Elements dict and write to geom file
    def writeGeom(self, elements):
        self.elements = elements
        element = elements.nextEle()
        while element is not None:
            self.writeGeomElement(element)
            element = elements.nextEle()

    # get the elements from Elements dict and write to scen file
    def writeScen(self, elements):
        self.elements = elements
        element = elements.nextEle()
        while element is not None:
            self.writeScenElement(element)
            element = elements.nextEle()

    def writeGeomHeader(self):
        line = '{"#MYNTS_GEOM":"Base topology file ' + self.file.name + '"\n'
        self.file.write(line)

    def writeScenHeader(self):
        line = '{"#MYNTS_SCEN":"Scenario ' + self.file.name + '"\n'
        self.file.write(line)

    def writeFooter(self):
        self.file.write("}\n")

    # write an element in geom format
    def writeGeomElement(self, element):
        line = ',"' + element.name + '":{"obj_type":"' + element.type + '"'
        #
        for prop in MyntsReqProps[element.type]:
            dave_prop = daveProp(prop)
            newValue = str(element.get(dave_prop))
            # write required text Props even without a value
            if element.get(dave_prop) is None:
                newValue = ""
            # convert numeric value and ignore if it has no value
            if dave_prop in MyntsNumProps:
                if convertPropValue2Mynts(dave_prop, newValue) == "-1":
                    continue
                newValue = convertPropValue2Mynts(dave_prop, newValue)
            if element.name.startswith("sink") and dave_prop == "lat":
                newValue = str(element.get(dave_prop) - 0.01)
            if element.name.startswith("source") and dave_prop == "lat":
                newValue = str(element.get(dave_prop) + 0.01)
            line = line + ', "' + prop + '":"' + newValue + '"'
        line = line + "}\n"
        self.file.write(line)
        if element.name.startswith("sink") or element.name.startswith(
            "source"
        ):
            line = ',"s_' + element.name + '":{"obj_type":"s"'
            line = line + ', "node1":"' + element.get("junction") + '"'
            line = line + ', "node2":"' + element.name + '"'
            line = line + "}\n"
            self.file.write(line)

    # write an element in scen format
    def writeScenElement(self, element):
        line = ',"' + element.name + '":{'
        #
        first_element = True
        for prop in element.props():
            newName = myntsProp(prop)
            newValue = str(element.get(prop))
            if (
                newName.casefold()
                in (prop.casefold() for prop in MyntsReqProps[element.type])
                or element.get(prop) is None
            ):
                continue
            if prop in MyntsNumProps:
                newValue = convertPropValue2Mynts(prop, newValue)
            if first_element:
                line = line + '"' + newName + '":"' + newValue + '"'
                first_element = False
                continue
            line = line + ', "' + newName + '":"' + newValue + '"'
        line = line + "}\n"
        self.file.write(line)


class DaVe2Mynts(Strategy):
    """
    class to convert dave data to Mynts output files, one for each format
    """

    def __init__(self, basefilepath):
        self.files = {}
        self.writer = None
        self.basefilepath = basefilepath
        self.openFiles(self.basefilepath)

    def __del__(self):
        for file in self.files.keys():
            self.files[file].close()

    # opens a file for each output format
    def openFiles(self, outfile):
        self.files["geom"] = Path(outfile + ".geom.jsn").open("w")
        self.files["scen"] = Path(outfile + ".scen.jsn").open("w")
        self.writer = MyntsWriter()

    # writes elements to geom and scen
    def execute(self, element_types) -> str:
        # write elements in geom file
        self.writer.setFile(self.files["geom"])
        self.writer.writeGeomHeader()
        for elements in element_types:
            self.writer.writeGeom(elements)
        self.writer.writeFooter()
        # write elements in scen file
        self.writer.setFile(self.files["scen"])
        self.writer.writeScenHeader()
        for elements in element_types:
            self.writer.writeScen(elements)
        self.writer.writeFooter()
        return "DaVe2Mynts"


[docs] def create_mynts(grid_data, output_folder, idx_ref="dave_name"): """ This function creates a network in MYNTS format based of an DAVE dataset INPUT: **grid_data** (attrdict) - calculated grid data from DAVE **output_folder** (str) - patht to the location where the results will be saved OPTIONAL: **idx_ref** (str, default='dave_name') - defines parameter which should use as referenz \ for setting the indices """ # set progress bar pbar = create_tqdm(desc="create mynts network") # seperate geocoordinates from geometry parameter into lat and long grid_data.hp_data.hp_junctions["long"] = ( grid_data.hp_data.hp_junctions.geometry.apply(lambda x: x.x) ) grid_data.hp_data.hp_junctions["lat"] = ( grid_data.hp_data.hp_junctions.geometry.apply(lambda x: x.y) ) # init data myntsconv = Converter( grid_data, basefilepath=output_folder ) # default file names # update progress pbar.update(50) print() # extract the data from DaVe nodes = Elements("n", myntsconv.nodedata) sinks = Elements("n", myntsconv.sinkdata) sources = Elements("n", myntsconv.sourcedata) pipes = Elements("p", myntsconv.pipedata) # stores all pipe elements valves = Elements("v", myntsconv.valvedata) compressors = Elements("c", myntsconv.compressordata) ###drives = Elements("d", myntsconv.compressordata) # update progress pbar.update(25) # init writing to Mynts geom.jsn file output_folder = myntsconv.getBasicPath() # basic output file path myntsconv.setStrategy(DaVe2Mynts(output_folder)) # define Strategy # fixed order to write the elements ###all_elements = [nodes, sinks, sources, pipes, valves, compressors, drives] all_elements = [nodes, sinks, sources, pipes, valves, compressors] myntsconv.executeStrategy(all_elements) # print(text, ": ", ele_type.type, " written to Mynts Geom") # update progress pbar.update(25)