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

47.83

/trnsysGUI/components/plugin/factory.py

import collections.abc as _cabc
import dataclasses as _dc
import importlib.resources as _ilr
import typing as _tp

import pytrnsys.utils.result as _res
import trnsysGUI.BlockItem as _bi
import trnsysGUI.PortItemBase as _pib
import trnsysGUI.createSinglePipePortItem as _cspi
import trnsysGUI.doublePipePortItem as _dpi
import trnsysGUI.internalPiping as _ip
import trnsysGUI.massFlowSolver.networkModel as _mfn
from . import graphics as _graphics
from . import plugin as _plugin
from .specification import load as _sload
from .specification import model as _smodel

_T_co = _tp.TypeVar("_T_co", covariant=True, bound=_pib.PortItemBase)


@_dc.dataclass
class Factory:
    specificationLoader: _sload.Loader

    @staticmethod
    def createDefault() -> "Factory":
        packageResourceLoader = _sload.Loader.createPackageResourceLoader()
        return Factory(packageResourceLoader)

    @staticmethod
    def getTypeNames() -> _cabc.Sequence[str]:
        traversable = _ilr.files(_plugin.__name__).joinpath("data")
        typeNames = [tt.name for tt in traversable.iterdir()]
        return typeNames

    @staticmethod
    def hasTypeName(candidateTypeName: str) -> bool:
        return candidateTypeName in Factory.getTypeNames()

    def create(self, typeName: str) -> _res.Result[_plugin.Plugin]:
        if not self.hasTypeName(typeName):
            raise ValueError("Unknown type.", typeName)

        specificationResult = self.specificationLoader.load(typeName)
        if _res.isError(specificationResult):
            return _res.error(specificationResult)
        specification = _res.value(specificationResult)

        width, height = specification.size
        graphics = _graphics.Graphics.createForTypeNameAndSize(
            typeName, width=width, height=height
        )

        internalPipingFactory = InternalPipingFactory(specification)

        plugin = _plugin.Plugin(
            typeName,
            specification.defaultName,
            graphics,
            internalPipingFactory,
        )

        return plugin


@_dc.dataclass
class _PortProperties:
    defaultName: str
    directionLabel: str
    direction: _mfn.PortItemDirection

    def getName(self, nameSpec: str | None) -> str:
        return nameSpec or self.defaultName


_INPUT_PORT_PROPERTIES = _PortProperties(
    "In", "in", _mfn.PortItemDirection.INPUT
)
_OUTPUT_PORT_PROPERTIES = _PortProperties(
    "Out", "out", _mfn.PortItemDirection.OUTPUT
)
_OUTPUT0_PORT_PROPERTIES = _PortProperties(
    "Out0", "out", _mfn.PortItemDirection.OUTPUT
)
_OUTPUT1_PORT_PROPERTIES = _PortProperties(
    "Out1", "out", _mfn.PortItemDirection.OUTPUT
)


