11662562960

Committed 29 Oct 2024 03:09PM UTC coverage: 67.508% (-0.08%) from 67.591%

Build # 11662562960

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #564 from SPF-OST/560-black-change-line-length-to-pep8-standard-of-79-and-check-ci-reaction

changed line length in black to 79

Run Details

1054 of 1475 new or added lines in 174 files covered. (71.46%)

150 existing lines in 74 files now uncovered.

10399 of 15404 relevant lines covered (67.51%)

0.68 hits per line

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

81.9

/trnsysGUI/MassFlowVisualizer.py

# pylint: skip-file
# type: ignore

import datetime as _dt
import itertools as _it
import numpy as _np
import typing as _tp

import pandas as _pd

import PyQt5.QtCore as _qtc
import PyQt5.QtWidgets as _qtw

import trnsysGUI.TVentil as _tv
import trnsysGUI.connection.connectionBase as _cb
import trnsysGUI.connection.doublePipeConnection as _dpc
import trnsysGUI.connection.names as _cnames
import trnsysGUI.connection.singlePipeConnection as _spc
import trnsysGUI.warningsAndErrors as _werrors
import trnsysGUI.massFlowSolver.names as _mnames
import trnsysGUI.massFlowSolver.networkModel as _mfn

_MAX_HEADER_LENGTH = 25


