Source code for pybel.struct.mutation.expansion.neighborhood

# -*- coding: utf-8 -*-

"""Functions for expanding the neighborhoods of nodes."""

import itertools as itt
from typing import Iterable

from ...filters.node_predicates import is_pathology
from ...filters.typing import EdgeIterator
from ...graph import BELGraph
from ...pipeline import uni_in_place_transformation
from ...utils import update_metadata
from ....dsl import BaseEntity

__all__ = [
    "expand_node_predecessors",
    "expand_node_successors",
    "expand_node_neighborhood",
    "expand_nodes_neighborhoods",
    "expand_all_node_neighborhoods",
    "expand_internal",
]


[docs]@uni_in_place_transformation def expand_node_predecessors(universe: BELGraph, graph: BELGraph, node: BaseEntity): """Expand around the predecessors of the given node in the result graph. :param universe: The graph containing the stuff to add :param graph: The graph to add stuff to :param node: A BEL node """ skip_successors = set() for successor in universe.successors(node): if successor in graph: skip_successors.add(successor) continue graph.add_node_from_data(successor) graph.add_edges_from( (source, successor, key, data) for source, successor, key, data in universe.out_edges(node, data=True, keys=True) if successor not in skip_successors ) update_metadata(universe, graph)
[docs]@uni_in_place_transformation def expand_node_successors(universe: BELGraph, graph: BELGraph, node: BaseEntity) -> None: """Expand around the successors of the given node in the result graph. :param universe: The graph containing the stuff to add :param graph: The graph to add stuff to :param node: A BEL node """ skip_predecessors = set() for predecessor in universe.predecessors(node): if predecessor in graph: skip_predecessors.add(predecessor) continue graph.add_node_from_data(predecessor) graph.add_edges_from( (predecessor, target, key, data) for predecessor, target, key, data in universe.in_edges(node, data=True, keys=True) if predecessor not in skip_predecessors ) update_metadata(universe, graph)
[docs]@uni_in_place_transformation def expand_node_neighborhood(universe: BELGraph, graph: BELGraph, node: BaseEntity) -> None: """Expand around the neighborhoods of the given node in the result graph. Note: expands complexes' members :param universe: The graph containing the stuff to add :param graph: The graph to add stuff to :param node: A BEL node """ expand_node_predecessors(universe, graph, node) expand_node_successors(universe, graph, node)
[docs]@uni_in_place_transformation def expand_nodes_neighborhoods(universe: BELGraph, graph: BELGraph, nodes: Iterable[BaseEntity]) -> None: """Expand around the neighborhoods of the given node in the result graph. :param universe: The graph containing the stuff to add :param graph: The graph to add stuff to :param nodes: Nodes from the query graph """ for node in nodes: expand_node_neighborhood(universe, graph, node)
[docs]@uni_in_place_transformation def expand_all_node_neighborhoods(universe: BELGraph, graph: BELGraph, filter_pathologies: bool = False) -> None: """Expand the neighborhoods of all nodes in the given graph. :param pybel.BELGraph universe: The graph containing the stuff to add :param pybel.BELGraph graph: The graph to add stuff to :param filter_pathologies: Should expansion take place around pathologies? """ for node in list(graph): if filter_pathologies and is_pathology(node): continue expand_node_neighborhood(universe, graph, node)
[docs]@uni_in_place_transformation def expand_internal( universe: BELGraph, graph: BELGraph, ) -> None: """Expand on edges between entities in the sub-graph that pass the given filters, in place. :param universe: The full graph :param graph: A sub-graph to find the upstream information """ for u, v, key in iterate_internal(universe, graph): graph.add_edge(u, v, key=key, **universe[u][v][key])
def iterate_internal(universe: BELGraph, graph: BELGraph) -> EdgeIterator: """Iterate over edges that are in the universe but not the target graph.""" for u, v in itt.product(graph, repeat=2): if graph.has_edge(u, v) or not universe.has_edge(u, v): continue for key in universe[u][v]: yield u, v, key