@_dc.dataclass
class InternalPipingFactory(_plugin.AbstractInternalPipingFactory):
    specification: _smodel.Specification

    def createInternalPiping(
        self, blockItem: _bi.BlockItem
    ) -> _plugin.CreatedInternalPiping:
        connectionsCreatedInternalPiping = (
            self._createInternalPipingForConnections(blockItem)
        )
        teePiecesCreatedInternalPiping = (
            self._createInternalPipingForTeePieces(blockItem)
        )

        combinedCreatedInternalPiping = (
            self._createCombinedCreatedInternalPiping(
                connectionsCreatedInternalPiping,
                teePiecesCreatedInternalPiping,
            )
        )
        return combinedCreatedInternalPiping

    def _createInternalPipingForConnections(
        self, blockItem: _bi.BlockItem
    ) -> _plugin.CreatedInternalPiping:
        connections = self.specification.connections

        if not connections:
            return _plugin.CreatedInternalPiping.empty()

        graphicalInputPorts = []
        graphicalOutputPorts = []

        nodes = []
        modelPortItemsToGraphicalPortItem = {}

        for connection in connections:
            graphicalInputPort, modelInputPort = (
                self._createGraphicalAndModelPort(
                    connection.input, _INPUT_PORT_PROPERTIES, blockItem
                )
            )

            graphicalInputPorts.append(graphicalInputPort)
            modelPortItemsToGraphicalPortItem[modelInputPort] = (
                graphicalInputPort
            )

            graphicalOutputPort, modelOutputPort = (
                self._createGraphicalAndModelPort(
                    connection.output, _OUTPUT_PORT_PROPERTIES, blockItem
                )
            )

            graphicalOutputPorts.append(graphicalOutputPort)
            modelPortItemsToGraphicalPortItem[modelOutputPort] = (
                graphicalOutputPort
            )

            pipe = _mfn.Pipe(modelInputPort, modelOutputPort, connection.name)
            nodes.append(pipe)

        internalPiping = _ip.InternalPiping(
            nodes, modelPortItemsToGraphicalPortItem
        )

        createdInternalPiping = _plugin.CreatedInternalPiping(
            graphicalInputPorts, graphicalOutputPorts, internalPiping
        )

        return createdInternalPiping

    def _createInternalPipingForTeePieces(
        self, blockItem: _bi.BlockItem
    ) -> _plugin.CreatedInternalPiping:
        graphicalInputPorts = list[_pib.PortItemBase]()
        graphicalOutputPorts = list[_pib.PortItemBase]()

        nodes = list[_mfn.Node]()
        modelPortItemsToGraphicalPortItem = dict[
            _mfn.PortItem, _pib.PortItemBase
        ]()

        teePieces = self.specification.teePieces

        if not teePieces:
            return _plugin.CreatedInternalPiping.empty()

        for teePiece in teePieces:
            self._createAndAddNodesAndPortItemsForTeePiece(
                blockItem,
                teePiece,
                nodes,
                modelPortItemsToGraphicalPortItem,
                graphicalInputPorts,
                graphicalOutputPorts,
            )

        internalPiping = _ip.InternalPiping(
            nodes, modelPortItemsToGraphicalPortItem
        )

        createdInternalPiping = _plugin.CreatedInternalPiping(
            graphicalInputPorts, graphicalOutputPorts, internalPiping
        )

        return createdInternalPiping

    def _createAndAddNodesAndPortItemsForTeePiece(
        self,
        blockItem: _bi.BlockItem,
        teePiece: _smodel.TeePiece,
        nodes: list[_mfn.Node],
        modelPortItemsToGraphicalPortItem: dict[
            _mfn.PortItem, _pib.PortItemBase
        ],
        graphicalInputPorts: list[_pib.PortItemBase],
        graphicalOutputPorts: list[_pib.PortItemBase],
    ) -> None:
        graphicalInputPort, modelInputPort = self._createGraphicalAndModelPort(
            teePiece.input, _INPUT_PORT_PROPERTIES, blockItem
        )

        graphicalOutput0Port, modelOutput0Port = (
            self._createGraphicalAndModelPort(
                teePiece.output0, _OUTPUT0_PORT_PROPERTIES, blockItem
            )
        )

        graphicalOutput1Port, modelOutput1Port = (
            self._createGraphicalAndModelPort(
                teePiece.output1, _OUTPUT1_PORT_PROPERTIES, blockItem
            )
        )

        graphicalInputPorts.append(graphicalInputPort)
        graphicalOutputPorts.append(graphicalOutput0Port)
        graphicalOutputPorts.append(graphicalOutput1Port)

        modelPortItemsToGraphicalPortItem[modelInputPort] = graphicalInputPort
        modelPortItemsToGraphicalPortItem[modelOutput0Port] = (
            graphicalOutput0Port
        )
        modelPortItemsToGraphicalPortItem[modelOutput1Port] = (
            graphicalOutput1Port
        )

        teePieceModel = _mfn.TeePiece(
            modelInputPort, modelOutput0Port, modelOutput1Port
        )

        nodes.append(teePieceModel)

    def _createGraphicalAndModelPort(
        self,
        portSpec: _smodel.Port,
        portProperties: _PortProperties,
        blockItem: _bi.BlockItem,
    ) -> tuple[_pib.PortItemBase, _mfn.PortItem]:
        graphicalPort = self._createPort(portSpec, portProperties, blockItem)

        posX, posY = portSpec.position
        graphicalPort.setPos(posX, posY)

        portName = portProperties.getName(portSpec.name)
        portItemType = _mfn.PortItemType(portSpec.type)
        modelInputPort = _mfn.PortItem(
            portName, portProperties.direction, portItemType
        )

        return graphicalPort, modelInputPort

    @staticmethod
    def _createPort(
        portSpec: _smodel.Port,
        portProperties: _PortProperties,
        blockItem: _bi.BlockItem,
    ) -> _pib.PortItemBase:
        match portSpec.type:
            case "standard":
                return _cspi.createSinglePipePortItem(
                    portProperties.directionLabel, blockItem
                )
            case "hot" | "cold":
                return _dpi.DoublePipePortItem(
                    portProperties.directionLabel, blockItem
                )
            case _:
                _tp.assert_never(portSpec.type)

    @staticmethod
    def _getDirectionLabel(
        direction: _mfn.PortItemDirection,
    ) -> _tp.Literal["i", "o"]:
        match direction:
            case _mfn.PortItemDirection.INPUT:
                return "i"
            case _mfn.PortItemDirection.OUTPUT:
                return "o"
            case _:
                _tp.assert_never(direction)

    @staticmethod
    def _createCombinedCreatedInternalPiping(
        connectionsCreatedInternalPiping: _plugin.CreatedInternalPiping,
        teePiecesCreatedInternalPiping: _plugin.CreatedInternalPiping,
    ) -> _plugin.CreatedInternalPiping:
        internalPiping = _ip.InternalPiping(
            [
                *connectionsCreatedInternalPiping.internalPiping.nodes,
                *teePiecesCreatedInternalPiping.internalPiping.nodes,
            ],
            {
                **connectionsCreatedInternalPiping.internalPiping.modelPortItemsToGraphicalPortItem,
                **teePiecesCreatedInternalPiping.internalPiping.modelPortItemsToGraphicalPortItem,
            },
        )
        createdInternalPiping = _plugin.CreatedInternalPiping(
            [
                *connectionsCreatedInternalPiping.inputPorts,
                *teePiecesCreatedInternalPiping.inputPorts,
            ],
            [
                *connectionsCreatedInternalPiping.outputPorts,
                *teePiecesCreatedInternalPiping.outputPorts,
            ],
            internalPiping,
        )
        return createdInternalPiping

