1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
|
import logging
from qtpy.QtCore import QPointF
from .base import ConnectionBase, FlowSceneBase, NodeBase
from .exceptions import (ConnectionCycleFailure, ConnectionPointFailure,
ConnectionPortNotEmptyFailure,
ConnectionRequiresPortFailure, ConnectionSelfFailure,
ConnectionDataTypeFailure, NodeConnectionFailure)
from .port import PortType, opposite_port
logger = logging.getLogger(__name__)
class NodeConnectionInteraction:
def __init__(self, node: NodeBase, connection: ConnectionBase, scene: FlowSceneBase):
'''
An interactive connection interaction to complete `connection` with the
given node
Parameters
----------
node : Node
connection : Connection
scene : FlowScene
'''
self._node = node
self._connection = connection
self._scene = scene
@property
def creates_cycle(self):
"""Would completing the connection introduce a cycle?"""
required_port = self.connection_required_port
return self.connection_node.has_connection_by_port_type(
self._node, required_port)
def can_connect(self) -> bool:
"""
Can connect when following conditions are met:
1) Connection 'requires' a port - i.e., is missing either a start
node or an end node
2) Connection's vacant end is above the node port in the user
interface
3) Node port is vacant
4) Connection does not introduce a cycle in the graph
5) Connection type equals node port type, or there is a registered
type conversion that can translate between the two
Parameters
----------
Returns
-------
(port_index, converter) : (int, TypeConverter)
where port_index is the index of the port to be connected
Raises
------
NodeConnectionFailure
ConnectionDataTypeFailure
If port data types are not compatible
"""
# 1) Connection requires a port
required_port = self.connection_required_port
if required_port == PortType.none:
raise ConnectionRequiresPortFailure('Connection requires a port')
elif required_port not in (PortType.input, PortType.output):
raise ValueError(f'Invalid port specified {required_port}')
# 1.5) Forbid connecting the node to itself
node = self.connection_node
if node == self._node:
raise ConnectionSelfFailure(f'Cannot connect {node} to itself')
# 2) connection point is on top of the node port
connection_point = self.connection_end_scene_position(required_port)
port = self.node_port_under_scene_point(required_port,
connection_point)
if not port:
raise ConnectionPointFailure(
f'Connection point {connection_point} is not on node {node}')
# 3) Node port is vacant
if not port.can_connect:
raise ConnectionPortNotEmptyFailure(
f'Port {required_port} {port} cannot connect'
)
# 4) Cycle check
if self.creates_cycle:
raise ConnectionCycleFailure(
f'Connecting {self._node} and {node} would introduce a '
f'cycle in the graph'
)
# 5) Connection type equals node port type, or there is a registered
# type conversion that can translate between the two
connection_data_type = self._connection.data_type(opposite_port(required_port))
candidate_node_data_type = port.data_type
if connection_data_type.id == candidate_node_data_type.id:
return port, None
registry = self._scene.registry
if required_port == PortType.input:
converter = registry.get_type_converter(connection_data_type,
candidate_node_data_type)
else:
converter = registry.get_type_converter(candidate_node_data_type,
connection_data_type)
if not converter:
raise ConnectionDataTypeFailure(
f'{connection_data_type} and {candidate_node_data_type} are not compatible'
)
return port, converter
def try_connect(self) -> bool:
"""
Try to connect the nodes. Steps::
1) Check conditions from 'can_connect'
1.5) If the connection is possible but a type conversion is needed, add
a converter node to the scene, and connect it properly
2) Assign node to required port in Connection
3) Assign Connection to empty port in NodeState
4) Adjust Connection geometry
5) Poke model to initiate data transfer
Returns
-------
value : bool
"""
# 1) Check conditions from 'can_connect'
try:
port, converter = self.can_connect()
except NodeConnectionFailure as ex:
logger.debug('Cannot connect node', exc_info=ex)
logger.info('Cannot connect node: %s', ex)
return False
# 1.5) If the connection is possible but a type conversion is needed,
# assign a convertor to connection
if converter:
self._connection.type_converter = converter
# 2) Assign node to required port in Connection
port.add_connection(self._connection)
# 3) Assign Connection to empty port in NodeState
# The port is not longer required after this function
self._connection.connect_to(port)
# 4) Adjust Connection geometry
self._node.graphics_object.move_connections()
# 5) Poke model to intiate data transfer
_, out_port = self._connection.ports
if out_port:
out_port.node.on_data_updated(out_port)
return True
def disconnect(self, port_to_disconnect: PortType) -> bool:
"""
1) Node and Connection should be already connected
2) If so, clear Connection entry in the NodeState
3) Propagate invalid data to IN node
4) Set Connection end to 'requiring a port'
Parameters
----------
port_to_disconnect : PortType
Returns
-------
value : bool
"""
port_index = self._connection.get_port_index(port_to_disconnect)
state = self._node.state
# clear pointer to Connection in the NodeState
state.erase_connection(port_to_disconnect, port_index,
self._connection)
# Propagate invalid data to IN node
self._connection.propagate_empty_data()
# clear Connection side
self._connection.clear_node(port_to_disconnect)
self._connection.required_port = port_to_disconnect
self._connection.graphics_object.grabMouse()
@property
def connection_required_port(self) -> PortType:
"""
The required port type to complete the connection
Returns
-------
value : PortType
"""
return self._connection.required_port
@property
def connection_node(self):
"""The node already specified for the connection"""
required_port = self.connection_required_port
return self._connection.get_node(opposite_port(required_port))
def connection_end_scene_position(self, port_type: PortType) -> QPointF:
"""
Connection end scene position
Parameters
----------
port_type : PortType
Returns
-------
value : QPointF
"""
go = self._connection.graphics_object
geometry = self._connection.geometry
end_point = geometry.get_end_point(port_type)
return go.mapToScene(end_point)
def node_port_scene_position(self, port_type: PortType, port_index: int) -> QPointF:
"""
Node port scene position
Parameters
----------
port_type : PortType
port_index : int
Returns
-------
value : QPointF
"""
port = self._node.state[port_type][port_index]
return port.get_mapped_scene_position(
self._node.graphics_object.sceneTransform())
def node_port_under_scene_point(self, port_type: PortType, scene_point: QPointF) -> NodeBase:
"""
Node port under scene point
Parameters
----------
port_type : PortType
p : QPointF
Returns
-------
value : int
"""
node_geom = self._node.geometry
scene_transform = self._node.graphics_object.sceneTransform()
return node_geom.check_hit_scene_point(port_type, scene_point, scene_transform)
def node_port_is_empty(self, port_type: PortType, port_index: int) -> bool:
"""
Node port is empty
Parameters
----------
port_type : PortType
port_index : int
Returns
-------
value : bool
"""
port = self._node.state[port_type][port_index]
return port.can_connect
|
