Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
node.properties = properties
nodes.append(node)
nodes_by_id[node_d.id] = node
for link_d in desc.links:
source_id = link_d.source_node_id
sink_id = link_d.sink_node_id
if source_id in nodes_not_found or sink_id in nodes_not_found:
continue
source = nodes_by_id[source_id]
sink = nodes_by_id[sink_id]
try:
link = SchemeLink(
source,
link_d.source_channel,
sink,
link_d.sink_channel,
enabled=link_d.enabled,
)
except (ValueError, IncompatibleChannelTypeError) as ex:
error_handler(ex)
else:
links.append(link)
for annot_d in desc.annotations:
params = annot_d.params
if annot_d.type == "text":
annot = SchemeTextAnnotation(
params.geometry, params.text, params.content_type, params.font
if existing:
assert len(existing) == 1
self.document.removeLink(existing[0])
for source_channel, sink_channel in links_to_remove:
links = self.scheme.find_links(
source_node=self.source_node,
source_channel=source_channel,
sink_node=self.sink_node,
sink_channel=sink_channel,
)
assert len(links) == 1
self.document.removeLink(links[0])
for source_channel, sink_channel in links_to_add:
link = scheme.SchemeLink(
self.source_node, source_channel, self.sink_node, sink_channel
)
self.document.addLink(link)
stack.endMacro()
for channel_el in etree.findall("channels/channel"):
in_caption = channel_el.get("inWidgetCaption")
out_caption = channel_el.get("outWidgetCaption")
if in_caption in widgets_not_found or out_caption in widgets_not_found:
continue
source = nodes_by_caption[out_caption]
sink = nodes_by_caption[in_caption]
enabled = channel_el.get("enabled") == "1"
signals = literal_eval(channel_el.get("signals"))
for source_channel, sink_channel in signals:
try:
link = SchemeLink(
source, source_channel, sink, sink_channel, enabled=enabled
)
except (ValueError, IncompatibleChannelTypeError) as ex:
error_handler(ex)
else:
links.append(link)
settings = etree.find("settings")
properties = {}
if settings is not None:
data = settings.attrib.get("settingsDictionary", None)
if data and allow_pickle_data:
try:
properties = literal_eval(data)
except Exception:
log.error("Could not load properties for the scheme.", exc_info=True)
def edit_links(self, source_node, sink_node, initial_links=None):
"""
Show and execute the `EditLinksDialog`.
Optional `initial_links` list can provide a list of initial
`(source, sink)` channel tuples to show in the view, otherwise
the dialog is populated with existing links in the scheme (passing
an empty list will disable all initial links).
"""
status, links_to_add, links_to_remove = edit_links(
self.scheme, source_node, sink_node, initial_links, parent=self.document
)
if status == EditLinksDialog.Accepted:
links_to_add = [
scheme.SchemeLink(source_node, source_channel, sink_node, sink_channel)
for source_channel, sink_channel in links_to_add
]
links_to_remove = [
self.scheme.find_links(
source_node, source_channel, sink_node, sink_channel
)
for source_channel, sink_channel in links_to_remove
]
links_to_remove = reduce(list.__add__, links_to_remove, [])
conflicting = [
_f
for _f in [
conflicting_single_link(self.scheme, link) for link in links_to_add
]
for link_el in etree.findall("links/link"):
source_id = link_el.get("source_node_id")
sink_id = link_el.get("sink_node_id")
if source_id in nodes_not_found or sink_id in nodes_not_found:
continue
source = id_to_node.get(source_id)
sink = id_to_node.get(sink_id)
source_channel = link_el.get("source_channel")
sink_channel = link_el.get("sink_channel")
enabled = link_el.get("enabled") == "true"
try:
link = SchemeLink(
source, source_channel, sink, sink_channel, enabled=enabled
)
except (ValueError, IncompatibleChannelTypeError) as ex:
error_handler(ex)
else:
links.append(link)
# Load node properties
for property_el in etree.findall("node_properties/properties"):
node_id = property_el.attrib.get("node_id")
if node_id in nodes_not_found:
continue
node = id_to_node[node_id]
"""
#: Z value of the item
Z_VALUE = 0
#: Runtime link state value
#: These are pulled from SchemeLink.State for ease of binding to it's
#: state
State = SchemeLink.State
#: The link has no associated state.
NoState = SchemeLink.NoState
#: Link is empty; the source node does not have any value on output
Empty = SchemeLink.Empty
#: Link is active; the source node has a valid value on output
Active = SchemeLink.Active
#: The link is pending; the sink node is scheduled for update
Pending = SchemeLink.Pending
def __init__(self, *args):
self.__boundingRect = None
super().__init__(*args)
self.setFlag(QGraphicsItem.ItemHasNoContents, True)
self.setAcceptedMouseButtons(Qt.RightButton | Qt.LeftButton)
self.setAcceptHoverEvents(True)
self.setZValue(self.Z_VALUE)
self.sourceItem = None
self.sourceAnchor = None
self.sinkItem = None
self.sinkAnchor = None