15605215655

Committed 12 Jun 2025 08:12AM UTC coverage: 27.513% (+0.1%) from 27.368%

Build # 15605215655

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #244 from SPF-OST/run-trnsys-from-dck-dir

Run trnsys from dck dir

Run Details

41 of 74 new or added lines in 4 files covered. (55.41%)

9 existing lines in 2 files now uncovered.

3843 of 13968 relevant lines covered (27.51%)

0.28 hits per line

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

41.69

/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.includedDdckFile as _idf
import pytrnsys.trnsys_util.readConfigTrnsys as readConfig
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.commands = []
        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.commands = []
        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.commands.append(tests[i].getExecuteTrnsys(self.inputs, useDeckName=tests[i].nameDck))

        self.runParallel()

    def runConfig(self) -> _res.Result[None]:
        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)

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

            test.deckTrnsys.writeDeck()

            test.cleanAndCreateResultsTempFolder()
            test.moveFileFromSource()

            test.clearOrCreateVariationDataFolder()
            for pathToCopy in self._allPathsToCopyToVariationDataFolder:
                test.copyPathToVariationDataFolder(pathToCopy.source, pathToCopy.target)

            if self.inputs["runCases"] == True:
                self.commands.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.commands,
                reduceCpu=int(self.inputs["reduceCpu"]),
                trackingFile=self.inputs["trackingFile"],
                masterFile=self.inputs["masterFile"],
            )
        elif "trackingFile" in self.inputs:
            runPar.runParallel(
                self.commands, reduceCpu=int(self.inputs["reduceCpu"]), trackingFile=self.inputs["trackingFile"]
            )
        else:
            runPar.runParallel(
                self.commands, 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 = _idf.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.includedDdckFile as _idf	1✔
18	import pytrnsys.trnsys_util.readConfigTrnsys as readConfig	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.commands = []	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✔
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 = []	×
NEW 146	self.commands = []	×
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
NEW 161	self.commands.append(tests[i].getExecuteTrnsys(self.inputs, useDeckName=tests[i].nameDck))	×
162
163	self.runParallel()	×
164
165	def runConfig(self) -> _res.Result[None]:	1✔
166	if self.inputs["runType"] == "runFromCases":	1✔
UNCOV 167	cases = self.readCasesToRun(self.inputs["pathWithCasesToRun"], self.inputs["fileWithCasesToRun"])	×
168	self.runFromNames(self.inputs["pathWithCasesToRun"], cases)	×
169	elif self.inputs["runType"] == "runFromFolder":	1✔
170	cases = self.readFromFolder(self.inputs["pathFolderToRun"])	×
171	self.runFromNames(self.inputs["pathFolderToRun"], cases)	×
172	elif self.inputs["runType"] == "runFromConfig":	1✔
173	if self.changeDDckFilesUsed == True:	1✔
174	# It actually loops around the changed files and then execute the parameters variations for each
175	# so the definitons of two weathers will run all variations in two cities
176	nameBase = self.nameBase	×
177	self.unscaledVariables = deepcopy(self.variablesOutput)	×
178	for i in range(len(self.sourceFilesToChange)):	×
179	originalSourceFile = self.sourceFilesToChange[i]	×
180	sourceFile = self.sourceFilesToChange[i]	×
181	for j in range(len(self.sinkFilesToChange[i])):	×
182	sinkFile = self.sinkFilesToChange[i][j]	×
183	self.changeDDckFile(sourceFile, sinkFile)	×
184	if "scalingVariable" in self.inputs.keys() and "scalingReference" in self.inputs.keys():	×
185	if "changeScalingFile" in self.inputs.keys():	×
186	self.scaleVariables(	×
187	self.inputs["scalingReference"],
188	self.inputs["changeScalingFile"][0],
189	self.inputs["changeScalingFile"][j + 1],
190	)
191	else:
192	self.scaleVariables(self.inputs["scalingReference"], originalSourceFile, sinkFile)	×
193
194	sourceFile = sinkFile # for each case the listddck will be changed to the new one, so we need to compare with the updated string	×
195
196	if self.foldersForDDckVariationUsed == True:	×
197	addFolder = self.foldersForDDckVariation[j]	×
198	originalPath = self.path	×
199	self.path = os.path.join(self.path, addFolder)	×
200	if not os.path.isdir(self.path):	×
201	os.mkdir(self.path)	×
202	else:
203	self.nameBase = nameBase + "-" + os.path.split(sinkFile)[-1]	×
204
205	result = self.buildTrnsysDeck()	×
206
207	if _res.isError(result):	×
208	error = _res.error(result)	×
209	self.logger.error(error.message)	×
210	return error	×
211
212	warnings = _res.value(result)	×
213	warnings.log(self.logger)	×
214
215	self.createDecksFromVariant()	×
216
217	if self.foldersForDDckVariationUsed:	×
218	self.path = originalPath # recall the original path, otherwise the next folder will be cerated inside the first	×
219	else:
220	result = self.buildTrnsysDeck()	1✔
221
222	if _res.isError(result):	1✔
223	error = _res.error(result)	×
224	self.logger.error(error.message)	×
225	return error	×
226
227	warnings = _res.value(result)	1✔
228	warnings.log(self.logger)	1✔
229
230	self.createDecksFromVariant()	1✔
231
232	def createDecksFromVariant(self, fitParameters={}):	1✔
233	variations = self.variablesOutput	1✔
234	parameters = self.parameters	1✔
235	parameters.update(fitParameters)	1✔
236
237	myDeckGenerator = createDeck.CreateTrnsysDeck(self.path, self.nameBase, variations)	1✔
238
239	myDeckGenerator.combineAllCases = self.inputs["combineAllCases"]	1✔
240
241	successfulCases = []	1✔
242
243	if "masterFile" in self.inputs:	1✔
244	if os.path.isfile(self.inputs["masterFile"]):	×
245	try:	×
246	masterDf = pd.read_csv(self.inputs["masterFile"], sep=";", index_col=0)	×
247	except:	×
248	self.logger.error("Unable to read " + self.inputs["masterFile"])	×
249	self.logger.error("Variation dck files of %s won't be created" % self.nameBase)	×
250	return	×
251
252	self.logger.info("Checking for successful runs in " + self.inputs["masterFile"])	×
253	for index, row in masterDf.iterrows():	×
254	if row["outcome"] == "success":	×
255	successfulCases.append(index)	×
256	else:
257	self.logger.info("Master file does not exist, no runs will be excluded")	×
258
259	# creates a list of decks with the appropriate name but nothing changed inside!!
260	if self.variationsUsed or (self.changeDDckFilesUsed == True and self.foldersForDDckVariationUsed == False):	1✔
261	if successfulCases:	×
262	fileName = myDeckGenerator.generateDecks(successfulCases=successfulCases)	×
263	else:
264	fileName = myDeckGenerator.generateDecks()	×
265	else:
266	fileName = []	1✔
267	fileName.append(self.nameBase)	1✔
268
269	if myDeckGenerator.noVariationCreated and self.variation:	1✔
270	self.logger.warning("No variation dck files created from " + self.nameBase)	×
271	return	×
272
273	tests = []	1✔
274
275	variablePath = self.path	1✔
276
277	for i in range(len(fileName)):	1✔
278	self.logger.debug("name to run :%s" % fileName[i])	1✔
279
280	# if useLocationStructure:
281	# variablePath = os.path.join(path,location) #assign subfolder for path
282
283	# Parameters changed by variation
284	localCopyPar = dict.copy(parameters) #	1✔
285
286	if self.variationsUsed:	1✔
287	myParameters = myDeckGenerator.getParameters(i)	×
288	localCopyPar.update(myParameters)	×
289
290	# We add to the global parameters that also need to be modified
291	# If we assign like localCopyPar = parameters, then the parameters will change with localCopyPar !!
292	# Otherwise we change the global parameter and some values of last variation will remain.
293
294	test = exeTrnsys.ExecuteTrnsys(variablePath, fileName[i])	1✔
295	tests.append(test)	1✔
296
297	test.setTrnsysExePath(self.inputs["trnsysExePath"])	1✔
298
299	test.setRemovePopUpWindow(self.inputs["removePopUpWindow"])	1✔
300
301	test.moveFileFromSource()	1✔
302
303	test.loadDeck(useDeckOutputPath=True)	1✔
304
305	test.changeParameter(localCopyPar)	1✔
306
307	test.changeAssignStatementsBasedOnUnitVariables(self._assignStatements)	1✔
308
309	if self.inputs["ignoreOnlinePlotter"] == True:	1✔
310	test.ignoreOnlinePlotter()	×
311
312	test.deckTrnsys.writeDeck()	1✔
313
314	test.cleanAndCreateResultsTempFolder()	1✔
315	test.moveFileFromSource()	1✔
316
317	test.clearOrCreateVariationDataFolder()	1✔
318	for pathToCopy in self._allPathsToCopyToVariationDataFolder:	1✔
319	test.copyPathToVariationDataFolder(pathToCopy.source, pathToCopy.target)	×
320
321	if self.inputs["runCases"] == True:	1✔
322	self.commands.append(test.getExecuteTrnsys(self.inputs))	1✔
323
324	def createLocationFolders(path, locations):	1✔
325	for location in locations:	×
326	if not os.path.exists(location):	×
327	os.makedirs(location)	×
328	print("created directory '") + path + location + "'"	×
329
330	def moveResultsFolder(path, resultsFolder, destinationFolder):	1✔
331	root_src_dir = os.path.join(path, resultsFolder)	×
332	root_target_dir = os.path.join(path, destinationFolder)	×
333
334	shutil.move(root_src_dir, root_target_dir)	×
335
336	def buildTrnsysDeck(self) -> _res.Result[_warn.ValueWithWarnings[str]]:	1✔
337	"""
338	It builds a TRNSYS Deck from a listDdck with pathDdck using the BuildingTrnsysDeck Class.
339	it reads the Deck list and writes a deck file. Afterwards it checks that the deck looks fine
340
341	"""
342	# I can create folders in another path to move them in the running folder and run them one by one
343	# path = "C:\Daten\OngoingProject\Ice-Ex\systemSimulations\\check\\"
344
345	deckExplanation = []	1✔
346	deckExplanation.append("! New deck built from list of ddcks. \n")	1✔
347	deck = _btd.BuildTrnsysDeck(	1✔
348	self.path,
349	self.nameBase,
350	self._includedDdckFiles,
351	self._defaultVisibility,
352	self._ddckPlaceHolderValuesJsonPath,
353	)
354	result = deck.readDeckList(	1✔
355	self.pathConfig,
356	doAutoUnitNumbering=self.inputs["doAutoUnitNumbering"],
357	dictPaths=self.dictDdckPaths,
358	replaceLineList=self.replaceLines,
359	)
360
361	if _res.isError(result):	1✔
362	return _res.error(result)	×
363	warnings = _res.value(result)	1✔
364
365	deck.overwriteForcedByUser = self.overwriteForcedByUser	1✔
366	deck.writeDeck(addedLines=deckExplanation)	1✔
367	self.overwriteForcedByUser = deck.overwriteForcedByUser	1✔
368
369	result = deck.checkTrnsysDeck(deck.nameDeck, check=self.inputs["checkDeck"])	1✔
370	if _res.isError(result):	1✔
371	return _res.error(result)	×
372
373	if self.inputs["generateUnitTypesUsed"] == True:	1✔
374	deck.saveUnitTypeFile()	1✔
375
376	if self.inputs["addAutomaticEnergyBalance"] == True:	1✔
377	deck.addAutomaticEnergyBalancePrinters()	1✔
378	deck.writeDeck() # Deck rewritten with added printer	1✔
379
380	deck.analyseDck()	1✔
381
382	return warnings.withValue(deck.nameDeck)	1✔
383
384	def runParallel(self, writeLogFile=True):	1✔
385	if writeLogFile:	×
386	self.writeRunLogFile()	×
387
388	if "masterFile" in self.inputs:	×
389	runPar.runParallel(	×
390	self.commands,
391	reduceCpu=int(self.inputs["reduceCpu"]),
392	trackingFile=self.inputs["trackingFile"],
393	masterFile=self.inputs["masterFile"],
394	)
395	elif "trackingFile" in self.inputs:	×
396	runPar.runParallel(	×
397	self.commands, reduceCpu=int(self.inputs["reduceCpu"]), trackingFile=self.inputs["trackingFile"]
398	)
399	else:
NEW 400	runPar.runParallel(	×
401	self.commands, reduceCpu=int(self.inputs["reduceCpu"]), outputFile=self.outputFileDebugRun
402	)
403
404	def writeRunLogFile(self):	1✔
405	logfile = open(os.path.join(self.path, "runLogFile.config"), "w")	×
406	username = os.getenv("username")	×
407	logfile.write("# Run created by " + username + "\n")	×
408	logfile.write("# Ddck repositories used:\n")	×
409	try:	×
410	import git	×
411
412	found = True	×
413	except ModuleNotFoundError:	×
414	found = False	×
415
416	for path in self.listDdckPaths:	×
417	logfile.write("# " + path + "\n")	×
418	logfile.write("# Revision Hash number: ")	×
419	if found:	×
420	try:	×
421	repo = git.Repo(path, search_parent_directories=True)	×
422	logfile.write(str(repo.head.object.hexsha) + "\n")	×
423	except git.exc.InvalidGitRepositoryError:	×
424	logfile.write("None - Not in a Git repository \n")	×
425	else:
426	logfile.write("not found \n")	×
427	logfile.write("\n# Config file used: \n\n")	×
428	for line in self.lines:	×
429	logfile.write(line + "\n")	×
430	logfile.close()	×
431
432	def readConfig(self, path, name, parseFileCreated=False):	1✔
433	"""
434	It reads the config file used for running TRNSYS and loads the self.inputs dictionary.
435	It also loads the readed lines into self.lines
436	"""
437	tool = readConfig.ReadConfigTrnsys()	1✔
438
439	self.lines = tool.readFile(path, name, self.inputs, parseFileCreated=parseFileCreated, controlDataType=False)	1✔
440
441	self.logger = log.getOrCreateCustomLogger("root", self.inputs["outputLevel"])	1✔
442
443	if "pathBaseSimulations" in self.inputs:	1✔
444	self.path = self.inputs["pathBaseSimulations"]	×
445	if self.inputs["addResultsFolder"]:	1✔
446	self.path = os.path.join(self.path, self.inputs["addResultsFolder"])	1✔
447
448	if not os.path.isdir(self.path):	1✔
449	os.mkdir(self.path)	1✔
450
451	def changeFile(self, source, end):	1✔
452	"""
453	It uses the self-lines readed by readConfig and change the lines from source to end.
454	This is used to change a ddck file readed for another. A typical example is the weather data file
455	Parameters
456	----------
457	source : str
458	string to be replaced in the config file in the self.lines field
459	end : str
460	str to replace the source in the config file in the self.lines field
461
462	Returns
463	-------
464
465	"""
466
467	found = False	×
468	for i in range(len(self.lines)):	×
469	lineFilter = self.lines[i]	×
470
471	if lineFilter == source:	×
472	self.lines[i] = end	×
473	found = True	×
474
475	if found == False:	×
476	self.logger.warning("change File was not able to change %s by %s" % (source, end))	×
477
478	def changeDDckFile(self, source, end):	1✔
479	"""
480	It uses the self.listDdck readed by readConfig and change the lines from source to end.
481	This is used to change a ddck file readed for another. A typical example is the weather data file
482	"""
483	found = False	×
484	nCharacters = len(source)	×
485
486	for i in range(len(self._includedDdckFiles)):	×
487	oldIncludedDdckFile = self._includedDdckFiles[i]	×
488
489	ddckFilePath = str(oldIncludedDdckFile.pathWithoutSuffix)	×
490
491	mySource = ddckFilePath[-nCharacters:] # I read only the last characters with the same size as the end file	×
492	if mySource == source:	×
493	newDdckFilePath = ddckFilePath[0:-nCharacters] + end	×
494	self.dictDdckPaths[newDdckFilePath] = self.dictDdckPaths[ddckFilePath]	×
495	newIncludedDdckFile = _idf.IncludedDdckFile(	×
496	_pl.Path(newDdckFilePath), oldIncludedDdckFile.componentName, oldIncludedDdckFile.defaultVisibility
497	)
498	self._includedDdckFiles[i] = newIncludedDdckFile	×
499
500	found = True	×
501
502	if not found:	×
503	self.logger.warning("change File was not able to change %s by %s" % (source, end))	×
504
505	def copyConfigFile(self, configPath, configName):	1✔
506	configFile = os.path.join(configPath, configName)	×
507	dstPath = os.path.join(configPath, self.inputs["addResultsFolder"], configName)	×
508	shutil.copyfile(configFile, dstPath)	×
509	self.logger.debug("copied config file to: %s" % dstPath)	×
510
511	def scaleVariables(self, reference, source, sink):	1✔
512	"""
513
514	Parameters
515	----------
516	reference : str
517	File path of the reference results file. Has to point to a json-File with the pytrnsys resuls file format
518	source : str
519	Substring to be replaced in reference
520	sink : str
521	String to replace the substrings in the reference
522
523	Returns
524	-------
525
526	"""
527	resultFile = reference.replace(source, sink)	×
528	with open(resultFile) as f_in:	×
529	resultsDict = json.load(f_in)	×
530
531	exec("scalingVariable=" + self.inputs["scalingVariable"], globals(), resultsDict)	×
532	loadDemand = resultsDict["scalingVariable"]	×
533
534	try:	×
535	exec("scalingElDemandVariable=" + self.inputs["scalingElDemandVariable"], globals(), resultsDict)	×
536	loadElDemand = resultsDict["scalingElDemandVariable"]	×
537	except:	×
538	pass	×
539
540	try:	×
541	exec("scaleHP=" + self.inputs["scaleHP"], globals(), resultsDict)	×
542	loadHPsize = resultsDict["scaleHP"]	×
543	except:	×
544	pass	×
545
546	for j in range(len(self.variablesOutput)):	×
547	for i in range(2, len(self.variablesOutput[j]), 1):	×
548	if self.variablesOutput[j][1] == "sizeHpUsed":	×
549	self.variablesOutput[j][i] = (	×
550	str(round(self.unscaledVariables[j][i], 3)) + "*" + str(round(loadHPsize, 3))
551	)
552	elif self.variablesOutput[j][1] == "AreaPvRoof":	×
553	self.variablesOutput[j][i] = (	×
554	str(round(self.unscaledVariables[j][i], 3)) + "*" + str(round(loadElDemand, 3))
555	)
556	else:
557	self.variablesOutput[j][i] = (	×
558	str(round(self.unscaledVariables[j][i], 3)) + "*" + str(round(loadDemand, 3))
559	)

SPF-OST / pytrnsys / 15605215655

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