14308126759

Committed 07 Apr 2025 11:33AM UTC coverage: 27.288% (+0.05%) from 27.237%

Build # 14308126759

Build Type

Pull #229

github

Committed by

web-flow

Commit Message

Merge 21490326f into b47fe3c44

Pull Request Pull Request #229: Implement `copyPathsToVariationFolder` config command.

Run Details

70 of 120 new or added lines in 3 files covered. (58.33%)

4 existing lines in 2 files now uncovered.

3814 of 13977 relevant lines covered (27.29%)

0.27 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

41.52

/pytrnsys/rsim/runParallelTrnsys.py

# pylint: skip-file
# type: ignore

import json
import os
import pathlib as _pl
import shutil
from copy import deepcopy

import pandas as pd

import pytrnsys.rsim.executeTrnsys as exeTrnsys
import pytrnsys.rsim.getConfigMixin as _gcm
import pytrnsys.rsim.runParallel as runPar
import pytrnsys.trnsys_util.buildTrnsysDeck as _btd
import pytrnsys.trnsys_util.createTrnsysDeck as createDeck
import pytrnsys.trnsys_util.readConfigTrnsys as readConfig
import pytrnsys.trnsys_util.replaceAssignStatements as _ras
import pytrnsys.utils.log as log
import pytrnsys.utils.result as _res
import pytrnsys.utils.warnings as _warn


