[core] Simplify handling of link assignment by value on Attribute

Centralize the handling of link assigment by value in a dedicated
private method.
Store link assignment by value in a separate member variable
`_linkExpression`, to avoid manipulating a internal `_value` that
can represent different concepts (value VS serialized edge expression).
This commit is contained in:
Yann Lanthony 2025-02-21 18:56:45 +01:00
parent 8ee7b50204
commit 6f0542c07e
3 changed files with 96 additions and 64 deletions

View file

@ -16,7 +16,7 @@ import meshroom.core
from meshroom.common import BaseObject, DictModel, Slot, Signal, Property
from meshroom.core import Version
from meshroom.core.attribute import Attribute, ListAttribute, GroupAttribute
from meshroom.core.exception import GraphCompatibilityError, StopGraphVisit, StopBranchVisit
from meshroom.core.exception import GraphCompatibilityError, InvalidEdgeError, StopGraphVisit, StopBranchVisit
from meshroom.core.graphIO import GraphIO, GraphSerializer, TemplateGraphSerializer, PartialGraphSerializer
from meshroom.core.node import BaseNode, Status, Node, CompatibilityNode
from meshroom.core.nodeFactory import nodeFactory
@ -485,41 +485,38 @@ class Graph(BaseObject):
node._applyExpr()
return node
def copyNode(self, srcNode, withEdges=False):
def copyNode(self, srcNode: Node, withEdges: bool=False):
"""
Get a copy instance of a node outside the graph.
Args:
srcNode (Node): the node to copy
withEdges (bool): whether to copy edges
srcNode: the node to copy
withEdges: whether to copy edges
Returns:
Node, dict: the created node instance,
a dictionary of linked attributes with their original value (empty if withEdges is True)
The created node instance and the mapping of skipped edge per attribute (always empty if `withEdges` is True).
"""
def _removeLinkExpressions(attribute: Attribute, removed: dict[Attribute, str]):
"""Recursively remove link expressions from the given root `attribute`."""
# Link expressions are only stored on input attributes.
if attribute.isOutput:
return
if attribute._linkExpression:
removed[attribute] = attribute._linkExpression
attribute._linkExpression = None
elif isinstance(attribute, (ListAttribute, GroupAttribute)):
for child in attribute.value:
_removeLinkExpressions(child, removed)
with GraphModification(self):
# create a new node of the same type and with the same attributes values
# keep links as-is so that CompatibilityNodes attributes can be created with correct automatic description
# (File params for link expressions)
node = nodeFactory(srcNode.toDict(), srcNode.nodeType) # use nodeType as name
# skip edges: filter out attributes which are links by resetting default values
node = nodeFactory(srcNode.toDict(), name=srcNode.nodeType)
skippedEdges = {}
if not withEdges:
for n, attr in node.attributes.items():
if attr.isOutput:
# edges are declared in input with an expression linking
# to another param (which could be an output)
continue
# find top-level links
if Attribute.isLinkExpression(attr.value):
skippedEdges[attr] = attr.value
attr.resetToDefaultValue()
# find links in ListAttribute children
elif isinstance(attr, (ListAttribute, GroupAttribute)):
for child in attr.value:
if Attribute.isLinkExpression(child.value):
skippedEdges[child] = child.value
child.resetToDefaultValue()
for _, attr in node.attributes.items():
_removeLinkExpressions(attr, skippedEdges)
return node, skippedEdges
def duplicateNodes(self, srcNodes):
@ -850,13 +847,16 @@ class Graph(BaseObject):
return set(self._nodes) - nodesWithInputLink
@changeTopology
def addEdge(self, srcAttr, dstAttr):
assert isinstance(srcAttr, Attribute)
assert isinstance(dstAttr, Attribute)
if srcAttr.node.graph != self or dstAttr.node.graph != self:
raise RuntimeError('The attributes of the edge should be part of a common graph.')
def addEdge(self, srcAttr: Attribute, dstAttr: Attribute):
if not (srcAttr.node.graph == dstAttr.node.graph == self):
raise InvalidEdgeError(
srcAttr.fullNameToGraph, dstAttr.fullNameToGraph, "Attributes do not belong to this Graph"
)
if dstAttr in self.edges.keys():
raise RuntimeError('Destination attribute "{}" is already connected.'.format(dstAttr.getFullNameToNode()))
raise InvalidEdgeError(
srcAttr.fullNameToNode, dstAttr.fullNameToNode, "Destination is already connected"
)
edge = Edge(srcAttr, dstAttr)
self.edges.add(edge)
self.markNodesDirty(dstAttr.node)