1	import collections.abc as _cabc	1✔
2	import dataclasses as _dc	1✔
3	import importlib.resources as _ilr	1✔
4	import typing as _tp	1✔
5
6	import pytrnsys.utils.result as _res	1✔
7	import trnsysGUI.BlockItem as _bi	1✔
8	import trnsysGUI.PortItemBase as _pib	1✔
9	import trnsysGUI.createSinglePipePortItem as _cspi	1✔
10	import trnsysGUI.doublePipePortItem as _dpi	1✔
11	import trnsysGUI.internalPiping as _ip	1✔
12	import trnsysGUI.massFlowSolver.networkModel as _mfn	1✔
13	from . import graphics as _graphics	1✔
14	from . import plugin as _plugin	1✔
15	from .specification import load as _sload	1✔
16	from .specification import model as _smodel	1✔
17
18	_T_co = _tp.TypeVar("_T_co", covariant=True, bound=_pib.PortItemBase)	1✔
19
20
21	@_dc.dataclass	1✔
22	class Factory:	1✔
23	specificationLoader: _sload.Loader	1✔
24
25	@staticmethod	1✔
26	def createDefault() -> "Factory":	1✔
27	packageResourceLoader = _sload.Loader.createPackageResourceLoader()	1✔
28	return Factory(packageResourceLoader)	1✔
29
30	@staticmethod	1✔
31	def getTypeNames() -> _cabc.Sequence[str]:	1✔
32	traversable = _ilr.files(_plugin.__name__).joinpath("data")	1✔
33	typeNames = [tt.name for tt in traversable.iterdir()]	1✔
34	return typeNames	1✔
35
36	@staticmethod	1✔
37	def hasTypeName(candidateTypeName: str) -> bool:	1✔
38	return candidateTypeName in Factory.getTypeNames()	1✔
39
40	def create(self, typeName: str) -> _res.Result[_plugin.Plugin]:	1✔
41	if not self.hasTypeName(typeName):	1✔
42	raise ValueError("Unknown type.", typeName)	1✔
43
44	specificationResult = self.specificationLoader.load(typeName)	1✔
45	if _res.isError(specificationResult):	1✔
46	return _res.error(specificationResult)	×
47	specification = _res.value(specificationResult)	1✔
48
49	width, height = specification.size	1✔
50	graphics = _graphics.Graphics.createForTypeNameAndSize(	1✔
51	typeName, width=width, height=height
52	)
53
54	internalPipingFactory = InternalPipingFactory(specification)	1✔
55
56	plugin = _plugin.Plugin(	1✔
57	typeName,
58	specification.defaultName,
59	graphics,
60	internalPipingFactory,
61	)
62
63	return plugin	1✔
64
65
66	@_dc.dataclass	1✔
67	class _PortProperties:	1✔
68	defaultName: str	1✔
69	directionLabel: str	1✔
70	direction: _mfn.PortItemDirection	1✔
71
72	def getName(self, nameSpec: str \| None) -> str:	1✔
73	return nameSpec or self.defaultName	×
74
75
76	_INPUT_PORT_PROPERTIES = _PortProperties(	1✔
77	"In", "in", _mfn.PortItemDirection.INPUT
78	)
79	_OUTPUT_PORT_PROPERTIES = _PortProperties(	1✔
80	"Out", "out", _mfn.PortItemDirection.OUTPUT
81	)
82	_OUTPUT0_PORT_PROPERTIES = _PortProperties(	1✔
83	"Out0", "out", _mfn.PortItemDirection.OUTPUT
84	)
85	_OUTPUT1_PORT_PROPERTIES = _PortProperties(	1✔
86	"Out1", "out", _mfn.PortItemDirection.OUTPUT
87	)
88
89
90	@_dc.dataclass	1✔
91	class InternalPipingFactory(_plugin.AbstractInternalPipingFactory):	1✔
92	specification: _smodel.Specification	1✔
93
94	def createInternalPiping(	1✔
95	self, blockItem: _bi.BlockItem
96	) -> _plugin.CreatedInternalPiping:
NEW 97	connectionsCreatedInternalPiping = (	×
98	self._createInternalPipingForConnections(blockItem)
99	)
NEW 100	teePiecesCreatedInternalPiping = (	×
101	self._createInternalPipingForTeePieces(blockItem)
102	)
103
NEW 104	combinedCreatedInternalPiping = (	×
105	self._createCombinedCreatedInternalPiping(
106	connectionsCreatedInternalPiping,
107	teePiecesCreatedInternalPiping,
108	)
109	)
110	return combinedCreatedInternalPiping	×
111
112	def _createInternalPipingForConnections(	1✔
113	self, blockItem: _bi.BlockItem
114	) -> _plugin.CreatedInternalPiping:
UNCOV 115	connections = self.specification.connections	×
116
117	if not connections:	×
118	return _plugin.CreatedInternalPiping.empty()	×
119
120	graphicalInputPorts = []	×
121	graphicalOutputPorts = []	×
122
123	nodes = []	×
124	modelPortItemsToGraphicalPortItem = {}	×
125
126	for connection in connections:	×
NEW 127	graphicalInputPort, modelInputPort = (	×
128	self._createGraphicalAndModelPort(
129	connection.input, _INPUT_PORT_PROPERTIES, blockItem
130	)
131	)
132
133	graphicalInputPorts.append(graphicalInputPort)	×
NEW 134	modelPortItemsToGraphicalPortItem[modelInputPort] = (	×
135	graphicalInputPort
136	)
137
NEW 138	graphicalOutputPort, modelOutputPort = (	×
139	self._createGraphicalAndModelPort(
140	connection.output, _OUTPUT_PORT_PROPERTIES, blockItem
141	)
142	)
143
144	graphicalOutputPorts.append(graphicalOutputPort)	×
NEW 145	modelPortItemsToGraphicalPortItem[modelOutputPort] = (	×
146	graphicalOutputPort
147	)
148
149	pipe = _mfn.Pipe(modelInputPort, modelOutputPort, connection.name)	×
150	nodes.append(pipe)	×
151
NEW 152	internalPiping = _ip.InternalPiping(	×
153	nodes, modelPortItemsToGraphicalPortItem
154	)
155
NEW 156	createdInternalPiping = _plugin.CreatedInternalPiping(	×
157	graphicalInputPorts, graphicalOutputPorts, internalPiping
158	)
159
160	return createdInternalPiping	×
161
162	def _createInternalPipingForTeePieces(	1✔
163	self, blockItem: _bi.BlockItem
164	) -> _plugin.CreatedInternalPiping:
165	graphicalInputPorts = list[_pib.PortItemBase]()	×
166	graphicalOutputPorts = list[_pib.PortItemBase]()	×
167
168	nodes = list[_mfn.Node]()	×
NEW 169	modelPortItemsToGraphicalPortItem = dict[	×
170	_mfn.PortItem, _pib.PortItemBase
171	]()
172
173	teePieces = self.specification.teePieces	×
174
175	if not teePieces:	×
176	return _plugin.CreatedInternalPiping.empty()	×
177
178	for teePiece in teePieces:	×
179	self._createAndAddNodesAndPortItemsForTeePiece(	×
180	blockItem,
181	teePiece,
182	nodes,
183	modelPortItemsToGraphicalPortItem,
184	graphicalInputPorts,
185	graphicalOutputPorts,
186	)
187
NEW 188	internalPiping = _ip.InternalPiping(	×
189	nodes, modelPortItemsToGraphicalPortItem
190	)
191
NEW 192	createdInternalPiping = _plugin.CreatedInternalPiping(	×
193	graphicalInputPorts, graphicalOutputPorts, internalPiping
194	)
195
196	return createdInternalPiping	×
197
198	def _createAndAddNodesAndPortItemsForTeePiece(	1✔
199	self,
200	blockItem: _bi.BlockItem,
201	teePiece: _smodel.TeePiece,
202	nodes: list[_mfn.Node],
203	modelPortItemsToGraphicalPortItem: dict[
204	_mfn.PortItem, _pib.PortItemBase
205	],
206	graphicalInputPorts: list[_pib.PortItemBase],
207	graphicalOutputPorts: list[_pib.PortItemBase],
208	) -> None:
209	graphicalInputPort, modelInputPort = self._createGraphicalAndModelPort(	×
210	teePiece.input, _INPUT_PORT_PROPERTIES, blockItem
211	)
212
NEW 213	graphicalOutput0Port, modelOutput0Port = (	×
214	self._createGraphicalAndModelPort(
215	teePiece.output0, _OUTPUT0_PORT_PROPERTIES, blockItem
216	)
217	)
218
NEW 219	graphicalOutput1Port, modelOutput1Port = (	×
220	self._createGraphicalAndModelPort(
221	teePiece.output1, _OUTPUT1_PORT_PROPERTIES, blockItem
222	)
223	)
224
225	graphicalInputPorts.append(graphicalInputPort)	×
226	graphicalOutputPorts.append(graphicalOutput0Port)	×
227	graphicalOutputPorts.append(graphicalOutput1Port)	×
228
229	modelPortItemsToGraphicalPortItem[modelInputPort] = graphicalInputPort	×
NEW 230	modelPortItemsToGraphicalPortItem[modelOutput0Port] = (	×
231	graphicalOutput0Port
232	)
NEW 233	modelPortItemsToGraphicalPortItem[modelOutput1Port] = (	×
234	graphicalOutput1Port
235	)
236
NEW 237	teePieceModel = _mfn.TeePiece(	×
238	modelInputPort, modelOutput0Port, modelOutput1Port
239	)
240
241	nodes.append(teePieceModel)	×
242
243	def _createGraphicalAndModelPort(	1✔
244	self,
245	portSpec: _smodel.Port,
246	portProperties: _PortProperties,
247	blockItem: _bi.BlockItem,
248	) -> tuple[_pib.PortItemBase, _mfn.PortItem]:
249	graphicalPort = self._createPort(portSpec, portProperties, blockItem)	×
250
251	posX, posY = portSpec.position	×
252	graphicalPort.setPos(posX, posY)	×
253
254	portName = portProperties.getName(portSpec.name)	×
255	portItemType = _mfn.PortItemType(portSpec.type)	×
NEW 256	modelInputPort = _mfn.PortItem(	×
257	portName, portProperties.direction, portItemType
258	)
259
260	return graphicalPort, modelInputPort	×
261
262	@staticmethod	1✔
263	def _createPort(	1✔
264	portSpec: _smodel.Port,
265	portProperties: _PortProperties,
266	blockItem: _bi.BlockItem,
267	) -> _pib.PortItemBase:
268	match portSpec.type:	×
269	case "standard":	×
NEW 270	return _cspi.createSinglePipePortItem(	×
271	portProperties.directionLabel, blockItem
272	)
273	case "hot" \| "cold":	×
NEW 274	return _dpi.DoublePipePortItem(	×
275	portProperties.directionLabel, blockItem
276	)
277	case _:	×
278	_tp.assert_never(portSpec.type)	×
279
280	@staticmethod	1✔
281	def _getDirectionLabel(	1✔
282	direction: _mfn.PortItemDirection,
283	) -> _tp.Literal["i", "o"]:
284	match direction:	×
285	case _mfn.PortItemDirection.INPUT:	×
286	return "i"	×
287	case _mfn.PortItemDirection.OUTPUT:	×
288	return "o"	×
289	case _:	×
290	_tp.assert_never(direction)	×
291
292	@staticmethod	1✔
293	def _createCombinedCreatedInternalPiping(	1✔
294	connectionsCreatedInternalPiping: _plugin.CreatedInternalPiping,
295	teePiecesCreatedInternalPiping: _plugin.CreatedInternalPiping,
296	) -> _plugin.CreatedInternalPiping:
297	internalPiping = _ip.InternalPiping(	×
298	[
299	*connectionsCreatedInternalPiping.internalPiping.nodes,
300	*teePiecesCreatedInternalPiping.internalPiping.nodes,
301	],
302	{
303	**connectionsCreatedInternalPiping.internalPiping.modelPortItemsToGraphicalPortItem,
304	**teePiecesCreatedInternalPiping.internalPiping.modelPortItemsToGraphicalPortItem,
305	},
306	)
307	createdInternalPiping = _plugin.CreatedInternalPiping(	×
308	[
309	*connectionsCreatedInternalPiping.inputPorts,
310	*teePiecesCreatedInternalPiping.inputPorts,
311	],
312	[
313	*connectionsCreatedInternalPiping.outputPorts,
314	*teePiecesCreatedInternalPiping.outputPorts,
315	],
316	internalPiping,
317	)
318	return createdInternalPiping	×

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