class RunParallelTrnsys(_gcm.GetConfigMixin):
    """
    Main class that represents a simulation job of pytrnsys. The standardized way of initiating it is by providing a
    config-file, in which case the run defined in the config case is started automatically.

    Args
    ----------
    path : str
        Path to the config file.
    name : str
        Main name of the Deck file, optional, default: "TrnsysRun". (Deprecated - should be defined in the config file.)
    configFile : str, None
        Name of the config file. If no argument is passed, the run has to be started by executing the methods externally
        , optional, default: None.

    Attributes
    ----------
    inputs : dict
        Dictionary containing the entries of the config file

    """

    def __init__(self, pathConfig, name="pytrnsysRun", configFile=None, runPath=None):
        super().__init__()

        self.pathConfig = pathConfig

        self.defaultInputs()
        self.cmds = []
        if runPath == None:
            self.path = os.getcwd()
        else:
            self.path = runPath

        if configFile is not None:
            self.readConfig(self.pathConfig, configFile)

            if "nameRef" in self.inputs:
                self.nameBase = self.inputs["nameRef"]
            else:
                if name == "pytrnsysRun":
                    self.nameBase = self.inputs["addResultsFolder"]

            if "projectPath" in self.inputs:
                self.path = self.inputs["projectPath"]
            elif runPath == None:
                self.path = os.getcwd()

            self.outputFileDebugRun = os.path.join(self.path, "debugParallelRun.dat")
            self.getConfig()

            result = self.runConfig()
            if _res.isError(result):
                _res.error(result).throw()

            self.runParallel()
        else:
            self.outputFileDebugRun = os.path.join(self.path, "debugParallelRun.dat")
            self.nameBase = name
            self.path = os.getcwd()

    def setDeckName(self, _name):
        self.nameBase = _name

    def setPath(self, path):
        self.path = path

    def defaultInputs(self):
        self.inputs = {}
        self.inputs["ignoreOnlinePlotter"] = False
        self.inputs["removePopUpWindow"] = False
        self.inputs["autoCloseOnlinePlotter"] = True
        self.inputs["checkDeck"] = True
        self.inputs["reduceCpu"] = 0
        self.inputs["combineAllCases"] = True
        self.inputs["parseFileCreated"] = True
        self.inputs["HOME$"] = None
        self.inputs["trnsysVersion"] = "TRNSYS_EXE"
        self.inputs["trnsysExePath"] = "enviromentalVariable"
        self.inputs["copyBuildingData"] = False  # activate when Type 55 is used or change the path to the source
        self.inputs["addResultsFolder"] = False
        self.inputs["rerunFailedCases"] = False
        self.inputs["scaling"] = False
        self.inputs["doAutoUnitNumbering"] = True
        self.inputs["addAutomaticEnergyBalance"] = True
        self.inputs["generateUnitTypesUsed"] = True
        self.inputs["runCases"] = True
        self.inputs["runType"] = "runFromConfig"
        self.inputs["outputLevel"] = "INFO"

        self.overwriteForcedByUser = False

        self.variablesOutput = []

    def readFromFolder(self, pathRun):
        fileNames = [name for name in os.listdir(pathRun) if os.path.isdir(pathRun + "\\" + name)]

        returnNames = []
        for n in fileNames:
            if n[0] == ".":  # filfer folders such as ".gle" and so on
                pass
            else:
                returnNames.append(n)

        return returnNames

    def readCasesToRun(self, pathRun, nameFileWithCasesToRun):
        fileToRunWithPath = os.path.join(pathRun, nameFileWithCasesToRun)
        file = open(fileToRunWithPath, "r")
        lines = file.readlines()
        cases = []

        for line in lines:
            if line == "\n" or line[0] == "#":  # ignoring blank lines and lines starting with #
                pass
            else:
                cases.append(line[:-1])  # remove \n

        return cases

    def runFromNames(self, path, fileNames):
        tests = []
        self.cmds = []
        for i in range(len(fileNames)):
            tests.append(exeTrnsys.ExecuteTrnsys(path, fileNames[i]))
            tests[i].setTrnsysExePath(self.inputs["trnsysExePath"])
            tests[i].loadDeck(check=self.inputs["checkDeck"])

            tests[i].changeParameter(self.parameters)

            if self.inputs["ignoreOnlinePlotter"] == True:
                tests[i].ignoreOnlinePlotter()

            tests[i].deckTrnsys.writeDeck()

            tests[i].setRemovePopUpWindow(self.inputs["removePopUpWindow"])

            self.cmds.append(tests[i].getExecuteTrnsys(self.inputs, useDeckName=tests[i].nameDck))

        self.runParallel()

    def runConfig(self) -> _res.Result[None]:
        """
        Runs the cases defined in the config file

        Returns
        -------

        """

        if self.inputs["runType"] == "runFromCases":
            cases = self.readCasesToRun(self.inputs["pathWithCasesToRun"], self.inputs["fileWithCasesToRun"])
            self.runFromNames(self.inputs["pathWithCasesToRun"], cases)
        elif self.inputs["runType"] == "runFromFolder":
            cases = self.readFromFolder(self.inputs["pathFolderToRun"])
            self.runFromNames(self.inputs["pathFolderToRun"], cases)
        elif self.inputs["runType"] == "runFromConfig":
            if self.changeDDckFilesUsed == True:
                # It actually loops around the changed files and then execute the parameters variations for each
                # so the definitons of two weathers will run all variations in two cities
                nameBase = self.nameBase
                self.unscaledVariables = deepcopy(self.variablesOutput)
                for i in range(len(self.sourceFilesToChange)):
                    originalSourceFile = self.sourceFilesToChange[i]
                    sourceFile = self.sourceFilesToChange[i]
                    for j in range(len(self.sinkFilesToChange[i])):
                        sinkFile = self.sinkFilesToChange[i][j]
                        self.changeDDckFile(sourceFile, sinkFile)
                        if "scalingVariable" in self.inputs.keys() and "scalingReference" in self.inputs.keys():
                            if "changeScalingFile" in self.inputs.keys():
                                self.scaleVariables(
                                    self.inputs["scalingReference"],
                                    self.inputs["changeScalingFile"][0],
                                    self.inputs["changeScalingFile"][j + 1],
                                )
                            else:
                                self.scaleVariables(self.inputs["scalingReference"], originalSourceFile, sinkFile)

                        sourceFile = sinkFile  # for each case the listddck will be changed to the new one, so we need to compare with the updated string

                        if self.foldersForDDckVariationUsed == True:
                            addFolder = self.foldersForDDckVariation[j]
                            originalPath = self.path
                            self.path = os.path.join(self.path, addFolder)
                            if not os.path.isdir(self.path):
                                os.mkdir(self.path)
                        else:
                            self.nameBase = nameBase + "-" + os.path.split(sinkFile)[-1]

                        result = self.buildTrnsysDeck()

                        if _res.isError(result):
                            error = _res.error(result)
                            self.logger.error(error.message)
                            return error

                        warnings = _res.value(result)
                        warnings.log(self.logger)

                        self.createDecksFromVariant()

                        if self.foldersForDDckVariationUsed:
                            self.path = originalPath  # recall the original path, otherwise the next folder will be cerated inside the first
            else:
                result = self.buildTrnsysDeck()

                if _res.isError(result):
                    error = _res.error(result)
                    self.logger.error(error.message)
                    return error

                warnings = _res.value(result)
                warnings.log(self.logger)

                self.createDecksFromVariant()

    def createDecksFromVariant(self, fitParameters={}):
        variations = self.variablesOutput
        parameters = self.parameters
        parameters.update(fitParameters)

        myDeckGenerator = createDeck.CreateTrnsysDeck(self.path, self.nameBase, variations)

        myDeckGenerator.combineAllCases = self.inputs["combineAllCases"]

        successfulCases = []

        if "masterFile" in self.inputs:
            if os.path.isfile(self.inputs["masterFile"]):
                try:
                    masterDf = pd.read_csv(self.inputs["masterFile"], sep=";", index_col=0)
                except:
                    self.logger.error("Unable to read " + self.inputs["masterFile"])
                    self.logger.error("Variation dck files of %s won't be created" % self.nameBase)
                    return

                self.logger.info("Checking for successful runs in " + self.inputs["masterFile"])
                for index, row in masterDf.iterrows():
                    if row["outcome"] == "success":
                        successfulCases.append(index)
            else:
                self.logger.info("Master file does not exist, no runs will be excluded")

        # creates a list of decks with the appropriate name but nothing changed inside!!
        if self.variationsUsed or (self.changeDDckFilesUsed == True and self.foldersForDDckVariationUsed == False):
            if successfulCases:
                fileName = myDeckGenerator.generateDecks(successfulCases=successfulCases)
            else:
                fileName = myDeckGenerator.generateDecks()
        else:
            fileName = []
            fileName.append(self.nameBase)

        if myDeckGenerator.noVariationCreated and self.variation:
            self.logger.warning("No variation dck files created from " + self.nameBase)
            return

        tests = []

        variablePath = self.path

        for i in range(len(fileName)):
            self.logger.debug("name to run :%s" % fileName[i])

            # if useLocationStructure:
            # variablePath = os.path.join(path,location) #assign subfolder for path

            # Parameters changed by variation
            localCopyPar = dict.copy(parameters)  #

            if self.variationsUsed:
                myParameters = myDeckGenerator.getParameters(i)
                localCopyPar.update(myParameters)

            # We add to the global parameters that also need to be modified
            # If we assign like localCopyPar = parameters, then the parameters will change with localCopyPar !!
            # Otherwise we change the global parameter and some values of last variation will remain.

            test = exeTrnsys.ExecuteTrnsys(variablePath, fileName[i])
            tests.append(test)

            test.setTrnsysExePath(self.inputs["trnsysExePath"])

            test.setRemovePopUpWindow(self.inputs["removePopUpWindow"])

            test.moveFileFromSource()

            test.loadDeck(useDeckOutputPath=True)

            test.changeParameter(localCopyPar)

            test.changeAssignStatementsBasedOnUnitVariables(self._assignStatements)
            
            for pathToCopy in self._allPathsToCopy:
                test.copyPathToVariationFolder(pathToCopy.source, pathToCopy.target)

            if self.inputs["ignoreOnlinePlotter"] == True:
                test.ignoreOnlinePlotter()

            test.deckTrnsys.writeDeck()

            test.cleanAndCreateResultsTempFolder()
            test.moveFileFromSource()

            if self.inputs["runCases"] == True:
                self.cmds.append(test.getExecuteTrnsys(self.inputs))

    def createLocationFolders(path, locations):
        for location in locations:
            if not os.path.exists(location):
                os.makedirs(location)
                print("created directory '") + path + location + "'"

    def moveResultsFolder(path, resultsFolder, destinationFolder):
        root_src_dir = os.path.join(path, resultsFolder)
        root_target_dir = os.path.join(path, destinationFolder)

        shutil.move(root_src_dir, root_target_dir)

    def buildTrnsysDeck(self) -> _res.Result[_warn.ValueWithWarnings[str]]:
        """
        It builds a TRNSYS Deck from a listDdck with pathDdck using the BuildingTrnsysDeck Class.
        it reads the Deck list and writes a deck file. Afterwards it checks that the deck looks fine

        """
        #  I can create folders in another path to move them in the running folder and run them one by one
        #  path = "C:\Daten\OngoingProject\Ice-Ex\systemSimulations\\check\\"

        deckExplanation = []
        deckExplanation.append("! ** New deck built from list of ddcks. **\n")
        deck = _btd.BuildTrnsysDeck(
            self.path,
            self.nameBase,
            self._includedDdckFiles,
            self._defaultVisibility,
            self._ddckPlaceHolderValuesJsonPath,
        )
        result = deck.readDeckList(
            self.pathConfig,
            doAutoUnitNumbering=self.inputs["doAutoUnitNumbering"],
            dictPaths=self.dictDdckPaths,
            replaceLineList=self.replaceLines,
        )

        if _res.isError(result):
            return _res.error(result)
        warnings = _res.value(result)

        deck.overwriteForcedByUser = self.overwriteForcedByUser
        deck.writeDeck(addedLines=deckExplanation)
        self.overwriteForcedByUser = deck.overwriteForcedByUser

        result = deck.checkTrnsysDeck(deck.nameDeck, check=self.inputs["checkDeck"])
        if _res.isError(result):
            return _res.error(result)

        if self.inputs["generateUnitTypesUsed"] == True:
            deck.saveUnitTypeFile()

        if self.inputs["addAutomaticEnergyBalance"] == True:
            deck.addAutomaticEnergyBalancePrinters()
            deck.writeDeck()  # Deck rewritten with added printer

        deck.analyseDck()

        return warnings.withValue(deck.nameDeck)

    def runParallel(self, writeLogFile=True):
        if writeLogFile:
            self.writeRunLogFile()

        if "masterFile" in self.inputs:
            runPar.runParallel(
                self.cmds,
                reduceCpu=int(self.inputs["reduceCpu"]),
                trackingFile=self.inputs["trackingFile"],
                masterFile=self.inputs["masterFile"],
            )
        elif "trackingFile" in self.inputs:
            runPar.runParallel(
                self.cmds, reduceCpu=int(self.inputs["reduceCpu"]), trackingFile=self.inputs["trackingFile"]
            )
        else:
            runPar.runParallel(self.cmds, reduceCpu=int(self.inputs["reduceCpu"]), outputFile=self.outputFileDebugRun)

    def writeRunLogFile(self):
        logfile = open(os.path.join(self.path, "runLogFile.config"), "w")
        username = os.getenv("username")
        logfile.write("# Run created by " + username + "\n")
        logfile.write("# Ddck repositories used:\n")
        try:
            import git

            found = True
        except ModuleNotFoundError:
            found = False

        for path in self.listDdckPaths:
            logfile.write("# " + path + "\n")
            logfile.write("# Revision Hash number: ")
            if found:
                try:
                    repo = git.Repo(path, search_parent_directories=True)
                    logfile.write(str(repo.head.object.hexsha) + "\n")
                except git.exc.InvalidGitRepositoryError:
                    logfile.write("None - Not in a Git repository \n")
            else:
                logfile.write("not found \n")
        logfile.write("\n# Config file used: \n\n")
        for line in self.lines:
            logfile.write(line + "\n")
        logfile.close()

    def readConfig(self, path, name, parseFileCreated=False):
        """
        It reads the config file used for running TRNSYS and loads the self.inputs dictionary.
        It also loads the readed lines into self.lines
        """
        tool = readConfig.ReadConfigTrnsys()

        self.lines = tool.readFile(path, name, self.inputs, parseFileCreated=parseFileCreated, controlDataType=False)

        self.logger = log.getOrCreateCustomLogger("root", self.inputs["outputLevel"])

        if "pathBaseSimulations" in self.inputs:
            self.path = self.inputs["pathBaseSimulations"]
        if self.inputs["addResultsFolder"]:
            self.path = os.path.join(self.path, self.inputs["addResultsFolder"])

            if not os.path.isdir(self.path):
                os.mkdir(self.path)

    def changeFile(self, source, end):
        """
        It uses the self-lines readed by readConfig and change the lines from source to end.
        This is used to change a ddck file readed for another. A typical example is the weather data file
        Parameters
        ----------
        source : str
            string to be replaced in the config file in the self.lines field
        end : str
            str to replace the source in the config file in the self.lines field

        Returns
        -------

        """

        found = False
        for i in range(len(self.lines)):
            lineFilter = self.lines[i]

            if lineFilter == source:
                self.lines[i] = end
                found = True

        if found == False:
            self.logger.warning("change File was not able to change %s by %s" % (source, end))

    def changeDDckFile(self, source, end):
        """
        It uses the  self.listDdck readed by readConfig and change the lines from source to end.
        This is used to change a ddck file readed for another. A typical example is the weather data file
        """
        found = False
        nCharacters = len(source)

        for i in range(len(self._includedDdckFiles)):
            oldIncludedDdckFile = self._includedDdckFiles[i]

            ddckFilePath = str(oldIncludedDdckFile.pathWithoutSuffix)

            mySource = ddckFilePath[-nCharacters:]  # I read only the last characters with the same size as the end file
            if mySource == source:
                newDdckFilePath = ddckFilePath[0:-nCharacters] + end
                self.dictDdckPaths[newDdckFilePath] = self.dictDdckPaths[ddckFilePath]
                newIncludedDdckFile = _btd.IncludedDdckFile(
                    _pl.Path(newDdckFilePath), oldIncludedDdckFile.componentName, oldIncludedDdckFile.defaultVisibility
                )
                self._includedDdckFiles[i] = newIncludedDdckFile

                found = True

        if not found:
            self.logger.warning("change File was not able to change %s by %s" % (source, end))

    def copyConfigFile(self, configPath, configName):
        configFile = os.path.join(configPath, configName)
        dstPath = os.path.join(configPath, self.inputs["addResultsFolder"], configName)
        shutil.copyfile(configFile, dstPath)
        self.logger.debug("copied config file to: %s" % dstPath)

    def scaleVariables(self, reference, source, sink):
        """

        Parameters
        ----------
        reference : str
            File path of the reference results file. Has to point to a json-File with the pytrnsys resuls file format
        source : str
            Substring to be replaced in reference
        sink : str
            String to replace the substrings in the reference

        Returns
        -------

        """
        resultFile = reference.replace(source, sink)
        with open(resultFile) as f_in:
            resultsDict = json.load(f_in)

        exec("scalingVariable=" + self.inputs["scalingVariable"], globals(), resultsDict)
        loadDemand = resultsDict["scalingVariable"]

        try:
            exec("scalingElDemandVariable=" + self.inputs["scalingElDemandVariable"], globals(), resultsDict)
            loadElDemand = resultsDict["scalingElDemandVariable"]
        except:
            pass

        try:
            exec("scaleHP=" + self.inputs["scaleHP"], globals(), resultsDict)
            loadHPsize = resultsDict["scaleHP"]
        except:
            pass

        for j in range(len(self.variablesOutput)):
            for i in range(2, len(self.variablesOutput[j]), 1):
                if self.variablesOutput[j][1] == "sizeHpUsed":
                    self.variablesOutput[j][i] = (
                        str(round(self.unscaledVariables[j][i], 3)) + "*" + str(round(loadHPsize, 3))
                    )
                elif self.variablesOutput[j][1] == "AreaPvRoof":
                    self.variablesOutput[j][i] = (
                        str(round(self.unscaledVariables[j][i], 3)) + "*" + str(round(loadElDemand, 3))
                    )
                else:
                    self.variablesOutput[j][i] = (
                        str(round(self.unscaledVariables[j][i], 3)) + "*" + str(round(loadDemand, 3))
                    )

