Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _get_path_nodes(nodes: Iterable[BaseNode], algorithm: Algorithm
) -> Tuple[List[Tuple[int, int]], List[BaseNode]]:
"""Calculates the contraction paths using `opt_einsum` methods.
Args:
nodes: An iterable of nodes.
algorithm: `opt_einsum` method to use for calculating the contraction path.
Returns:
The optimal contraction path as returned by `opt_einsum`.
"""
sorted_nodes = sorted(nodes, key=lambda n: n.signature)
input_sets = [set(node.edges) for node in sorted_nodes]
output_set = get_subgraph_dangling(nodes)
size_dict = {edge: edge.dimension for edge in get_all_edges(nodes)}
return algorithm(input_sets, output_set, size_dict), sorted_nodes
if final node has more than one edge,
`output_edge_order` must be pronvided.
ignore_edge_order: An option to ignore the output edge
order.
Returns:
Final node after full contraction.
"""
nodes_set = set(nodes)
edges = get_all_edges(nodes_set)
#output edge order has to be determinded before any contraction
#(edges are refreshed after contractions)
if not ignore_edge_order:
if output_edge_order is None:
output_edge_order = list(get_subgraph_dangling(nodes))
if len(output_edge_order) > 1:
raise ValueError("The final node after contraction has more than "
"one remaining edge. In this case `output_edge_order` "
"has to be provided.")
if set(output_edge_order) != get_subgraph_dangling(nodes):
raise ValueError(
"output edges are not equal to the remaining "
"non-contracted edges of the final node."
)
for edge in edges:
if not edge.is_disabled: #if its disabled we already contracted it
if edge.is_trace():
nodes_set.remove(edge.node1)
nodes_set.add(contract_parallel(edge))
Final node after full contraction.
"""
nodes_set = set(nodes)
edges = get_all_edges(nodes_set)
#output edge order has to be determinded before any contraction
#(edges are refreshed after contractions)
if not ignore_edge_order:
if output_edge_order is None:
output_edge_order = list(get_subgraph_dangling(nodes))
if len(output_edge_order) > 1:
raise ValueError("The final node after contraction has more than "
"one remaining edge. In this case `output_edge_order` "
"has to be provided.")
if set(output_edge_order) != get_subgraph_dangling(nodes):
raise ValueError(
"output edges are not equal to the remaining "
"non-contracted edges of the final node."
)
for edge in edges:
if not edge.is_disabled: #if its disabled we already contracted it
if edge.is_trace():
nodes_set.remove(edge.node1)
nodes_set.add(contract_parallel(edge))
if len(nodes_set) == 1:
# There's nothing to contract.
if ignore_edge_order:
return list(nodes_set)[0]
return list(nodes_set)[0].reorder_edges(output_edge_order)