compositional_graph_models/data.py
def typed_topological_nodes_generator(
    graph: dgl.DGLGraph, node_mask: Optional[torch.Tensor] = None
) -> List[torch.Tensor]:
    """
    Generates a topological traversal for the nodes of `graph`, guaranteeing the
    following two properties:
        - A node is visited only if all of its descendants have been visited.
        - At each step, all nodes visited have the same function type (FUNCTION_FIELD).

    Additionally, the algorithm will heuristically attempt to maximize parallelism by
    greedily selecting the node type with the most available nodes at each step.

    Parameters
    ----------
    graph
        The graph whose nodes to traverse. May have multiple independent graphs
        packaged together with `dgl.batch()`.
    node_mask
        If provided, any node with a 0 in this mask is ignored.
        Must be a bool tensor or None.

    Returns
    -------
    List[Tensor]
        The i-th element in this list is a tensor consisting of the nodes to visit
        at step i. The union of these lists is exactly the node indices of `graph`.
    """
    visited = torch.zeros(graph.num_nodes(), dtype=torch.bool, device=graph.device)
    remaining_dependencies = graph.in_degrees()
    order: List[torch.Tensor] = []

    if node_mask is not None:
        opposite_mask = ~node_mask
        visited |= opposite_mask
        remaining_dependencies *= node_mask
        (opposite_indices,) = torch.nonzero(opposite_mask, as_tuple=True)
        _, masked_successors = graph.out_edges(opposite_indices)
        visit_counts = torch.bincount(masked_successors, minlength=graph.num_nodes())
        remaining_dependencies -= visit_counts

    while not visited.all():
        # Find unvisited nodes with no unfulfilled dependencies
        available = visited.bitwise_not() & (remaining_dependencies == 0)

        # Greedily select the type with the most available nodes
        available_types = graph.ndata[FUNCTION_FIELD].masked_select(available)
        step_type, _ = available_types.mode()

        # Find all nodes of this type with no additional dependencies
        step_mask = available & (graph.ndata[FUNCTION_FIELD] == step_type)

        # Flag these nodes as visited, convert the mask to indices, and add to the order
        visited |= step_mask
        (step_indices,) = torch.nonzero(step_mask, as_tuple=True)
        order.append(step_indices)

        # Tick down the dependency count of these nodes' successors
        _, successor_indices = graph.out_edges(step_indices)
        visit_counts = torch.bincount(successor_indices, minlength=graph.num_nodes())
        remaining_dependencies -= visit_counts

    return order

    Generates a topological traversal for the nodes of `graph`, guaranteeing the
    following two properties:
        - A node is visited only if all of its descendants have been visited.
        - At each step, all nodes visited have the same function type (FUNCTION_FIELD).

    Additionally, the algorithm will heuristically attempt to maximize parallelism by
    greedily selecting the node type with the most available nodes at each step.

    Parameters
    ----------
    graph
        The graph whose nodes to traverse. May have multiple independent graphs
        packaged together with `dgl.batch()`.
    node_mask
        If provided, any node with a 0 in this mask is ignored.
        Must be a bool tensor or None.

    Returns
    -------
    List[Tensor]
        The i-th element in this list is a tensor consisting of the nodes to visit
        at step i. The union of these lists is exactly the node indices of `graph`.
    
Generates a topological traversal for the nodes of `graph`, guaranteeing the
following two properties:
A node is visited only if all of its descendants have been visited.
At each step, all nodes visited have the same function type (FUNCTION_FIELD).

Additionally, the algorithm will heuristically attempt to maximize parallelism by
greedily selecting the node type with the most available nodes at each step.

Parameters

graph
The graph whose nodes to traverse. May have multiple independent graphs
packaged together with `dgl.batch()`.
node_mask
If provided, any node with a 0 in this mask is ignored.
Must be a bool tensor or None.

Returns