1	# pylint: skip-file
2	# type: ignore
3
4	import json	1✔
5	import os	1✔
6	import pathlib as _pl	1✔
7	import shutil	1✔
8	from copy import deepcopy	1✔
9
10	import pandas as pd	1✔
11
12	import pytrnsys.rsim.executeTrnsys as exeTrnsys	1✔
13	import pytrnsys.rsim.getConfigMixin as _gcm	1✔
14	import pytrnsys.rsim.runParallel as runPar	1✔
15	import pytrnsys.trnsys_util.buildTrnsysDeck as _btd	1✔
16	import pytrnsys.trnsys_util.createTrnsysDeck as createDeck	1✔
17	import pytrnsys.trnsys_util.readConfigTrnsys as readConfig	1✔
18	import pytrnsys.trnsys_util.replaceAssignStatements as _ras	1✔
19	import pytrnsys.utils.log as log	1✔
20	import pytrnsys.utils.result as _res	1✔
21	import pytrnsys.utils.warnings as _warn	1✔
22
23
24	class RunParallelTrnsys(_gcm.GetConfigMixin):	1✔
25	"""	1✔
26	Main class that represents a simulation job of pytrnsys. The standardized way of initiating it is by providing a
27	config-file, in which case the run defined in the config case is started automatically.
28
29	Args
30	----------
31	path : str
32	Path to the config file.
33	name : str
34	Main name of the Deck file, optional, default: "TrnsysRun". (Deprecated - should be defined in the config file.)
35	configFile : str, None
36	Name of the config file. If no argument is passed, the run has to be started by executing the methods externally
37	, optional, default: None.
38
39	Attributes
40	----------
41	inputs : dict
42	Dictionary containing the entries of the config file
43
44	"""
45
46	def __init__(self, pathConfig, name="pytrnsysRun", configFile=None, runPath=None):	1✔
47	super().__init__()	1✔
48
49	self.pathConfig = pathConfig	1✔
50
51	self.defaultInputs()	1✔
52	self.cmds = []	1✔
53	if runPath == None:	1✔
54	self.path = os.getcwd()	1✔
55	else:
56	self.path = runPath	×
57
58	if configFile is not None:	1✔
59	self.readConfig(self.pathConfig, configFile)	×
60
61	if "nameRef" in self.inputs:	×
62	self.nameBase = self.inputs["nameRef"]	×
63	else:
64	if name == "pytrnsysRun":	×
65	self.nameBase = self.inputs["addResultsFolder"]	×
66
67	if "projectPath" in self.inputs:	×
68	self.path = self.inputs["projectPath"]	×
69	elif runPath == None:	×
70	self.path = os.getcwd()	×
71
72	self.outputFileDebugRun = os.path.join(self.path, "debugParallelRun.dat")	×
73	self.getConfig()	×
74
75	result = self.runConfig()	×
76	if _res.isError(result):	×
77	_res.error(result).throw()	×
78
79	self.runParallel()	×
80	else:
81	self.outputFileDebugRun = os.path.join(self.path, "debugParallelRun.dat")	1✔
82	self.nameBase = name	1✔
83	self.path = os.getcwd()	1✔
84
85	def setDeckName(self, _name):	1✔
UNCOV 86	self.nameBase = _name	×
87
88	def setPath(self, path):	1✔
89	self.path = path	×
90
91	def defaultInputs(self):	1✔
92	self.inputs = {}	1✔
93	self.inputs["ignoreOnlinePlotter"] = False	1✔
94	self.inputs["removePopUpWindow"] = False	1✔
95	self.inputs["autoCloseOnlinePlotter"] = True	1✔
96	self.inputs["checkDeck"] = True	1✔
97	self.inputs["reduceCpu"] = 0	1✔
98	self.inputs["combineAllCases"] = True	1✔
99	self.inputs["parseFileCreated"] = True	1✔
100	self.inputs["HOME$"] = None	1✔
101	self.inputs["trnsysVersion"] = "TRNSYS_EXE"	1✔
102	self.inputs["trnsysExePath"] = "enviromentalVariable"	1✔
103	self.inputs["copyBuildingData"] = False # activate when Type 55 is used or change the path to the source	1✔
104	self.inputs["addResultsFolder"] = False	1✔
105	self.inputs["rerunFailedCases"] = False	1✔
106	self.inputs["scaling"] = False	1✔
107	self.inputs["doAutoUnitNumbering"] = True	1✔
108	self.inputs["addAutomaticEnergyBalance"] = True	1✔
109	self.inputs["generateUnitTypesUsed"] = True	1✔
110	self.inputs["runCases"] = True	1✔
111	self.inputs["runType"] = "runFromConfig"	1✔
112	self.inputs["outputLevel"] = "INFO"	1✔
113
114	self.overwriteForcedByUser = False	1✔
115
116	self.variablesOutput = []	1✔
117
118	def readFromFolder(self, pathRun):	1✔
119	fileNames = [name for name in os.listdir(pathRun) if os.path.isdir(pathRun + "\\" + name)]	×
120
121	returnNames = []	×
122	for n in fileNames:	×
123	if n[0] == ".": # filfer folders such as ".gle" and so on	×
124	pass	×
125	else:
126	returnNames.append(n)	×
127
128	return returnNames	×
129
130	def readCasesToRun(self, pathRun, nameFileWithCasesToRun):	1✔
131	fileToRunWithPath = os.path.join(pathRun, nameFileWithCasesToRun)	×
132	file = open(fileToRunWithPath, "r")	×
133	lines = file.readlines()	×
134	cases = []	×
135
136	for line in lines:	×
137	if line == "\n" or line[0] == "#": # ignoring blank lines and lines starting with #	×
138	pass	×
139	else:
140	cases.append(line[:-1]) # remove \n	×
141
142	return cases	×
143
144	def runFromNames(self, path, fileNames):	1✔
145	tests = []	×
146	self.cmds = []	×
147	for i in range(len(fileNames)):	×
148	tests.append(exeTrnsys.ExecuteTrnsys(path, fileNames[i]))	×
149	tests[i].setTrnsysExePath(self.inputs["trnsysExePath"])	×
150	tests[i].loadDeck(check=self.inputs["checkDeck"])	×
151
152	tests[i].changeParameter(self.parameters)	×
153
154	if self.inputs["ignoreOnlinePlotter"] == True:	×
155	tests[i].ignoreOnlinePlotter()	×
156
157	tests[i].deckTrnsys.writeDeck()	×
158
159	tests[i].setRemovePopUpWindow(self.inputs["removePopUpWindow"])	×
160
161	self.cmds.append(tests[i].getExecuteTrnsys(self.inputs, useDeckName=tests[i].nameDck))	×
162
163	self.runParallel()	×
164
165	def runConfig(self) -> _res.Result[None]:	1✔
166	"""
167	Runs the cases defined in the config file
168
169	Returns
170	-------
171
172	"""
173
174	if self.inputs["runType"] == "runFromCases":	1✔
175	cases = self.readCasesToRun(self.inputs["pathWithCasesToRun"], self.inputs["fileWithCasesToRun"])	×
176	self.runFromNames(self.inputs["pathWithCasesToRun"], cases)	×
177	elif self.inputs["runType"] == "runFromFolder":	1✔
178	cases = self.readFromFolder(self.inputs["pathFolderToRun"])	×
179	self.runFromNames(self.inputs["pathFolderToRun"], cases)	×
180	elif self.inputs["runType"] == "runFromConfig":	1✔
181	if self.changeDDckFilesUsed == True:	1✔
182	# It actually loops around the changed files and then execute the parameters variations for each
183	# so the definitons of two weathers will run all variations in two cities
184	nameBase = self.nameBase	×
185	self.unscaledVariables = deepcopy(self.variablesOutput)	×
186	for i in range(len(self.sourceFilesToChange)):	×
187	originalSourceFile = self.sourceFilesToChange[i]	×
188	sourceFile = self.sourceFilesToChange[i]	×
189	for j in range(len(self.sinkFilesToChange[i])):	×
190	sinkFile = self.sinkFilesToChange[i][j]	×
191	self.changeDDckFile(sourceFile, sinkFile)	×
192	if "scalingVariable" in self.inputs.keys() and "scalingReference" in self.inputs.keys():	×
193	if "changeScalingFile" in self.inputs.keys():	×
194	self.scaleVariables(	×
195	self.inputs["scalingReference"],
196	self.inputs["changeScalingFile"][0],
197	self.inputs["changeScalingFile"][j + 1],
198	)
199	else:
200	self.scaleVariables(self.inputs["scalingReference"], originalSourceFile, sinkFile)	×
201
202	sourceFile = sinkFile # for each case the listddck will be changed to the new one, so we need to compare with the updated string	×
203
204	if self.foldersForDDckVariationUsed == True:	×
205	addFolder = self.foldersForDDckVariation[j]	×
206	originalPath = self.path	×
207	self.path = os.path.join(self.path, addFolder)	×
208	if not os.path.isdir(self.path):	×
209	os.mkdir(self.path)	×
210	else:
211	self.nameBase = nameBase + "-" + os.path.split(sinkFile)[-1]	×
212
213	result = self.buildTrnsysDeck()	×
214
215	if _res.isError(result):	×
216	error = _res.error(result)	×
217	self.logger.error(error.message)	×
218	return error	×
219
220	warnings = _res.value(result)	×
221	warnings.log(self.logger)	×
222
223	self.createDecksFromVariant()	×
224
225	if self.foldersForDDckVariationUsed:	×
226	self.path = originalPath # recall the original path, otherwise the next folder will be cerated inside the first	×
227	else:
228	result = self.buildTrnsysDeck()	1✔
229
230	if _res.isError(result):	1✔
231	error = _res.error(result)	×
232	self.logger.error(error.message)	×
233	return error	×
234
235	warnings = _res.value(result)	1✔
236	warnings.log(self.logger)	1✔
237
238	self.createDecksFromVariant()	1✔
239
240	def createDecksFromVariant(self, fitParameters={}):	1✔
241	variations = self.variablesOutput	1✔
242	parameters = self.parameters	1✔
243	parameters.update(fitParameters)	1✔
244
245	myDeckGenerator = createDeck.CreateTrnsysDeck(self.path, self.nameBase, variations)	1✔
246
247	myDeckGenerator.combineAllCases = self.inputs["combineAllCases"]	1✔
248
249	successfulCases = []	1✔
250
251	if "masterFile" in self.inputs:	1✔
252	if os.path.isfile(self.inputs["masterFile"]):	×
253	try:	×
254	masterDf = pd.read_csv(self.inputs["masterFile"], sep=";", index_col=0)	×
255	except:	×
256	self.logger.error("Unable to read " + self.inputs["masterFile"])	×
257	self.logger.error("Variation dck files of %s won't be created" % self.nameBase)	×
258	return	×
259
260	self.logger.info("Checking for successful runs in " + self.inputs["masterFile"])	×
261	for index, row in masterDf.iterrows():	×
262	if row["outcome"] == "success":	×
263	successfulCases.append(index)	×
264	else:
265	self.logger.info("Master file does not exist, no runs will be excluded")	×
266
267	# creates a list of decks with the appropriate name but nothing changed inside!!
268	if self.variationsUsed or (self.changeDDckFilesUsed == True and self.foldersForDDckVariationUsed == False):	1✔
269	if successfulCases:	×
270	fileName = myDeckGenerator.generateDecks(successfulCases=successfulCases)	×
271	else:
272	fileName = myDeckGenerator.generateDecks()	×
273	else:
274	fileName = []	1✔
275	fileName.append(self.nameBase)	1✔
276
277	if myDeckGenerator.noVariationCreated and self.variation:	1✔
278	self.logger.warning("No variation dck files created from " + self.nameBase)	×
279	return	×
280
281	tests = []	1✔
282
283	variablePath = self.path	1✔
284
285	for i in range(len(fileName)):	1✔
286	self.logger.debug("name to run :%s" % fileName[i])	1✔
287
288	# if useLocationStructure:
289	# variablePath = os.path.join(path,location) #assign subfolder for path
290
291	# Parameters changed by variation
292	localCopyPar = dict.copy(parameters) #	1✔
293
294	if self.variationsUsed:	1✔
295	myParameters = myDeckGenerator.getParameters(i)	×
296	localCopyPar.update(myParameters)	×
297
298	# We add to the global parameters that also need to be modified
299	# If we assign like localCopyPar = parameters, then the parameters will change with localCopyPar !!
300	# Otherwise we change the global parameter and some values of last variation will remain.
301
302	test = exeTrnsys.ExecuteTrnsys(variablePath, fileName[i])	1✔
303	tests.append(test)	1✔
304
305	test.setTrnsysExePath(self.inputs["trnsysExePath"])	1✔
306
307	test.setRemovePopUpWindow(self.inputs["removePopUpWindow"])	1✔
308
309	test.moveFileFromSource()	1✔
310
311	test.loadDeck(useDeckOutputPath=True)	1✔
312
313	test.changeParameter(localCopyPar)	1✔
314
315	test.changeAssignStatementsBasedOnUnitVariables(self._assignStatements)	1✔
316
317	for pathToCopy in self._allPathsToCopy:	1✔
NEW 318	test.copyPathToVariationFolder(pathToCopy.source, pathToCopy.target)	×
319
320	if self.inputs["ignoreOnlinePlotter"] == True:	1✔
NEW 321	test.ignoreOnlinePlotter()	×
322
323	test.deckTrnsys.writeDeck()	1✔
324
325	test.cleanAndCreateResultsTempFolder()	1✔
326	test.moveFileFromSource()	1✔
327
328	if self.inputs["runCases"] == True:	1✔
329	self.cmds.append(test.getExecuteTrnsys(self.inputs))	1✔
330
331	def createLocationFolders(path, locations):	1✔
332	for location in locations:	×
333	if not os.path.exists(location):	×
334	os.makedirs(location)	×
335	print("created directory '") + path + location + "'"	×
336
337	def moveResultsFolder(path, resultsFolder, destinationFolder):	1✔
338	root_src_dir = os.path.join(path, resultsFolder)	×
339	root_target_dir = os.path.join(path, destinationFolder)	×
340
341	shutil.move(root_src_dir, root_target_dir)	×
342
343	def buildTrnsysDeck(self) -> _res.Result[_warn.ValueWithWarnings[str]]:	1✔
344	"""
345	It builds a TRNSYS Deck from a listDdck with pathDdck using the BuildingTrnsysDeck Class.
346	it reads the Deck list and writes a deck file. Afterwards it checks that the deck looks fine
347
348	"""
349	# I can create folders in another path to move them in the running folder and run them one by one
350	# path = "C:\Daten\OngoingProject\Ice-Ex\systemSimulations\\check\\"
351
352	deckExplanation = []	1✔
353	deckExplanation.append("! New deck built from list of ddcks. \n")	1✔
354	deck = _btd.BuildTrnsysDeck(	1✔
355	self.path,
356	self.nameBase,
357	self._includedDdckFiles,
358	self._defaultVisibility,
359	self._ddckPlaceHolderValuesJsonPath,
360	)
361	result = deck.readDeckList(	1✔
362	self.pathConfig,
363	doAutoUnitNumbering=self.inputs["doAutoUnitNumbering"],
364	dictPaths=self.dictDdckPaths,
365	replaceLineList=self.replaceLines,
366	)
367
368	if _res.isError(result):	1✔
369	return _res.error(result)	×
370	warnings = _res.value(result)	1✔
371
372	deck.overwriteForcedByUser = self.overwriteForcedByUser	1✔
373	deck.writeDeck(addedLines=deckExplanation)	1✔
374	self.overwriteForcedByUser = deck.overwriteForcedByUser	1✔
375
376	result = deck.checkTrnsysDeck(deck.nameDeck, check=self.inputs["checkDeck"])	1✔
377	if _res.isError(result):	1✔
378	return _res.error(result)	×
379
380	if self.inputs["generateUnitTypesUsed"] == True:	1✔
381	deck.saveUnitTypeFile()	1✔
382
383	if self.inputs["addAutomaticEnergyBalance"] == True:	1✔
384	deck.addAutomaticEnergyBalancePrinters()	1✔
385	deck.writeDeck() # Deck rewritten with added printer	1✔
386
387	deck.analyseDck()	1✔
388
389	return warnings.withValue(deck.nameDeck)	1✔
390
391	def runParallel(self, writeLogFile=True):	1✔
392	if writeLogFile:	×
393	self.writeRunLogFile()	×
394
395	if "masterFile" in self.inputs:	×
396	runPar.runParallel(	×
397	self.cmds,
398	reduceCpu=int(self.inputs["reduceCpu"]),
399	trackingFile=self.inputs["trackingFile"],
400	masterFile=self.inputs["masterFile"],
401	)
402	elif "trackingFile" in self.inputs:	×
403	runPar.runParallel(	×
404	self.cmds, reduceCpu=int(self.inputs["reduceCpu"]), trackingFile=self.inputs["trackingFile"]
405	)
406	else:
407	runPar.runParallel(self.cmds, reduceCpu=int(self.inputs["reduceCpu"]), outputFile=self.outputFileDebugRun)	×
408
409	def writeRunLogFile(self):	1✔
410	logfile = open(os.path.join(self.path, "runLogFile.config"), "w")	×
411	username = os.getenv("username")	×
412	logfile.write("# Run created by " + username + "\n")	×
413	logfile.write("# Ddck repositories used:\n")	×
414	try:	×
415	import git	×
416
417	found = True	×
418	except ModuleNotFoundError:	×
419	found = False	×
420
421	for path in self.listDdckPaths:	×
422	logfile.write("# " + path + "\n")	×
423	logfile.write("# Revision Hash number: ")	×
424	if found:	×
425	try:	×
426	repo = git.Repo(path, search_parent_directories=True)	×
427	logfile.write(str(repo.head.object.hexsha) + "\n")	×
428	except git.exc.InvalidGitRepositoryError:	×
429	logfile.write("None - Not in a Git repository \n")	×
430	else:
431	logfile.write("not found \n")	×
432	logfile.write("\n# Config file used: \n\n")	×
433	for line in self.lines:	×
434	logfile.write(line + "\n")	×
435	logfile.close()	×
436
437	def readConfig(self, path, name, parseFileCreated=False):	1✔
438	"""
439	It reads the config file used for running TRNSYS and loads the self.inputs dictionary.
440	It also loads the readed lines into self.lines
441	"""
442	tool = readConfig.ReadConfigTrnsys()	1✔
443
444	self.lines = tool.readFile(path, name, self.inputs, parseFileCreated=parseFileCreated, controlDataType=False)	1✔
445
446	self.logger = log.getOrCreateCustomLogger("root", self.inputs["outputLevel"])	1✔
447
448	if "pathBaseSimulations" in self.inputs:	1✔
449	self.path = self.inputs["pathBaseSimulations"]	×
450	if self.inputs["addResultsFolder"]:	1✔
451	self.path = os.path.join(self.path, self.inputs["addResultsFolder"])	1✔
452
453	if not os.path.isdir(self.path):	1✔
454	os.mkdir(self.path)	1✔
455
456	def changeFile(self, source, end):	1✔
457	"""
458	It uses the self-lines readed by readConfig and change the lines from source to end.
459	This is used to change a ddck file readed for another. A typical example is the weather data file
460	Parameters
461	----------
462	source : str
463	string to be replaced in the config file in the self.lines field
464	end : str
465	str to replace the source in the config file in the self.lines field
466
467	Returns
468	-------
469
470	"""
471
472	found = False	×
473	for i in range(len(self.lines)):	×
474	lineFilter = self.lines[i]	×
475
476	if lineFilter == source:	×
477	self.lines[i] = end	×
478	found = True	×
479
480	if found == False:	×
481	self.logger.warning("change File was not able to change %s by %s" % (source, end))	×
482
483	def changeDDckFile(self, source, end):	1✔
484	"""
485	It uses the self.listDdck readed by readConfig and change the lines from source to end.
486	This is used to change a ddck file readed for another. A typical example is the weather data file
487	"""
488	found = False	×
489	nCharacters = len(source)	×
490
491	for i in range(len(self._includedDdckFiles)):	×
492	oldIncludedDdckFile = self._includedDdckFiles[i]	×
493
494	ddckFilePath = str(oldIncludedDdckFile.pathWithoutSuffix)	×
495
496	mySource = ddckFilePath[-nCharacters:] # I read only the last characters with the same size as the end file	×
497	if mySource == source:	×
498	newDdckFilePath = ddckFilePath[0:-nCharacters] + end	×
499	self.dictDdckPaths[newDdckFilePath] = self.dictDdckPaths[ddckFilePath]	×
500	newIncludedDdckFile = _btd.IncludedDdckFile(	×
501	_pl.Path(newDdckFilePath), oldIncludedDdckFile.componentName, oldIncludedDdckFile.defaultVisibility
502	)
503	self._includedDdckFiles[i] = newIncludedDdckFile	×
504
505	found = True	×
506
507	if not found:	×
508	self.logger.warning("change File was not able to change %s by %s" % (source, end))	×
509
510	def copyConfigFile(self, configPath, configName):	1✔
511	configFile = os.path.join(configPath, configName)	×
512	dstPath = os.path.join(configPath, self.inputs["addResultsFolder"], configName)	×
513	shutil.copyfile(configFile, dstPath)	×
514	self.logger.debug("copied config file to: %s" % dstPath)	×
515
516	def scaleVariables(self, reference, source, sink):	1✔
517	"""
518
519	Parameters
520	----------
521	reference : str
522	File path of the reference results file. Has to point to a json-File with the pytrnsys resuls file format
523	source : str
524	Substring to be replaced in reference
525	sink : str
526	String to replace the substrings in the reference
527
528	Returns
529	-------
530
531	"""
532	resultFile = reference.replace(source, sink)	×
533	with open(resultFile) as f_in:	×
534	resultsDict = json.load(f_in)	×
535
536	exec("scalingVariable=" + self.inputs["scalingVariable"], globals(), resultsDict)	×
537	loadDemand = resultsDict["scalingVariable"]	×
538
539	try:	×
540	exec("scalingElDemandVariable=" + self.inputs["scalingElDemandVariable"], globals(), resultsDict)	×
541	loadElDemand = resultsDict["scalingElDemandVariable"]	×
542	except:	×
543	pass	×
544
545	try:	×
546	exec("scaleHP=" + self.inputs["scaleHP"], globals(), resultsDict)	×
547	loadHPsize = resultsDict["scaleHP"]	×
548	except:	×
549	pass	×
550
551	for j in range(len(self.variablesOutput)):	×
552	for i in range(2, len(self.variablesOutput[j]), 1):	×
553	if self.variablesOutput[j][1] == "sizeHpUsed":	×
554	self.variablesOutput[j][i] = (	×
555	str(round(self.unscaledVariables[j][i], 3)) + "*" + str(round(loadHPsize, 3))
556	)
557	elif self.variablesOutput[j][1] == "AreaPvRoof":	×
558	self.variablesOutput[j][i] = (	×
559	str(round(self.unscaledVariables[j][i], 3)) + "*" + str(round(loadElDemand, 3))
560	)
561	else:
562	self.variablesOutput[j][i] = (	×
563	str(round(self.unscaledVariables[j][i], 3)) + "*" + str(round(loadDemand, 3))
564	)

SPF-OST / pytrnsys / 14308126759

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous