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

82.46

/trnsysGUI/proforma/createModelConnections.py

import collections.abc as _cabc
import typing as _tp

import pytrnsys.utils.result as _res

import trnsysGUI.internalPiping as _ip
import trnsysGUI.massFlowSolver.networkModel as _mfn

from . import models as _models


def createModelConnectionsFromInternalPiping(
    internalPiping: _ip.InternalPiping,
) -> _res.Result[_cabc.Sequence[_models.Connection]]:
    connections = []
    for node in internalPiping.nodes:
        if not isinstance(node, _mfn.Pipe):
            return _res.Error(
                f"`{node.name}` is a `{node.__class__.__name__}`, but only direct pipes are supported."
            )
        pipe = node

        inputPort = _models.InputPort(pipe.fromPort.name)
        outputPort = _models.OutputPort(pipe.toPort.name)

        connection = _models.Connection(pipe.name, inputPort, outputPort)

        connections.append(connection)

    return connections


_StringMapping = _tp.Mapping[str, _tp.Any]


def createModelConnectionsFromProforma(
    serializedConnections: _cabc.Sequence[_StringMapping],
    variables: _cabc.Sequence[_models.Variable],
) -> _res.Result[_cabc.Sequence[_models.Connection]]:
    variablesByOrder = {v.order: v for v in variables}

    connections = [
        _createConnection(s, variablesByOrder) for s in serializedConnections
    ]

    return connections


def _createConnection(
    serializedConnection: _StringMapping,
    variablesByOrder: _cabc.Mapping[int, _models.Variable],
) -> _models.Connection:
    serializedInputPort = serializedConnection["input"]
    inputPort = _createInputPort(serializedInputPort, variablesByOrder)

    outputPort = _createOutputPort(serializedConnection, variablesByOrder)

    name = serializedConnection.get("@name")

    fluidDensityVariable = _getOptionalVariable(
        serializedInputPort, "fluidDensity", variablesByOrder
    )
    fluidHeatCapacityVariable = _getOptionalVariable(
        serializedInputPort, "fluidHeatCapacity", variablesByOrder
    )
    fluid = _models.Fluid(fluidDensityVariable, fluidHeatCapacityVariable)

    connection = _models.Connection(name, inputPort, outputPort, fluid)

    return connection


def _createInputPort(
    serializedInputPort: _StringMapping,
    variablesByOrder: _cabc.Mapping[int, _models.Variable],
) -> _models.InputPort:
    name = serializedInputPort["@name"]
    massFlowRateVariable = _getVariable(
        serializedInputPort, "massFlowRate", variablesByOrder
    )
    temperatureVariable = _getVariable(
        serializedInputPort, "temperature", variablesByOrder
    )
    inputPort = _models.InputPort(
        name, temperatureVariable, massFlowRateVariable
    )
    return inputPort


def _createOutputPort(
    serializedConnection: _StringMapping,
    variablesByOrder: _cabc.Mapping[int, _models.Variable],
) -> _models.OutputPort:
    serializedOutputPort = serializedConnection["output"]
    name = serializedOutputPort["@name"]
    temperature = _getVariable(
        serializedOutputPort, "temperature", variablesByOrder
    )
    reverseTemperature = _getOptionalVariable(
        serializedOutputPort, "reverseTemperature", variablesByOrder
    )
    outputPort = _models.OutputPort(name, temperature, reverseTemperature)
    return outputPort


def _getVariable(
    serializedPort: _StringMapping,
    variableName: str,
    variablesByOrder: _cabc.Mapping[int, _models.Variable],
) -> _models.Variable:
    variable = _getOptionalVariable(
        serializedPort, variableName, variablesByOrder
    )
    portName = serializedPort["@name"]
    assert (
        variable
    ), f"No associated `{variableName}` variable for port {portName}"
    return variable


def _getOptionalVariable(
    serializedPort: _StringMapping,
    variableName: str,
    variablesByOrder: _cabc.Mapping[int, _models.Variable],
) -> _tp.Optional[_models.Variable]:
    child = serializedPort.get(variableName)
    if not child:
        return None

    order = child["variableReference"]["order"]
    variable = variablesByOrder[order]
    return variable

1	import collections.abc as _cabc	1✔
2	import typing as _tp	1✔
3
4	import pytrnsys.utils.result as _res	1✔
5
6	import trnsysGUI.internalPiping as _ip	1✔
7	import trnsysGUI.massFlowSolver.networkModel as _mfn	1✔
8
9	from . import models as _models	1✔
10
11
12	def createModelConnectionsFromInternalPiping(	1✔
13	internalPiping: _ip.InternalPiping,
14	) -> _res.Result[_cabc.Sequence[_models.Connection]]:
15	connections = []	×
16	for node in internalPiping.nodes:	×
17	if not isinstance(node, _mfn.Pipe):	×
NEW 18	return _res.Error(	×
19	f"`{node.name}` is a `{node.__class__.__name__}`, but only direct pipes are supported."
20	)
UNCOV 21	pipe = node	×
22
23	inputPort = _models.InputPort(pipe.fromPort.name)	×
24	outputPort = _models.OutputPort(pipe.toPort.name)	×
25
26	connection = _models.Connection(pipe.name, inputPort, outputPort)	×
27
28	connections.append(connection)	×
29
30	return connections	×
31
32
33	_StringMapping = _tp.Mapping[str, _tp.Any]	1✔
34
35
36	def createModelConnectionsFromProforma(	1✔
37	serializedConnections: _cabc.Sequence[_StringMapping],
38	variables: _cabc.Sequence[_models.Variable],
39	) -> _res.Result[_cabc.Sequence[_models.Connection]]:
40	variablesByOrder = {v.order: v for v in variables}	1✔
41
42	connections = [	1✔
43	_createConnection(s, variablesByOrder) for s in serializedConnections
44	]
45
46	return connections	1✔
47
48
49	def _createConnection(	1✔
50	serializedConnection: _StringMapping,
51	variablesByOrder: _cabc.Mapping[int, _models.Variable],
52	) -> _models.Connection:
53	serializedInputPort = serializedConnection["input"]	1✔
54	inputPort = _createInputPort(serializedInputPort, variablesByOrder)	1✔
55
56	outputPort = _createOutputPort(serializedConnection, variablesByOrder)	1✔
57
58	name = serializedConnection.get("@name")	1✔
59
60	fluidDensityVariable = _getOptionalVariable(	1✔
61	serializedInputPort, "fluidDensity", variablesByOrder
62	)
63	fluidHeatCapacityVariable = _getOptionalVariable(	1✔
64	serializedInputPort, "fluidHeatCapacity", variablesByOrder
65	)
66	fluid = _models.Fluid(fluidDensityVariable, fluidHeatCapacityVariable)	1✔
67
68	connection = _models.Connection(name, inputPort, outputPort, fluid)	1✔
69
70	return connection	1✔
71
72
73	def _createInputPort(	1✔
74	serializedInputPort: _StringMapping,
75	variablesByOrder: _cabc.Mapping[int, _models.Variable],
76	) -> _models.InputPort:
77	name = serializedInputPort["@name"]	1✔
78	massFlowRateVariable = _getVariable(	1✔
79	serializedInputPort, "massFlowRate", variablesByOrder
80	)
81	temperatureVariable = _getVariable(	1✔
82	serializedInputPort, "temperature", variablesByOrder
83	)
84	inputPort = _models.InputPort(	1✔
85	name, temperatureVariable, massFlowRateVariable
86	)
87	return inputPort	1✔
88
89
90	def _createOutputPort(	1✔
91	serializedConnection: _StringMapping,
92	variablesByOrder: _cabc.Mapping[int, _models.Variable],
93	) -> _models.OutputPort:
94	serializedOutputPort = serializedConnection["output"]	1✔
95	name = serializedOutputPort["@name"]	1✔
96	temperature = _getVariable(	1✔
97	serializedOutputPort, "temperature", variablesByOrder
98	)
99	reverseTemperature = _getOptionalVariable(	1✔
100	serializedOutputPort, "reverseTemperature", variablesByOrder
101	)
102	outputPort = _models.OutputPort(name, temperature, reverseTemperature)	1✔
103	return outputPort	1✔
104
105
106	def _getVariable(	1✔
107	serializedPort: _StringMapping,
108	variableName: str,
109	variablesByOrder: _cabc.Mapping[int, _models.Variable],
110	) -> _models.Variable:
111	variable = _getOptionalVariable(	1✔
112	serializedPort, variableName, variablesByOrder
113	)
114	portName = serializedPort["@name"]	1✔
115	assert (	1✔
116	variable
117	), f"No associated `{variableName}` variable for port {portName}"
118	return variable	1✔
119
120
121	def _getOptionalVariable(	1✔
122	serializedPort: _StringMapping,
123	variableName: str,
124	variablesByOrder: _cabc.Mapping[int, _models.Variable],
125	) -> _tp.Optional[_models.Variable]:
126	child = serializedPort.get(variableName)	1✔
127	if not child:	1✔
128	return None	1✔
129
130	order = child["variableReference"]["order"]	1✔
131	variable = variablesByOrder[order]	1✔
132	return variable	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