List[Tensor]
The i-th element in this list is a tensor consisting of the nodes to visit
at step i. The union of these lists is exactly the node indices of `graph`.
Generates a topological traversal for the nodes of ` graph ` guaranteeing the following two properties : A node is visited only if all of its descendants have been visited . At each step all nodes visited have the same function type ( FUNCTION_FIELD ) . Additionally the algorithm will heuristically attempt to maximize parallelism by greedily selecting the node type with the most available nodes at each step . Parameters graph The graph whose nodes to traverse . May have multiple independent graphs packaged together with ` dgl . batch () ` . node_mask If provided any node with a 0 in this mask is ignored . Must be a bool tensor or None . Returns List [ Tensor ] The i - th element in this list is a tensor consisting of the nodes to visit at step i . The union of these lists is exactly the node indices of ` graph ` .
def typed_topological_nodes_generator(
    graph: dgl.DGLGraph, node_mask: Optional[torch.Tensor] = None
) -> List[torch.Tensor]:
    visited = torch.zeros(graph.num_nodes(), dtype=torch.bool, device=graph.device)
    remaining_dependencies = graph.in_degrees()
    order: List[torch.Tensor] = []
    if node_mask is not None:
        opposite_mask = ~node_mask
        visited |= opposite_mask
        remaining_dependencies *= node_mask
        (opposite_indices,) = torch.nonzero(opposite_mask, as_tuple=True)
        _, masked_successors = graph.out_edges(opposite_indices)
        visit_counts = torch.bincount(masked_successors, minlength=graph.num_nodes())
        remaining_dependencies -= visit_counts
    while not visited.all():
        available = visited.bitwise_not() & (remaining_dependencies == 0)
        available_types = graph.ndata[FUNCTION_FIELD].masked_select(available)
        step_type, _ = available_types.mode()
        step_mask = available & (graph.ndata[FUNCTION_FIELD] == step_type)
        visited |= step_mask
        (step_indices,) = torch.nonzero(step_mask, as_tuple=True)
        order.append(step_indices)
        _, successor_indices = graph.out_edges(step_indices)
        visit_counts = torch.bincount(successor_indices, minlength=graph.num_nodes())
        remaining_dependencies -= visit_counts
    return order
def typed_topological_nodes_generator ( graph : dgl . DGLGraph , node_mask : Optional [ torch . Tensor ] = None ) -> List [ torch . Tensor ] : visited = torch . zeros ( graph . num_nodes ( ) , dtype = torch . bool , device = graph . device ) remaining_dependencies = graph . in_degrees ( ) order : List [ torch . Tensor ] = [ ] if node_mask is not None : opposite_mask = ~ node_mask visited |= opposite_mask remaining_dependencies *= node_mask ( opposite_indices , ) = torch . nonzero ( opposite_mask , as_tuple = True ) _ , masked_successors = graph . out_edges ( opposite_indices ) visit_counts = torch . bincount ( masked_successors , minlength = graph . num_nodes ( ) ) remaining_dependencies -= visit_counts while not visited . all ( ) : available = visited . bitwise_not ( ) & ( remaining_dependencies == 0 ) available_types = graph . ndata [ FUNCTION_FIELD ] . masked_select ( available ) step_type , _ = available_types . mode ( ) step_mask = available & ( graph . ndata [ FUNCTION_FIELD ] == step_type ) visited |= step_mask ( step_indices , ) = torch . nonzero ( step_mask , as_tuple = True ) order . append ( step_indices ) _ , successor_indices = graph . out_edges ( step_indices ) visit_counts = torch . bincount ( successor_indices , minlength = graph . num_nodes ( ) ) remaining_dependencies -= visit_counts return order
Generates a topological traversal for the nodes of `graph`, guaranteeing the
following two properties:
 A node is visited only if all of its descendants have been visited.
Generates a topological traversal for the nodes of ` graph ` guaranteeing the following two properties : A node is visited only if all of its descendants have been visited .
"""
    Generates a topological traversal for the nodes of `graph`, guaranteeing the
    following two properties:
        - A node is visited only if all of its descendants have been visited.
        - At each step, all nodes visited have the same function type (FUNCTION_FIELD).

    Additionally, the algorithm will heuristically attempt to maximize parallelism by
    greedily selecting the node type with the most available nodes at each step.

    Parameters
    ----------
    graph
        The graph whose nodes to traverse. May have multiple independent graphs
        packaged together with `dgl.batch()`.
    node_mask
        If provided, any node with a 0 in this mask is ignored.
        Must be a bool tensor or None.

    Returns
    -------
    List[Tensor]
        The i-th element in this list is a tensor consisting of the nodes to visit
        at step i. The union of these lists is exactly the node indices of `graph`.
    """ # Find unvisited nodes with no unfulfilled dependencies # Greedily select the type with the most available nodes # Find all nodes of this type with no additional dependencies # Flag these nodes as visited, convert the mask to indices, and add to the order # Tick down the dependency count of these nodes' successors
graph
dgl.DGLGraph
node_mask
Optional[torch.Tensor]