class MassFlowVisualizer(_qtw.QDialog):
    def __init__(self, parent, mfrFile, tempFile):

        super(MassFlowVisualizer, self).__init__(parent)

        self.logger = parent.logger

        self.dataFilePath = mfrFile
        self.tempDatafilePath = tempFile
        self.loadedFile = False
        self.tempLoadedFile = False
        self.loadFile()
        self.loadTempFile()
        self.maxTimeStep = len(self.massFlowData.index) - 1
        self.showMass = False

        self.setMinimumSize(1000, 200)

        self.parent = parent
        self.timeStep = 0
        self.timeSteps = len(self.massFlowData.index) - 1  # 0 to 364

        # threshold values for positive list
        self.medianValue = 0
        self.lowerQuarter = 0
        self.upperQuarter = 0
        self.minValue = 0
        self.maxValue = 0

        self.medianValueMfr = 0
        self.lowerQuarterMfr = 0
        self.upperQuarterMfr = 0
        self.minValueMfr = 0
        self.maxValueMfr = 0

        self.getThresholdValues()
        self.getTempThresholdValues()

        timeStepIncreaseLabelText = "Increase time-step by ( 30 = 1 Hour ):\n              ( 720 = 1 Day )"
        self._timeStepIncreaseLabel = _qtw.QLabel(timeStepIncreaseLabelText)
        self._timeStepIncrease = 1
        self._timeStepIncreaseLineEdit = _qtw.QLineEdit()
        self._timeStepIncreaseLineEdit.setText(str(self._timeStepIncrease))
        self._timeStepIncreaseLineEdit.editingFinished.connect(
            self._onTimeStepIncreaseLineEditEditingFinished
        )

        self.slider = _qtw.QSlider(parent)
        self.setSlider()
        self.slider.sliderReleased.connect(self.testValChange)
        self.slider.sliderPressed.connect(self.pressedSlider)
        self.slider.valueChanged.connect(self.moveValues)
        self.slider.setTickInterval(24)

        self.qtm = _qtc.QTimer(parent)
        self.lines = None
        self.started = False

        self.paused = True

        nameLabel = _qtw.QLabel("Name:")
        self.currentStepLabel = _qtw.QLabel(
            "Time: " + str(self.convertTime(self.getTime(0)))
        )
        self.le = _qtw.QLineEdit("NONE")

        self.showMassButton = _qtw.QPushButton("Show mass")  # comment out
        self.togglePauseButton = _qtw.QPushButton("Pause/Continue")
        self.cancelButton = _qtw.QPushButton("Cancel")

        minColorLabel = _qtw.QLabel("Min")
        minColorLabel.setStyleSheet("color: blue")
        maxColorLabel = _qtw.QLabel("Max")
        maxColorLabel.setStyleSheet("color: red")
        minToFiftyPercentile = _qtw.QLabel("Min to 50%")
        minToFiftyPercentile.setStyleSheet("color: cyan")
        fiftyPercentileToMax = _qtw.QLabel("50% to max")
        fiftyPercentileToMax.setStyleSheet("color: pink")
        arrowLabel = _qtw.QLabel("-->")
        arrowLabel2 = _qtw.QLabel("-->")
        arrowLabel3 = _qtw.QLabel("-->")

        colorLayout = _qtw.QHBoxLayout()
        colorLayout.addWidget(minColorLabel)
        colorLayout.addWidget(arrowLabel)
        colorLayout.addWidget(minToFiftyPercentile)
        colorLayout.addWidget(arrowLabel2)
        colorLayout.addWidget(fiftyPercentileToMax)
        colorLayout.addWidget(arrowLabel3)
        colorLayout.addWidget(maxColorLabel)
        colorLayout.addStretch()

        buttonLayout = _qtw.QHBoxLayout()
        buttonLayout.addStretch()
        buttonLayout.addWidget(self._timeStepIncreaseLabel)
        buttonLayout.addWidget(self._timeStepIncreaseLineEdit)
        buttonLayout.addWidget(self.showMassButton)  # comment out
        buttonLayout.addWidget(self.togglePauseButton)
        buttonLayout.addWidget(self.cancelButton)
        layout = _qtw.QGridLayout()
        layout.addWidget(nameLabel, 0, 0)
        layout.addWidget(self.le, 0, 1)
        layout.addLayout(colorLayout, 1, 0, 1, 0)
        layout.addLayout(buttonLayout, 2, 0, 2, 0)
        layout.addWidget(self.currentStepLabel, 3, 0, 1, 2)
        layout.addWidget(
            self.slider, 4, 0, 1, 2
        )  # Only for debug (Why do I need a 3 here instead of a 2 for int:row?)

        self.setLayout(layout)

        self.showMassButton.clicked.connect(self.showMassBtn)  # comment out
        self.togglePauseButton.clicked.connect(self.togglePause)
        self.cancelButton.clicked.connect(self.cancel)

        self.advance()

        self.setWindowTitle("Flow visualizer")
        ph = parent.geometry().height()
        pw = parent.geometry().width()
        px = parent.geometry().x()
        py = parent.geometry().y()
        dw = self.width()
        dh = self.height()
        self.move(parent.editor.diagramView.geometry().topLeft())
        self.show()

    def _onTimeStepIncreaseLineEditEditingFinished(self) -> None:
        try:
            lineEditText = self._timeStepIncreaseLineEdit.text()
            timeStepIncrease = int(lineEditText)
        except ValueError:
            timeStepIncrease = None

        if timeStepIncrease is None or not (
            1 <= timeStepIncrease <= self.maxTimeStep
        ):
            errorMessage = f"The time-step increase must be an integer between 1 and {self.maxTimeStep}, inclusive."
            _werrors.showMessageBox(errorMessage, title="Invalid value")
            self._timeStepIncreaseLineEdit.setText(str(self._timeStepIncrease))
            return

        self._timeStepIncrease = timeStepIncrease

    def togglePause(self):
        if self.paused:
            self.continueVis()
        else:
            self.pauseVis()

    def cancel(self):
        self.pauseVis()
        self.close()
        self.parent.editor.updateConnGrads()

    # comment out
    def showMassBtn(self):
        """
        For showing the mass under the connections
        -------

        """
        self.showMass = not self.showMass

        for t in self.parent.editor.trnsysObj:
            if isinstance(t, _cb.ConnectionBase):
                t.massFlowLabel.setVisible(self.showMass)

        self.logger.debug(
            "%s %s %s"
            % (str(self.minValue), str(self.medianValue), str(self.maxValue))
        )

    def loadFile(self):
        if not self.loadedFile:
            self.massFlowData = _pd.read_csv(
                self.dataFilePath, sep="\t"
            ).rename(columns=lambda x: x.strip())
            _truncateColumnNames(self.massFlowData)
        self.loadedFile = True

    def loadTempFile(self):
        if not self.tempLoadedFile:
            self.tempMassFlowData = _pd.read_csv(
                self.tempDatafilePath, sep="\t"
            ).rename(columns=lambda x: x.strip())
            _truncateColumnNames(self.tempMassFlowData)
        self.tempLoadedFile = True

    def start(self):

        self.paused = False
        self.qtm = _qtc.QTimer(self.parent)
        self.qtm.timeout.connect(self.advance)
        self.qtm.timeout.connect(self.increaseValue)
        self.qtm.start(1000)

    def advance(self):
        if self.timeStep == self.maxTimeStep:
            self.logger.debug("reached end of data, returning")
            self.qtm.stop()

        if self.loadedFile:
            i = 0
            for t in self.parent.editor.trnsysObj:
                if isinstance(t, _spc.SinglePipeConnection):
                    mfrVariableName = _mnames.getCanonicalMassFlowVariableName(
                        componentDisplayName=t.getDisplayName(),
                        pipeName=t.modelPipe.name,
                    )
                    temperatureVariableName = (
                        _cnames.getTemperatureVariableName(
                            t, _mfn.PortItemType.STANDARD
                        )
                    )

                    if (
                        mfrVariableName in self.massFlowData.columns
                        and temperatureVariableName in self.tempMassFlowData
                    ):
                        massFlow = self._getMassFlow(
                            mfrVariableName, self.timeStep
                        )
                        temperature = self._getTemperature(
                            temperatureVariableName, self.timeStep
                        )

                        t.setMassFlowAndTemperature(massFlow, temperature)
                        thickValue = self.getThickness(massFlow)
                        self.logger.debug("Thickvalue: " + str(thickValue))
                        if (
                            self.massFlowData[mfrVariableName].iloc[
                                self.timeStep
                            ]
                            == 0
                        ):
                            t.setColor(thickValue, mfr="ZeroMfr")
                        elif round(abs(temperature)) == self.maxValue:
                            t.setColor(thickValue, mfr="max")
                        elif round(abs(temperature)) == self.minValue:
                            t.setColor(thickValue, mfr="min")
                        elif (
                            self.minValue
                            < round(abs(temperature))
                            <= self.lowerQuarter
                        ):
                            t.setColor(thickValue, mfr="minTo25")
                        elif (
                            self.lowerQuarter
                            < round(abs(temperature))
                            <= self.medianValue
                        ):
                            t.setColor(thickValue, mfr="25To50")
                        elif (
                            self.medianValue
                            < round(abs(temperature))
                            <= self.upperQuarter
                        ):
                            t.setColor(thickValue, mfr="50To75")
                        elif (
                            self.upperQuarter
                            < round(abs(temperature))
                            < self.maxValue
                        ):
                            t.setColor(thickValue, mfr="75ToMax")
                        else:
                            t.setColor(thickValue, mfr="test")
                        i += 1
                if isinstance(t, _dpc.DoublePipeConnection):
                    coldMassFlowVariableName = (
                        _mnames.getCanonicalMassFlowVariableName(
                            componentDisplayName=t.displayName,
                            pipeName=t.coldModelPipe.name,
                        )
                    )
                    coldTemperatureVariableName = (
                        _cnames.getTemperatureVariableName(
                            t, _mfn.PortItemType.COLD
                        )
                    )

                    hotMassFlowVariableName = (
                        _mnames.getCanonicalMassFlowVariableName(
                            componentDisplayName=t.displayName,
                            pipeName=t.hotModelPipe.name,
                        )
                    )
                    hotTemperatureVariableName = (
                        _cnames.getTemperatureVariableName(
                            t, _mfn.PortItemType.HOT
                        )
                    )

                    coldMassFlow = self._getMassFlow(
                        coldMassFlowVariableName, self.timeStep
                    )
                    coldTemperature = self._getTemperature(
                        coldTemperatureVariableName, self.timeStep
                    )

                    hotMassFlow = self._getMassFlow(
                        hotMassFlowVariableName, self.timeStep
                    )
                    hotTemperature = self._getTemperature(
                        hotTemperatureVariableName, self.timeStep
                    )

                    t.setMassFlowAndTemperature(
                        coldMassFlow,
                        coldTemperature,
                        hotMassFlow,
                        hotTemperature,
                    )

                elif isinstance(t, _tv.TVentil):
                    valvePositionVariableName = _mnames.getInputVariableName(
                        t, t.modelDiverter
                    )
                    if valvePositionVariableName in self.massFlowData.columns:
                        valvePosition = str(
                            self.massFlowData[valvePositionVariableName].iloc[
                                self.timeStep
                            ]
                        )
                        t.setPositionForMassFlowSolver(valvePosition)
                        t.posLabel.setPlainText(valvePosition)
                        self.logger.debug(
                            "valve position: " + str(valvePosition)
                        )

        else:
            return

    def _getMassFlow(self, mfrVariableName: str, timeStep: int) -> float:
        truncatedMfrVariableName = _truncateName(mfrVariableName)
        mass = self.massFlowData[truncatedMfrVariableName].iloc[timeStep]
        return mass

    def _getTemperature(
        self, temperatureVariableName: str, timeStep: int
    ) -> float:
        truncatedTemperatureVariableName = _truncateName(
            temperatureVariableName
        )
        return self.tempMassFlowData[truncatedTemperatureVariableName].iloc[
            timeStep
        ]

    def pauseVis(self):
        self.paused = True
        self.qtm.stop()

    def continueVis(self):
        if self.started:
            self.paused = False
            self.qtm.start(1000)
        else:
            self.start()

    def setSlider(self):
        self.slider.setOrientation(1)
        self.slider.setMinimum(0)
        self.slider.setMaximum(self.timeSteps)
        self.slider.setTickInterval(1)
        self.slider.setTickPosition(2)
        if self.checkTimeStep and self.checkTempTimeStep():
            self.slider.setVisible(True)
            self.slider.setEnabled(False)
            self.increaseValue()
        else:
            self.slider.setVisible(True)
            self.slider.setEnabled(True)

    def testValChange(self):
        """
        Updates mass flow visualizer after releasing the mouse
        """
        val = self.slider.value()
        self.logger.debug("Slider value has changed to " + str(val))
        time = self.getTime(val)
        self.currentStepLabel.setText("Time :" + str(self.convertTime(time)))
        self.timeStep = val
        self.advance()

    def moveValues(self):
        """
        Updates mass flow visualizer when moving the mouse
        """
        val = self.slider.value()
        self.logger.debug("Slider value is still: " + str(val))
        time = self.getTime(val)
        self.currentStepLabel.setText("Time :" + str(self.convertTime(time)))
        self.timeStep = val
        self.advance()

    def increaseValue(self):
        """
        For automatic slider movement

        """

        self.timeStep += self._timeStepIncrease
        if self.timeStep > self.maxTimeStep:
            self.timeStep = 0
        self.slider.setValue(self.timeStep)

    def decreaseValue(self):
        """
        For automatic slider movement

        """

        self.timeStep -= self._timeStepIncrease
        if self.timeStep < 0:
            self.timeStep = self.maxTimeStep
        self.slider.setValue(self.timeStep)

    def checkTimeStep(self):
        """
        Check individual columns of the data frame, If a column has rows with different values, return False.
        Else, continue to next column. Return True if no such column can be found.

        False indicates the rows are not identical and slider should be enabled
        True indicates the rows are identical and slider should be disabled

        """
        massFlowDataDup = self.massFlowData
        massFlowDataDup = massFlowDataDup.drop(massFlowDataDup.index[0])
        for items in round(abs(massFlowDataDup)).nunique().iteritems():
            if items[0] != "TIME":
                if items[1] > 1:
                    return False
        return True

    def checkTempTimeStep(self):
        tempMassFlowDataDup = self.tempMassFlowData
        tempMassFlowDataDup = tempMassFlowDataDup.drop(
            tempMassFlowDataDup.index[0]
        )
        for items in tempMassFlowDataDup.nunique().items():
            if items[0] != "TIME":
                if items[1] > 1:
                    return False
        return True

    def getThresholdValues(self):
        """
        Access the data frame, convert into a nested list.
        Merge the nested list into one single list.
        Remove 'nan' from the list.
        Convert the values inside the list into absolute values then round them off.
        Split the list into two for negative and positive values.
        Get the minimum value, the 25th percentile, the median value, the 75th percentile and
        the max value from the lists.
        Returns
        -------
        """

        data = (
            self.massFlowData.values.tolist()
        )  # data frame converted to nested list
        for sublist in data:  # delete the time column from the list
            del sublist[0]
        data = list(
            _it.chain.from_iterable(data)
        )  # nested list combined into one list
        cleanedData = [
            x for x in data if str(x) != "nan"
        ]  # remove nan from list
        cleanedData = [
            round(abs(num)) for num in cleanedData
        ]  # get absolute value and round off
        noDuplicateData = list(dict.fromkeys(cleanedData))

        self.medianValueMfr = _np.percentile(
            noDuplicateData, 50
        )  # median value / 50th percentile
        self.lowerQuarterMfr = _np.percentile(
            noDuplicateData, 25
        )  # 25th percentile
        self.upperQuarterMfr = _np.percentile(
            noDuplicateData, 75
        )  # 75th percentile
        self.minValueMfr = _np.min(
            noDuplicateData
        )  # minimum value excluding 0
        self.maxValueMfr = _np.max(noDuplicateData)  # max value

    def getThickness(self, mass: float) -> int:
        mass = abs(mass)
        if mass == self.minValueMfr:
            return 2
        elif self.minValueMfr < mass <= self.medianValueMfr:
            return 4
        elif self.medianValueMfr < mass < self.maxValueMfr:
            return 6
        elif mass == self.maxValueMfr:
            return 8
        else:
            return 2

    def getTempThresholdValues(self):
        """
        Access the data frame, convert into a nested list.
        Merge the nested list into one single list.
        Remove 'nan' from the list.
        Convert the values inside the list into absolute values then round them off.
        Split the list into two for negative and positive values.
        Get the minimum value, the 25th percentile, the median value, the 75th percentile and
        the max value from the lists.
        Returns
        -------
        """

        data = (
            self.tempMassFlowData.values.tolist()
        )  # data frame converted to nested list
        for sublist in data:  # delete the time column from the list
            del sublist[0]
        data = list(
            _it.chain.from_iterable(data)
        )  # nested list combined into one list
        cleanedData = [
            x for x in data if str(x) != "nan"
        ]  # remove nan from list
        cleanedData = [
            round(abs(num)) for num in cleanedData
        ]  # get absolute value and round off
        noDuplicateData = list(dict.fromkeys(cleanedData))

        self.medianValue = _np.percentile(
            noDuplicateData, 50
        )  # median value / 50th percentile
        self.lowerQuarter = _np.percentile(
            noDuplicateData, 25
        )  # 25th percentile
        self.upperQuarter = _np.percentile(
            noDuplicateData, 75
        )  # 75th percentile
        self.minValue = _np.min(noDuplicateData)  # minimum value excluding 0
        self.maxValue = _np.max(noDuplicateData)  # max value

        # print(noDuplicateData)
        # sys.exit()

    def getTime(self, row):
        """
        Gets the time of the current time step
        """
        data = self.massFlowData
        timeColumn = data[data.columns[0]]
        self.logger.debug(str(timeColumn[row]))
        return timeColumn[row]

    def convertTime(self, time):
        """
        Convert the time into YYYY--MM--DD HH:MM:SS format
        """
        noOfHours = 8760
        decHour = float(time) / float(noOfHours)
        base = _dt.datetime(_dt.MINYEAR, 1, 1)
        result = base + _dt.timedelta(
            seconds=(base.replace(year=base.year + 1) - base).total_seconds()
            * decHour
        )
        return str(result)

    def pressedSlider(self):
        self.pauseVis()

    def closeEvent(self, a0):
        self.pauseVis()
        super().closeEvent(a0)

    def keyPressEvent(self, e):
        if e.key() == _qtc.Qt.Key_Up:
            self.logger.debug("Up is pressed")
            self.increaseValue()
        elif e.key() == _qtc.Qt.Key_Down:
            self.logger.debug("Down is pressed")
            self.decreaseValue()


