11576810878

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

Build # 11576810878

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

86.42

/trnsysGUI/diagram/Decoder.py

# pylint: skip-file
# type: ignore

import json as _json
import typing as _tp

import trnsysGUI.blockItems.getBlockItem as _gbi
import trnsysGUI.connection.connectionBase as _cb
import trnsysGUI.connection.doublePipeConnection as _dpc
import trnsysGUI.connection.singlePipeConnection as _spc
import trnsysGUI.connection.undo as _cundo
import trnsysGUI.doublePipePortItem as _dbi
import trnsysGUI.singlePipePortItem as _spi


class Decoder(_json.JSONDecoder):
    """
    Decodes the diagram
    """

    _UNSET_ERROR_NAME = "ERROR_NAME_NOT_SET_IN_DECODE"

    def __init__(self, *args, **kwargs):

        self.editor = kwargs["editor"]
        self.logger = self.editor.logger
        kwargs.pop("editor")
        _json.JSONDecoder.__init__(
            self, object_hook=self.object_hook, *args, **kwargs
        )

    def object_hook(
        self, arr: _tp.Mapping[str, _tp.Mapping[str, _tp.Any]]
    ) -> _tp.Any:
        """
        This is the decoding function. object_hook seems to get executed for every sub dictionary in the json file.
        By looking for the specific key containing the name of dict elements, one can extract the needed dict.
        The name of the dicts is important because the order in which they are loaded matters (some objects depend on others)
        Parameters
        ----------
        arr

        Returns
        -------

        """

        resBlockList = []

        if ".__BlockDct__" in arr:

            resConnList = []

            sortedItems = sorted(arr.items(), key=lambda t: t[0])

            sortedValues: _tp.Sequence[_tp.Mapping[str, _tp.Any]]
            sortedValues = [v for _, v in sortedItems]

            namesManager = self.editor.namesManager

            for i in sortedValues:
                if type(i) is not dict:
                    continue

                # Adding the blocks to the scene could also be done inside Decoder now (before not possible because
                # no way of accessing the diagramView)

                elif ".__BlockDict__" in i:
                    componentType = i["BlockName"]
                    self.logger.debug("Found a block ")

                    if componentType == "GraphicalItem":
                        blockItem = _gbi.createBlockItem(
                            "GraphicalItem", self.editor, namesManager
                        )
                    elif (
                        componentType == "Control"
                        or componentType == "MasterControl"
                    ):
                        self.logger.warning(
                            f"BlockItem: '{componentType}' is no longer supported in the GUI."
                        )
                        continue
                    else:
                        blockItem = _gbi.createBlockItem(
                            componentType,
                            self.editor,
                            namesManager,
                            displayName=i["BlockDisplayName"],
                        )

                    self.editor.trnsysObj.append(blockItem)

                    blockItem.decode(i, resBlockList)

                elif ".__ConnectionDict__" in i:
                    fromPortId, toPortId = self._getPortIds(i)

                    # Looking for the ports the connection is connected to
                    fromPort = None
                    toPort = None
                    for connBl in resBlockList:
                        for p in connBl.inputs + connBl.outputs:
                            if p.id == fromPortId:
                                fromPort = p
                            if p.id == toPortId:
                                toPort = p

                    connection: _cb.ConnectionBase
                    match (fromPort, toPort):
                        case (None, _):
                            raise AssertionError("Couldn't find from port.")
                        case (_, None):
                            raise AssertionError("Couldn't find to port.")
                        case (
                            _spi.SinglePipePortItem(),
                            _spi.SinglePipePortItem(),
                        ):
                            connection = _spc.SinglePipeConnection(
                                self._UNSET_ERROR_NAME,
                                fromPort,
                                toPort,
                                self.editor,
                            )
                        case (
                            _dbi.DoublePipePortItem(),
                            _dbi.DoublePipePortItem(),
                        ):
                            connection = _dpc.DoublePipeConnection(
                                self._UNSET_ERROR_NAME,
                                fromPort,
                                toPort,
                                self.editor,
                            )
                        case _:
                            raise AssertionError(
                                "`fromPort' and `toPort' have different types."
                            )

                    connection.decode(i)
                    _cundo.reAddConnection(connection)
                    namesManager.addName(connection.displayName)

                    self.editor.diagramScene.addItem(connection)
                    resConnList.append(connection)

                elif "__idDct__" in i:
                    resBlockList.append(i)
                elif "__nameDct__" in i:
                    resBlockList.append(i)
                else:
                    self.logger.debug(
                        "Error: Not recognized object in decoder, " + str(i)
                    )

            return resBlockList, resConnList

        return arr

    @staticmethod
    def _getPortIds(i):
        if "PortFromID" in i and "PortToID" in i:
            # Legacy port ID naming
            fromPortId = i["PortFromID"]
            toPortId = i["PortToID"]
        elif "fromPortId" in i and "toPortId" in i:
            fromPortId = i["fromPortId"]
            toPortId = i["toPortId"]
        else:
            raise AssertionError("Could not find port IDs for connection.")
        return fromPortId, toPortId

1	# pylint: skip-file
2	# type: ignore
3
4	import json as _json	1✔
5	import typing as _tp	1✔
6
7	import trnsysGUI.blockItems.getBlockItem as _gbi	1✔
8	import trnsysGUI.connection.connectionBase as _cb	1✔
9	import trnsysGUI.connection.doublePipeConnection as _dpc	1✔
10	import trnsysGUI.connection.singlePipeConnection as _spc	1✔
11	import trnsysGUI.connection.undo as _cundo	1✔
12	import trnsysGUI.doublePipePortItem as _dbi	1✔
13	import trnsysGUI.singlePipePortItem as _spi	1✔
14
15
16	class Decoder(_json.JSONDecoder):	1✔
17	"""	1✔
18	Decodes the diagram
19	"""
20
21	_UNSET_ERROR_NAME = "ERROR_NAME_NOT_SET_IN_DECODE"	1✔
22
23	def __init__(self, args, *kwargs):	1✔
24
25	self.editor = kwargs["editor"]	1✔
26	self.logger = self.editor.logger	1✔
27	kwargs.pop("editor")	1✔
28	_json.JSONDecoder.__init__(	1✔
29	self, object_hook=self.object_hook, args, *kwargs
30	)
31
32	def object_hook(	1✔
33	self, arr: _tp.Mapping[str, _tp.Mapping[str, _tp.Any]]
34	) -> _tp.Any:
35	"""
36	This is the decoding function. object_hook seems to get executed for every sub dictionary in the json file.
37	By looking for the specific key containing the name of dict elements, one can extract the needed dict.
38	The name of the dicts is important because the order in which they are loaded matters (some objects depend on others)
39	Parameters
40	----------
41	arr
42
43	Returns
44	-------
45
46	"""
47
48	resBlockList = []	1✔
49
50	if ".__BlockDct__" in arr:	1✔
51
52	resConnList = []	1✔
53
54	sortedItems = sorted(arr.items(), key=lambda t: t[0])	1✔
55
56	sortedValues: _tp.Sequence[_tp.Mapping[str, _tp.Any]]
57	sortedValues = [v for _, v in sortedItems]	1✔
58
59	namesManager = self.editor.namesManager	1✔
60
61	for i in sortedValues:	1✔
62	if type(i) is not dict:	1✔
63	continue	1✔
64
65	# Adding the blocks to the scene could also be done inside Decoder now (before not possible because
66	# no way of accessing the diagramView)
67
68	elif ".__BlockDict__" in i:	1✔
69	componentType = i["BlockName"]	1✔
70	self.logger.debug("Found a block ")	1✔
71
72	if componentType == "GraphicalItem":	1✔
NEW 73	blockItem = _gbi.createBlockItem(	×
74	"GraphicalItem", self.editor, namesManager
75	)
76	elif (	1✔
77	componentType == "Control"
78	or componentType == "MasterControl"
79	):
NEW 80	self.logger.warning(	×
81	f"BlockItem: '{componentType}' is no longer supported in the GUI."
82	)
UNCOV 83	continue	×
84	else:
85	blockItem = _gbi.createBlockItem(	1✔
86	componentType,
87	self.editor,
88	namesManager,
89	displayName=i["BlockDisplayName"],
90	)
91
92	self.editor.trnsysObj.append(blockItem)	1✔
93
94	blockItem.decode(i, resBlockList)	1✔
95
96	elif ".__ConnectionDict__" in i:	1✔
97	fromPortId, toPortId = self._getPortIds(i)	1✔
98
99	# Looking for the ports the connection is connected to
100	fromPort = None	1✔
101	toPort = None	1✔
102	for connBl in resBlockList:	1✔
103	for p in connBl.inputs + connBl.outputs:	1✔
104	if p.id == fromPortId:	1✔
105	fromPort = p	1✔
106	if p.id == toPortId:	1✔
107	toPort = p	1✔
108
109	connection: _cb.ConnectionBase
110	match (fromPort, toPort):	1✔
111	case (None, _):	1✔
112	raise AssertionError("Couldn't find from port.")	×
113	case (_, None):	1✔
114	raise AssertionError("Couldn't find to port.")	×
115	case (	1✔
116	_spi.SinglePipePortItem(),
117	_spi.SinglePipePortItem(),
118	):
119	connection = _spc.SinglePipeConnection(	1✔
120	self._UNSET_ERROR_NAME,
121	fromPort,
122	toPort,
123	self.editor,
124	)
125	case (	1✔
126	_dbi.DoublePipePortItem(),
127	_dbi.DoublePipePortItem(),
128	):
129	connection = _dpc.DoublePipeConnection(	1✔
130	self._UNSET_ERROR_NAME,
131	fromPort,
132	toPort,
133	self.editor,
134	)
135	case _:	×
NEW 136	raise AssertionError(	×
137	"`fromPort' and `toPort' have different types."
138	)
139
140	connection.decode(i)	1✔
141	_cundo.reAddConnection(connection)	1✔
142	namesManager.addName(connection.displayName)	1✔
143
144	self.editor.diagramScene.addItem(connection)	1✔
145	resConnList.append(connection)	1✔
146
147	elif "__idDct__" in i:	1✔
148	resBlockList.append(i)	1✔
149	elif "__nameDct__" in i:	1✔
150	resBlockList.append(i)	1✔
151	else:
NEW 152	self.logger.debug(	×
153	"Error: Not recognized object in decoder, " + str(i)
154	)
155
156	return resBlockList, resConnList	1✔
157
158	return arr	1✔
159
160	@staticmethod	1✔
161	def _getPortIds(i):	1✔
162	if "PortFromID" in i and "PortToID" in i:	1✔
163	# Legacy port ID naming
164	fromPortId = i["PortFromID"]	×
165	toPortId = i["PortToID"]	×
166	elif "fromPortId" in i and "toPortId" in i:	1✔
167	fromPortId = i["fromPortId"]	1✔
168	toPortId = i["toPortId"]	1✔
169	else:
170	raise AssertionError("Could not find port IDs for connection.")	×
171	return fromPortId, toPortId	1✔

SPF-OST / pytrnsys_gui / 11576810878

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