def _truncateColumnNames(df: _pd.DataFrame) -> None:
    _ensureNamesDontCollideAfterTruncating(df.columns)
    df.columns = [_truncateName(n) for n in df.columns]


def _ensureNamesDontCollideAfterTruncating(
    columnNames: _tp.Sequence[str],
) -> None:
    sortedColumnNames = sorted(columnNames)
    groupedNames = [
        list(g) for _, g in _it.groupby(sortedColumnNames, key=_truncateName)
    ]
    collidingNames = _flatten(g for g in groupedNames if len(g) > 1)
    if collidingNames:
        formattedCollidingNames = "\n\t".join(collidingNames)
        errorMessage = (
            f"The following column names collide after truncating them to "
            f"{_MAX_HEADER_LENGTH} characters:\n\t{formattedCollidingNames}"
        )
        raise ValueError(errorMessage)


def _truncateName(name: str):
    return name[:_MAX_HEADER_LENGTH]


def _flatten(iterable: _tp.Iterable[_tp.Iterable[str]]) -> _tp.Sequence[str]:
    return list(_it.chain.from_iterable(iterable))

1	# pylint: skip-file
2	# type: ignore
3
4	import datetime as _dt	1✔
5	import itertools as _it	1✔
6	import numpy as _np	1✔
7	import typing as _tp	1✔
8
9	import pandas as _pd	1✔
10
11	import PyQt5.QtCore as _qtc	1✔
12	import PyQt5.QtWidgets as _qtw	1✔
13
14	import trnsysGUI.TVentil as _tv	1✔
15	import trnsysGUI.connection.connectionBase as _cb	1✔
16	import trnsysGUI.connection.doublePipeConnection as _dpc	1✔
17	import trnsysGUI.connection.names as _cnames	1✔
18	import trnsysGUI.connection.singlePipeConnection as _spc	1✔
19	import trnsysGUI.warningsAndErrors as _werrors	1✔
20	import trnsysGUI.massFlowSolver.names as _mnames	1✔
21	import trnsysGUI.massFlowSolver.networkModel as _mfn	1✔
22
23	_MAX_HEADER_LENGTH = 25	1✔
24
25
26	class MassFlowVisualizer(_qtw.QDialog):	1✔
27	def __init__(self, parent, mfrFile, tempFile):	1✔
28
29	super(MassFlowVisualizer, self).__init__(parent)	1✔
30
31	self.logger = parent.logger	1✔
32
33	self.dataFilePath = mfrFile	1✔
34	self.tempDatafilePath = tempFile	1✔
35	self.loadedFile = False	1✔
36	self.tempLoadedFile = False	1✔
37	self.loadFile()	1✔
38	self.loadTempFile()	1✔
39	self.maxTimeStep = len(self.massFlowData.index) - 1	1✔
40	self.showMass = False	1✔
41
42	self.setMinimumSize(1000, 200)	1✔
43
44	self.parent = parent	1✔
45	self.timeStep = 0	1✔
46	self.timeSteps = len(self.massFlowData.index) - 1 # 0 to 364	1✔
47
48	# threshold values for positive list
49	self.medianValue = 0	1✔
50	self.lowerQuarter = 0	1✔
51	self.upperQuarter = 0	1✔
52	self.minValue = 0	1✔
53	self.maxValue = 0	1✔
54
55	self.medianValueMfr = 0	1✔
56	self.lowerQuarterMfr = 0	1✔
57	self.upperQuarterMfr = 0	1✔
58	self.minValueMfr = 0	1✔
59	self.maxValueMfr = 0	1✔
60
61	self.getThresholdValues()	1✔
62	self.getTempThresholdValues()	1✔
63
64	timeStepIncreaseLabelText = "Increase time-step by ( 30 = 1 Hour ):\n ( 720 = 1 Day )"	1✔
65	self._timeStepIncreaseLabel = _qtw.QLabel(timeStepIncreaseLabelText)	1✔
66	self._timeStepIncrease = 1	1✔
67	self._timeStepIncreaseLineEdit = _qtw.QLineEdit()	1✔
68	self._timeStepIncreaseLineEdit.setText(str(self._timeStepIncrease))	1✔
69	self._timeStepIncreaseLineEdit.editingFinished.connect(	1✔
70	self._onTimeStepIncreaseLineEditEditingFinished
71	)
72
73	self.slider = _qtw.QSlider(parent)	1✔
74	self.setSlider()	1✔
75	self.slider.sliderReleased.connect(self.testValChange)	1✔
76	self.slider.sliderPressed.connect(self.pressedSlider)	1✔
77	self.slider.valueChanged.connect(self.moveValues)	1✔
78	self.slider.setTickInterval(24)	1✔
79
80	self.qtm = _qtc.QTimer(parent)	1✔
81	self.lines = None	1✔
82	self.started = False	1✔
83
84	self.paused = True	1✔
85
86	nameLabel = _qtw.QLabel("Name:")	1✔
87	self.currentStepLabel = _qtw.QLabel(	1✔
88	"Time: " + str(self.convertTime(self.getTime(0)))
89	)
90	self.le = _qtw.QLineEdit("NONE")	1✔
91
92	self.showMassButton = _qtw.QPushButton("Show mass") # comment out	1✔
93	self.togglePauseButton = _qtw.QPushButton("Pause/Continue")	1✔
94	self.cancelButton = _qtw.QPushButton("Cancel")	1✔
95
96	minColorLabel = _qtw.QLabel("Min")	1✔
97	minColorLabel.setStyleSheet("color: blue")	1✔
98	maxColorLabel = _qtw.QLabel("Max")	1✔
99	maxColorLabel.setStyleSheet("color: red")	1✔
100	minToFiftyPercentile = _qtw.QLabel("Min to 50%")	1✔
101	minToFiftyPercentile.setStyleSheet("color: cyan")	1✔
102	fiftyPercentileToMax = _qtw.QLabel("50% to max")	1✔
103	fiftyPercentileToMax.setStyleSheet("color: pink")	1✔
104	arrowLabel = _qtw.QLabel("-->")	1✔
105	arrowLabel2 = _qtw.QLabel("-->")	1✔
106	arrowLabel3 = _qtw.QLabel("-->")	1✔
107
108	colorLayout = _qtw.QHBoxLayout()	1✔
109	colorLayout.addWidget(minColorLabel)	1✔
110	colorLayout.addWidget(arrowLabel)	1✔
111	colorLayout.addWidget(minToFiftyPercentile)	1✔
112	colorLayout.addWidget(arrowLabel2)	1✔
113	colorLayout.addWidget(fiftyPercentileToMax)	1✔
114	colorLayout.addWidget(arrowLabel3)	1✔
115	colorLayout.addWidget(maxColorLabel)	1✔
116	colorLayout.addStretch()	1✔
117
118	buttonLayout = _qtw.QHBoxLayout()	1✔
119	buttonLayout.addStretch()	1✔
120	buttonLayout.addWidget(self._timeStepIncreaseLabel)	1✔
121	buttonLayout.addWidget(self._timeStepIncreaseLineEdit)	1✔
122	buttonLayout.addWidget(self.showMassButton) # comment out	1✔
123	buttonLayout.addWidget(self.togglePauseButton)	1✔
124	buttonLayout.addWidget(self.cancelButton)	1✔
125	layout = _qtw.QGridLayout()	1✔
126	layout.addWidget(nameLabel, 0, 0)	1✔
127	layout.addWidget(self.le, 0, 1)	1✔
128	layout.addLayout(colorLayout, 1, 0, 1, 0)	1✔
129	layout.addLayout(buttonLayout, 2, 0, 2, 0)	1✔
130	layout.addWidget(self.currentStepLabel, 3, 0, 1, 2)	1✔
131	layout.addWidget(	1✔
132	self.slider, 4, 0, 1, 2
133	) # Only for debug (Why do I need a 3 here instead of a 2 for int:row?)
134
135	self.setLayout(layout)	1✔
136
137	self.showMassButton.clicked.connect(self.showMassBtn) # comment out	1✔
138	self.togglePauseButton.clicked.connect(self.togglePause)	1✔
139	self.cancelButton.clicked.connect(self.cancel)	1✔
140
141	self.advance()	1✔
142
143	self.setWindowTitle("Flow visualizer")	1✔
144	ph = parent.geometry().height()	1✔
145	pw = parent.geometry().width()	1✔
146	px = parent.geometry().x()	1✔
147	py = parent.geometry().y()	1✔
148	dw = self.width()	1✔
149	dh = self.height()	1✔
150	self.move(parent.editor.diagramView.geometry().topLeft())	1✔
151	self.show()	1✔
152
153	def _onTimeStepIncreaseLineEditEditingFinished(self) -> None:	1✔
154	try:	×
155	lineEditText = self._timeStepIncreaseLineEdit.text()	×
156	timeStepIncrease = int(lineEditText)	×
157	except ValueError:	×
158	timeStepIncrease = None	×
159
NEW 160	if timeStepIncrease is None or not (	×
161	1 <= timeStepIncrease <= self.maxTimeStep
162	):
163	errorMessage = f"The time-step increase must be an integer between 1 and {self.maxTimeStep}, inclusive."	×
164	_werrors.showMessageBox(errorMessage, title="Invalid value")	×
165	self._timeStepIncreaseLineEdit.setText(str(self._timeStepIncrease))	×
166	return	×
167
168	self._timeStepIncrease = timeStepIncrease	×
169
170	def togglePause(self):	1✔
171	if self.paused:	×
172	self.continueVis()	×
173	else:
174	self.pauseVis()	×
175
176	def cancel(self):	1✔
177	self.pauseVis()	×
178	self.close()	×
179	self.parent.editor.updateConnGrads()	×
180
181	# comment out
182	def showMassBtn(self):	1✔
183	"""
184	For showing the mass under the connections
185	-------
186
187	"""
188	self.showMass = not self.showMass	×
189
190	for t in self.parent.editor.trnsysObj:	×
191	if isinstance(t, _cb.ConnectionBase):	×
192	t.massFlowLabel.setVisible(self.showMass)	×
193
NEW 194	self.logger.debug(	×
195	"%s %s %s"
196	% (str(self.minValue), str(self.medianValue), str(self.maxValue))
197	)
198
199	def loadFile(self):	1✔
200	if not self.loadedFile:	1✔
201	self.massFlowData = _pd.read_csv(	1✔
202	self.dataFilePath, sep="\t"
203	).rename(columns=lambda x: x.strip())
204	_truncateColumnNames(self.massFlowData)	1✔
205	self.loadedFile = True	1✔
206
207	def loadTempFile(self):	1✔
208	if not self.tempLoadedFile:	1✔
209	self.tempMassFlowData = _pd.read_csv(	1✔
210	self.tempDatafilePath, sep="\t"
211	).rename(columns=lambda x: x.strip())
212	_truncateColumnNames(self.tempMassFlowData)	1✔
213	self.tempLoadedFile = True	1✔
214
215	def start(self):	1✔
216
217	self.paused = False	×
218	self.qtm = _qtc.QTimer(self.parent)	×
219	self.qtm.timeout.connect(self.advance)	×
220	self.qtm.timeout.connect(self.increaseValue)	×
221	self.qtm.start(1000)	×
222
223	def advance(self):	1✔
224	if self.timeStep == self.maxTimeStep:	1✔
225	self.logger.debug("reached end of data, returning")	1✔
226	self.qtm.stop()	1✔
227
228	if self.loadedFile:	1✔
229	i = 0	1✔
230	for t in self.parent.editor.trnsysObj:	1✔
231	if isinstance(t, _spc.SinglePipeConnection):	1✔
232	mfrVariableName = _mnames.getCanonicalMassFlowVariableName(	1✔
233	componentDisplayName=t.getDisplayName(),
234	pipeName=t.modelPipe.name,
235	)
236	temperatureVariableName = (	1✔
237	_cnames.getTemperatureVariableName(
238	t, _mfn.PortItemType.STANDARD
239	)
240	)
241
242	if (	1✔
243	mfrVariableName in self.massFlowData.columns
244	and temperatureVariableName in self.tempMassFlowData
245	):
246	massFlow = self._getMassFlow(	1✔
247	mfrVariableName, self.timeStep
248	)
249	temperature = self._getTemperature(	1✔
250	temperatureVariableName, self.timeStep
251	)
252
253	t.setMassFlowAndTemperature(massFlow, temperature)	1✔
254	thickValue = self.getThickness(massFlow)	1✔
255	self.logger.debug("Thickvalue: " + str(thickValue))	1✔
256	if (	1✔
257	self.massFlowData[mfrVariableName].iloc[
258	self.timeStep
259	]
260	== 0
261	):
262	t.setColor(thickValue, mfr="ZeroMfr")	1✔
263	elif round(abs(temperature)) == self.maxValue:	1✔
264	t.setColor(thickValue, mfr="max")	1✔
265	elif round(abs(temperature)) == self.minValue:	1✔
266	t.setColor(thickValue, mfr="min")	1✔
267	elif (	1✔
268	self.minValue
269	< round(abs(temperature))
270	<= self.lowerQuarter
271	):
272	t.setColor(thickValue, mfr="minTo25")	1✔
273	elif (	1✔
274	self.lowerQuarter
275	< round(abs(temperature))
276	<= self.medianValue
277	):
278	t.setColor(thickValue, mfr="25To50")	1✔
279	elif (	1✔
280	self.medianValue
281	< round(abs(temperature))
282	<= self.upperQuarter
283	):
284	t.setColor(thickValue, mfr="50To75")	1✔
285	elif (	1✔
286	self.upperQuarter
287	< round(abs(temperature))
288	< self.maxValue
289	):
290	t.setColor(thickValue, mfr="75ToMax")	1✔
291	else:
292	t.setColor(thickValue, mfr="test")	×
293	i += 1	1✔
294	if isinstance(t, _dpc.DoublePipeConnection):	1✔
295	coldMassFlowVariableName = (	1✔
296	_mnames.getCanonicalMassFlowVariableName(
297	componentDisplayName=t.displayName,
298	pipeName=t.coldModelPipe.name,
299	)
300	)
301	coldTemperatureVariableName = (	1✔
302	_cnames.getTemperatureVariableName(
303	t, _mfn.PortItemType.COLD
304	)
305	)
306
307	hotMassFlowVariableName = (	1✔
308	_mnames.getCanonicalMassFlowVariableName(
309	componentDisplayName=t.displayName,
310	pipeName=t.hotModelPipe.name,
311	)
312	)
313	hotTemperatureVariableName = (	1✔
314	_cnames.getTemperatureVariableName(
315	t, _mfn.PortItemType.HOT
316	)
317	)
318
319	coldMassFlow = self._getMassFlow(	1✔
320	coldMassFlowVariableName, self.timeStep
321	)
322	coldTemperature = self._getTemperature(	1✔
323	coldTemperatureVariableName, self.timeStep
324	)
325
326	hotMassFlow = self._getMassFlow(	1✔
327	hotMassFlowVariableName, self.timeStep
328	)
329	hotTemperature = self._getTemperature(	1✔
330	hotTemperatureVariableName, self.timeStep
331	)
332
333	t.setMassFlowAndTemperature(	1✔
334	coldMassFlow,
335	coldTemperature,
336	hotMassFlow,
337	hotTemperature,
338	)
339
340	elif isinstance(t, _tv.TVentil):	1✔
341	valvePositionVariableName = _mnames.getInputVariableName(	1✔
342	t, t.modelDiverter
343	)
344	if valvePositionVariableName in self.massFlowData.columns:	1✔
345	valvePosition = str(	1✔
346	self.massFlowData[valvePositionVariableName].iloc[
347	self.timeStep
348	]
349	)
350	t.setPositionForMassFlowSolver(valvePosition)	1✔
351	t.posLabel.setPlainText(valvePosition)	1✔
352	self.logger.debug(	1✔
353	"valve position: " + str(valvePosition)
354	)
355
356	else:
357	return	×
358
359	def _getMassFlow(self, mfrVariableName: str, timeStep: int) -> float:	1✔
360	truncatedMfrVariableName = _truncateName(mfrVariableName)	1✔
361	mass = self.massFlowData[truncatedMfrVariableName].iloc[timeStep]	1✔
362	return mass	1✔
363
364	def _getTemperature(	1✔
365	self, temperatureVariableName: str, timeStep: int
366	) -> float:
367	truncatedTemperatureVariableName = _truncateName(	1✔
368	temperatureVariableName
369	)
370	return self.tempMassFlowData[truncatedTemperatureVariableName].iloc[	1✔
371	timeStep
372	]
373
374	def pauseVis(self):	1✔
375	self.paused = True	1✔
376	self.qtm.stop()	1✔
377
378	def continueVis(self):	1✔
379	if self.started:	×
380	self.paused = False	×
381	self.qtm.start(1000)	×
382	else:
383	self.start()	×
384
385	def setSlider(self):	1✔
386	self.slider.setOrientation(1)	1✔
387	self.slider.setMinimum(0)	1✔
388	self.slider.setMaximum(self.timeSteps)	1✔
389	self.slider.setTickInterval(1)	1✔
390	self.slider.setTickPosition(2)	1✔
391	if self.checkTimeStep and self.checkTempTimeStep():	1✔
392	self.slider.setVisible(True)	1✔
393	self.slider.setEnabled(False)	1✔
394	self.increaseValue()	1✔
395	else:
396	self.slider.setVisible(True)	1✔
397	self.slider.setEnabled(True)	1✔
398
399	def testValChange(self):	1✔
400	"""
401	Updates mass flow visualizer after releasing the mouse
402	"""
403	val = self.slider.value()	×
404	self.logger.debug("Slider value has changed to " + str(val))	×
405	time = self.getTime(val)	×
406	self.currentStepLabel.setText("Time :" + str(self.convertTime(time)))	×
407	self.timeStep = val	×
408	self.advance()	×
409
410	def moveValues(self):	1✔
411	"""
412	Updates mass flow visualizer when moving the mouse
413	"""
414	val = self.slider.value()	1✔
415	self.logger.debug("Slider value is still: " + str(val))	1✔
416	time = self.getTime(val)	1✔
417	self.currentStepLabel.setText("Time :" + str(self.convertTime(time)))	1✔
418	self.timeStep = val	1✔
419	self.advance()	1✔
420
421	def increaseValue(self):	1✔
422	"""
423	For automatic slider movement
424
425	"""
426
427	self.timeStep += self._timeStepIncrease	1✔
428	if self.timeStep > self.maxTimeStep:	1✔
429	self.timeStep = 0	×
430	self.slider.setValue(self.timeStep)	1✔
431
432	def decreaseValue(self):	1✔
433	"""
434	For automatic slider movement
435
436	"""
437
438	self.timeStep -= self._timeStepIncrease	×
439	if self.timeStep < 0:	×
440	self.timeStep = self.maxTimeStep	×
441	self.slider.setValue(self.timeStep)	×
442
443	def checkTimeStep(self):	1✔
444	"""
445	Check individual columns of the data frame, If a column has rows with different values, return False.
446	Else, continue to next column. Return True if no such column can be found.
447
448	False indicates the rows are not identical and slider should be enabled
449	True indicates the rows are identical and slider should be disabled
450
451	"""
452	massFlowDataDup = self.massFlowData	×
453	massFlowDataDup = massFlowDataDup.drop(massFlowDataDup.index[0])	×
454	for items in round(abs(massFlowDataDup)).nunique().iteritems():	×
455	if items[0] != "TIME":	×
456	if items[1] > 1:	×
457	return False	×
458	return True	×
459
460	def checkTempTimeStep(self):	1✔
461	tempMassFlowDataDup = self.tempMassFlowData	1✔
462	tempMassFlowDataDup = tempMassFlowDataDup.drop(	1✔
463	tempMassFlowDataDup.index[0]
464	)
465	for items in tempMassFlowDataDup.nunique().items():	1✔
466	if items[0] != "TIME":	1✔
467	if items[1] > 1:	1✔
468	return False	1✔
469	return True	1✔
470
471	def getThresholdValues(self):	1✔
472	"""
473	Access the data frame, convert into a nested list.
474	Merge the nested list into one single list.
475	Remove 'nan' from the list.
476	Convert the values inside the list into absolute values then round them off.
477	Split the list into two for negative and positive values.
478	Get the minimum value, the 25th percentile, the median value, the 75th percentile and
479	the max value from the lists.
480	Returns
481	-------
482	"""
483
484	data = (	1✔
485	self.massFlowData.values.tolist()
486	) # data frame converted to nested list
487	for sublist in data: # delete the time column from the list	1✔
488	del sublist[0]	1✔
489	data = list(	1✔
490	_it.chain.from_iterable(data)
491	) # nested list combined into one list
492	cleanedData = [	1✔
493	x for x in data if str(x) != "nan"
494	] # remove nan from list
495	cleanedData = [	1✔
496	round(abs(num)) for num in cleanedData
497	] # get absolute value and round off
498	noDuplicateData = list(dict.fromkeys(cleanedData))	1✔
499
500	self.medianValueMfr = _np.percentile(	1✔
501	noDuplicateData, 50
502	) # median value / 50th percentile
503	self.lowerQuarterMfr = _np.percentile(	1✔
504	noDuplicateData, 25
505	) # 25th percentile
506	self.upperQuarterMfr = _np.percentile(	1✔
507	noDuplicateData, 75
508	) # 75th percentile
509	self.minValueMfr = _np.min(	1✔
510	noDuplicateData
511	) # minimum value excluding 0
512	self.maxValueMfr = _np.max(noDuplicateData) # max value	1✔
513
514	def getThickness(self, mass: float) -> int:	1✔
515	mass = abs(mass)	1✔
516	if mass == self.minValueMfr:	1✔
517	return 2	1✔
518	elif self.minValueMfr < mass <= self.medianValueMfr:	1✔
519	return 4	1✔
520	elif self.medianValueMfr < mass < self.maxValueMfr:	1✔
521	return 6	×
522	elif mass == self.maxValueMfr:	1✔
523	return 8	1✔
524	else:
525	return 2	×
526
527	def getTempThresholdValues(self):	1✔
528	"""
529	Access the data frame, convert into a nested list.
530	Merge the nested list into one single list.
531	Remove 'nan' from the list.
532	Convert the values inside the list into absolute values then round them off.
533	Split the list into two for negative and positive values.
534	Get the minimum value, the 25th percentile, the median value, the 75th percentile and
535	the max value from the lists.
536	Returns
537	-------
538	"""
539
540	data = (	1✔
541	self.tempMassFlowData.values.tolist()
542	) # data frame converted to nested list
543	for sublist in data: # delete the time column from the list	1✔
544	del sublist[0]	1✔
545	data = list(	1✔
546	_it.chain.from_iterable(data)
547	) # nested list combined into one list
548	cleanedData = [	1✔
549	x for x in data if str(x) != "nan"
550	] # remove nan from list
551	cleanedData = [	1✔
552	round(abs(num)) for num in cleanedData
553	] # get absolute value and round off
554	noDuplicateData = list(dict.fromkeys(cleanedData))	1✔
555
556	self.medianValue = _np.percentile(	1✔
557	noDuplicateData, 50
558	) # median value / 50th percentile
559	self.lowerQuarter = _np.percentile(	1✔
560	noDuplicateData, 25
561	) # 25th percentile
562	self.upperQuarter = _np.percentile(	1✔
563	noDuplicateData, 75
564	) # 75th percentile
565	self.minValue = _np.min(noDuplicateData) # minimum value excluding 0	1✔
566	self.maxValue = _np.max(noDuplicateData) # max value	1✔
567
568	# print(noDuplicateData)
569	# sys.exit()
570
571	def getTime(self, row):	1✔
572	"""
573	Gets the time of the current time step
574	"""
575	data = self.massFlowData	1✔
576	timeColumn = data[data.columns[0]]	1✔
577	self.logger.debug(str(timeColumn[row]))	1✔
578	return timeColumn[row]	1✔
579
580	def convertTime(self, time):	1✔
581	"""
582	Convert the time into YYYY--MM--DD HH:MM:SS format
583	"""
584	noOfHours = 8760	1✔
585	decHour = float(time) / float(noOfHours)	1✔
586	base = _dt.datetime(_dt.MINYEAR, 1, 1)	1✔
587	result = base + _dt.timedelta(	1✔
588	seconds=(base.replace(year=base.year + 1) - base).total_seconds()
589	* decHour
590	)
591	return str(result)	1✔
592
593	def pressedSlider(self):	1✔
594	self.pauseVis()	×
595
596	def closeEvent(self, a0):	1✔
597	self.pauseVis()	1✔
598	super().closeEvent(a0)	1✔
599
600	def keyPressEvent(self, e):	1✔
601	if e.key() == _qtc.Qt.Key_Up:	×
602	self.logger.debug("Up is pressed")	×
603	self.increaseValue()	×
604	elif e.key() == _qtc.Qt.Key_Down:	×
605	self.logger.debug("Down is pressed")	×
606	self.decreaseValue()	×
607
608
609	def _truncateColumnNames(df: _pd.DataFrame) -> None:	1✔
610	_ensureNamesDontCollideAfterTruncating(df.columns)	1✔
611	df.columns = [_truncateName(n) for n in df.columns]	1✔
612
613
614	def _ensureNamesDontCollideAfterTruncating(	1✔
615	columnNames: _tp.Sequence[str],
616	) -> None:
617	sortedColumnNames = sorted(columnNames)	1✔
618	groupedNames = [	1✔
619	list(g) for _, g in _it.groupby(sortedColumnNames, key=_truncateName)
620	]
621	collidingNames = _flatten(g for g in groupedNames if len(g) > 1)	1✔
622	if collidingNames:	1✔
623	formattedCollidingNames = "\n\t".join(collidingNames)	×
624	errorMessage = (	×
625	f"The following column names collide after truncating them to "
626	f"{_MAX_HEADER_LENGTH} characters:\n\t{formattedCollidingNames}"
627	)
628	raise ValueError(errorMessage)	×
629
630
631	def _truncateName(name: str):	1✔
632	return name[:_MAX_HEADER_LENGTH]	1✔
633
634
635	def _flatten(iterable: _tp.Iterable[_tp.Iterable[str]]) -> _tp.Sequence[str]:	1✔
636	return list(_it.chain.from_iterable(iterable))	1✔

SPF-OST / pytrnsys_gui / 11662562960

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