Skip to content

ragraph.graph

Graph class module

The Graph class is the core of this package. It is the internal storage format that facilitates all conversions. These class definitions are designed to accommodate a wide range of graph formats.

Graph

Graph(
    nodes: Optional[Iterable[Node]] = None,
    edges: Optional[Iterable[Edge]] = None,
    add_parents: bool = False,
    add_children: bool = False,
    name: Optional[str] = None,
    kind: Optional[str] = None,
    labels: Optional[List[str]] = None,
    weights: Optional[Dict[str, Union[int, float]]] = None,
    annotations: Optional[
        Union[Annotations, Dict[str, Any]]
    ] = None,
    uuid: Optional[Union[str, UUID]] = None,
)

Bases: Metadata

Graph of Node objects and Edge objects between them.

Parameters:

Name Type Description Default
nodes Optional[Iterable[Node]]

Iterable of graph nodes.

None
edges Optional[Iterable[Edge]]

Iterable of graph edges.

None
add_parents bool

Recursively add parent nodes of the provided nodes.

False
add_children bool

Recursively add child nodes of the provided nodes.

False
name Optional[str]

Instance name. Given a UUID if none provided.

None
kind Optional[str]

Kind or main category of this instance.

None
labels Optional[List[str]]

Labels categorizing this instance.

None
weights Optional[Dict[str, Union[int, float]]]

Dictionary of weights attached to this instance.

None
annotations Optional[Union[Annotations, Dict[str, Any]]]

Miscellaneous properties of this instance.

None
uuid Optional[Union[str, UUID]]

Fixed UUID if desired, generated when left set to None.

None
Source code in ragraph/graph.py
def __init__(
    self,
    nodes: Optional[Iterable[Node]] = None,
    edges: Optional[Iterable[Edge]] = None,
    add_parents: bool = False,
    add_children: bool = False,
    name: Optional[str] = None,
    kind: Optional[str] = None,
    labels: Optional[List[str]] = None,
    weights: Optional[Dict[str, Union[int, float]]] = None,
    annotations: Optional[Union[Annotations, Dict[str, Any]]] = None,
    uuid: Optional[Union[str, UUID]] = None,
):
    Metadata.__init__(
        self,
        name=name,
        kind=kind,
        labels=labels,
        weights=weights,
        annotations=annotations,
        uuid=uuid,
    )
    self._nodes: Dict[str, Node] = OrderedDict()
    self._node_uuid_dict: Optional[Dict[UUID, Node]] = None  # Lazy
    self._edges: Dict[str, Edge] = OrderedDict()
    self._edge_uuid_dict: Optional[Dict[UUID, Edge]] = None  # lazy
    self._directed_edges: defaultdict = defaultdict(lambda: defaultdict(list))
    self._reversed_edges: defaultdict = defaultdict(lambda: defaultdict(list))
    self.add_parents = add_parents
    self.add_children = add_children
    setattr(self, "nodes", nodes)
    setattr(self, "edges", edges)

directed_edges property

directed_edges: Dict[str, Dict[str, List[Edge]]]

Nested edge dictionary from source name to target name to a list of edges.

edge_count property

edge_count: int

Number of edges in the graph.

edge_dict property

edge_dict: Dict[str, Edge]

Edge dictionary by edge name.

edge_gen property

edge_gen: Generator[Edge, None, None]

Yield edges.

edge_kinds property

edge_kinds: List[str]

List of unique edge kinds in the graph.

edge_labels property

edge_labels: List[str]

Alphabetically sorted list of unique edge labels in the graph.

edge_list property

edge_list: List[Edge]

Edges as a list.

edge_uuid_dict property

edge_uuid_dict: Dict[UUID, Edge]

Edge dictionary by node UUID.

edge_weight_labels property

edge_weight_labels: List[str]

List of unique edge weight labels in the graph.

edges property writable

edges: List[Edge]

Edges as a list.

json_dict property

json_dict: Dict[str, Any]

JSON dictionary representation.

leafs property

leafs: List[Node]

List of roots in the graph.

max_depth property

max_depth: int

Maximum node depth in the graph.

node_count property

node_count: int

Number of nodes in the graph.

node_dict property

node_dict: Dict[str, Node]

Node dictionary by node name.

node_gen property

node_gen: Generator[Node, None, None]

Yield nodes.

node_kinds property

node_kinds: List[str]

List of unique node kinds in the graph.

node_labels property

node_labels: List[str]

Alphabetically sorted list of unique node labels in the graph.

node_list property

node_list: List[Node]

Nodes as a list.

node_uuid_dict property

node_uuid_dict: Dict[UUID, Node]

Node dictionary by node UUID.

node_weight_labels property

node_weight_labels: List[str]

List of unique node weight labels in the graph.

nodes property writable

nodes: List[Node]

Nodes as a list.

reversed_edges property

reversed_edges: Dict[str, Dict[str, List[Edge]]]

Nested edge dictionary from target name to source name to a list of edges.

roots property

roots: List[Node]

List of roots in the graph.

add_edge

add_edge(edge: Edge)

Add a new edge to the graph. Source and target need to exist in the graph.

Parameters:

Name Type Description Default
edge Edge

Edge to add.

required

Raises:

Type Description
ValueError

if the source or target doesn't exist or the edge instance already does.

Source code in ragraph/graph.py
def add_edge(self, edge: Edge):
    """Add a new edge to the graph. Source and target need to exist in the graph.

    Arguments:
        edge: Edge to add.

    Raises:
        ValueError: if the source or target doesn't exist or the edge instance already does.
    """
    source = edge.source.name
    target = edge.target.name
    if source not in self.node_dict:
        raise ValueError("Source node %r does not exist in graph." % edge.source)
    if target not in self.node_dict:
        raise ValueError("Target node %r does not exist in graph." % edge.target)
    if edge.name in self._edges:
        if edge == self._edges[edge.name]:
            return
        raise ValueError(f"Edge name '{edge.name}' already exists in the graph.")
    if edge.uuid in self.edge_uuid_dict:
        if edge == self.edge_uuid_dict[edge.uuid]:
            return
        raise ValueError(f"Edge UUID '{edge.name}' already exists in the graph.")
    self._edges[edge.name] = edge
    self.edge_uuid_dict[edge.uuid] = edge
    self._directed_edges[source][target].append(edge)
    self._reversed_edges[target][source].append(edge)

add_node

add_node(node: Node)

Add a new node to this graph.

If the node instance already exists in the graph, it is ignored.

Parameters:

Name Type Description Default
node Node

Node to add.

required
Source code in ragraph/graph.py
def add_node(self, node: Node):
    """Add a new node to this graph.

    If the node instance already exists in the graph, it is ignored.

    Arguments:
        node: Node to add.
    """
    if not isinstance(node, Node):
        raise TypeError("%r is not an instance of Node but of %r." % (node, type(node)))

    # Append own node if not exists.
    if node.name in self._nodes:
        if node == self._nodes[node.name]:
            return
        raise ValueError(f"Another Node named '{node.name}' already exists in this Graph.")
    if node.uuid in self.node_uuid_dict:
        if node == self.node_uuid_dict[node.uuid]:
            return
        raise ValueError(f"Another Node with UUID '{node.uuid}' already exists in this Graph.")
    self._nodes[node.name] = node
    self.node_uuid_dict[node.uuid] = node

    # Append parent
    if self.add_parents and node.parent:
        self.add_node(node.parent)

    # Append children
    if self.add_children and node.children:
        for child in node.children:
            self.add_node(child)

as_dict

as_dict(use_uuid: bool = False) -> Dict[str, Any]

Return a copy as a (serializable) dictionary.

Parameters:

Name Type Description Default
use_uuid bool

Whether to use UUIDs instead of names.

False

Returns:

Name Type Description
nodes Dict[str, Any]

Node names or UUIDs to Node dictionaries.

kind Dict[str, Any]

Kind as str.

labels Dict[str, Any]

Labels as list of str.

weights Dict[str, Any]

Weights as dict.

annotations Dict[str, Any]

Annotations as a dictionary.

uuid Dict[str, Any]

UUID as str if toggled.

Source code in ragraph/graph.py
def as_dict(self, use_uuid: bool = False) -> Dict[str, Any]:
    """Return a copy as a (serializable) dictionary.

    Arguments:
        use_uuid: Whether to use UUIDs instead of names.

    Returns:
        nodes: Node names or UUIDs to Node dictionaries.
        kind: Kind as str.
        labels: Labels as list of str.
        weights: Weights as dict.
        annotations: Annotations as a dictionary.
        uuid: UUID as str if toggled.

    """
    if use_uuid:
        return dict(
            nodes={str(n.uuid): n.json_dict for n in self.nodes},
            edges={str(e.uuid): e.json_dict for e in self.edges},
            name=self.name,
            kind=self.kind,
            labels=self.labels,
            weights=self.weights,
            annotations=self.annotations.as_dict(),
            uuid=str(self.uuid),
        )
    else:
        return dict(
            nodes={n.name: n.as_dict(use_uuid=False) for n in self.nodes},
            edges={e.name: e.as_dict(use_uuid=False) for e in self.edges},
            name=self.name,
            kind=self.kind,
            labels=self.labels,
            weights=self.weights,
            annotations=self.annotations.as_dict(),
        )

check_consistency

check_consistency(raise_error: bool = True) -> bool

Check the consistency of this graph.

Parameters:

Name Type Description Default
raise_error bool

Whether to raise an error instead of returning a bool.

True

Returns:

Type Description
bool

Whether nodes aren't their own ancestor/descendant and if all their children

bool

and parents exist in the graph. Raises an error for inconsistencies if

bool

raise_error is set to True.

Source code in ragraph/graph.py
def check_consistency(self, raise_error: bool = True) -> bool:
    """Check the consistency of this graph.

    Arguments:
        raise_error: Whether to raise an error instead of returning a bool.

    Returns:
        Whether nodes aren't their own ancestor/descendant and if all their children
        and parents exist in the graph. Raises an error for inconsistencies if
        `raise_error` is set to `True`.
    """
    consistent = True

    try:
        for node in self.node_list:
            a = node
            while a.parent:
                if a.parent == node:
                    raise ConsistencyError("Node {} is in its own ancestors.".format(node))
                a = a.parent
            ds = node.children
            while ds:
                if node in ds:
                    raise ConsistencyError("Node {} is in its own descendants.".format(node))
                ds = [c for d in ds for c in d.children]
            if node.parent and node.parent not in self.node_list:
                raise ConsistencyError(
                    "Node {}'s parent {} is missing in the graph.".format(node, node.parent)
                )
            if node.parent and node not in node.parent.children:
                raise ConsistencyError(
                    "Node {}'s does not occur in parent {}'s children.".format(
                        node, node.parent
                    )
                )
            for child in node.children:
                if child not in self.node_list:
                    raise ConsistencyError(
                        "Node {}'s child {} is missing in the graph.".format(node, child)
                    )
                if child.parent != node:
                    raise ConsistencyError(
                        "Node {}'s child has a different parent {}.".format(node, child.parent)
                    )
            if node.is_bus and not node.parent:
                raise ConsistencyError(
                    "Node {} is a bus node, but has no parent to be it for.".format(node)
                )

        for edge in self.edge_list:
            if edge.source.name not in self.node_dict:
                raise ConsistencyError(f"Edge source not in Graph: {repr(edge.source)}")
            if edge.target.name not in self.node_dict:
                raise ConsistencyError(f"Edge target not in Graph: {repr(edge.source)}")

    except ConsistencyError as e:
        consistent = False
        if raise_error:
            raise e

    return consistent

del_edge

del_edge(edge: Union[Edge, str, UUID])

Remove an edge from the graph.

Source code in ragraph/graph.py
def del_edge(self, edge: Union[Edge, str, UUID]):
    """Remove an edge from the graph."""
    if isinstance(edge, str):
        edge = self._edges[edge]
    elif isinstance(edge, UUID):
        edge = self.edge_uuid_dict[edge]

    self.edge_uuid_dict.pop(edge.uuid)
    self._edges.pop(edge.name)

    source = edge.source.name
    target = edge.target.name

    self._directed_edges[source][target].remove(edge)
    if not self._directed_edges[source][target]:
        self._directed_edges[source].pop(target)

    self._reversed_edges[target][source].remove(edge)
    if not self._reversed_edges[source][target]:
        self._reversed_edges[source].pop(target)

del_node

1
2
3
del_node(
    node: Union[Node, str, UUID], inherit: bool = False
)

Remove a node and related edges from the graph by name or Node instance.

Parameters:

Name Type Description Default
node Union[Node, str, UUID]

Node to remove.

required
inherit bool

Whether to inherit hierarchy for underlying nodes to the parent.

False
Source code in ragraph/graph.py
def del_node(self, node: Union[Node, str, UUID], inherit: bool = False):
    """Remove a node and related edges from the graph by name or Node instance.

    Arguments:
        node: Node to remove.
        inherit: Whether to inherit hierarchy for underlying nodes to the parent.
    """
    if isinstance(node, str):
        node = self._nodes[node]
    elif isinstance(node, UUID):
        node = self.node_uuid_dict[node]

    if inherit and node.parent and node.children:
        for child in node.children:  # Is bus does not carry over to new parent.
            child.is_bus = False
        node.parent.children = node.parent.children + node.children
    node.children = []
    node.parent = None
    for e in list(self.edges_from(node)):
        self.del_edge(e)
    for e in list(self.edges_to(node)):
        self.del_edge(e)
    self.node_uuid_dict.pop(node.uuid)
    self._nodes.pop(node.name)

edges_between

1
2
3
4
5
6
edges_between(
    source: Union[str, Node],
    target: Union[str, Node],
    inherit: bool = False,
    loops: bool = True,
) -> Generator[Edge, None, None]

Yield edges between source and target node.

Parameters:

Name Type Description Default
source Union[str, Node]

Source node.

required
target Union[str, Node]

Target node.

required
inherit bool

Whether to include edges between descendants of given nodes.

False
loops bool

Whether to include self-loop edges.

True
Source code in ragraph/graph.py
def edges_between(
    self,
    source: Union[str, Node],
    target: Union[str, Node],
    inherit: bool = False,
    loops: bool = True,
) -> Generator[Edge, None, None]:
    """Yield edges between source and target node.

    Arguments:
        source: Source node.
        target: Target node.
        inherit: Whether to include edges between descendants of given nodes.
        loops: Whether to include self-loop edges.
    """
    if source == target and not loops:
        return
    source_node = source if isinstance(source, Node) else self._nodes[source]
    target_node = target if isinstance(target, Node) else self._nodes[target]
    sources, targets = [source_node.name], [target_node.name]
    if inherit:
        sources.extend(n.name for n in source_node.descendant_gen)
        targets.extend(n.name for n in target_node.descendant_gen)

    for source in sources:
        for target in targets:
            if source == target and not loops:
                continue
            yield from self._directed_edges[source][target]

edges_between_all

1
2
3
4
5
6
edges_between_all(
    sources: Iterable[Union[str, Node]],
    targets: Iterable[Union[str, Node]],
    inherit: bool = False,
    loops: bool = True,
) -> Generator[Edge, None, None]

Yield all edges between a set of source and a set of target nodes.

Parameters:

Name Type Description Default
sources Iterable[Union[str, Node]]

Source nodes.

required
targets Iterable[Union[str, Node]]

Target nodes.

required
inherit bool

Whether to include edges between descendants of given nodes.

False
loops bool

Whether to include self-loop edges.

True
Source code in ragraph/graph.py
def edges_between_all(
    self,
    sources: Iterable[Union[str, Node]],
    targets: Iterable[Union[str, Node]],
    inherit: bool = False,
    loops: bool = True,
) -> Generator[Edge, None, None]:
    """Yield all edges between a set of source and a set of target nodes.

    Arguments:
        sources: Source nodes.
        targets: Target nodes.
        inherit: Whether to include edges between descendants of given nodes.
        loops: Whether to include self-loop edges.
    """
    for source in sources:
        for target in targets:
            yield from self.edges_between(source, target, inherit=inherit, loops=loops)

edges_from

1
2
3
edges_from(
    source: Union[str, Node]
) -> Generator[Edge, None, None]

Yield edges originating from a given source node.

Source code in ragraph/graph.py
def edges_from(self, source: Union[str, Node]) -> Generator[Edge, None, None]:
    """Yield edges originating from a given source node."""
    if isinstance(source, Node):
        source = source.name
    for target in self._directed_edges[source]:
        yield from self._directed_edges[source][target]

edges_to

1
2
3
edges_to(
    target: Union[str, Node]
) -> Generator[Edge, None, None]

Yield edges towards a given target node.

Source code in ragraph/graph.py
def edges_to(self, target: Union[str, Node]) -> Generator[Edge, None, None]:
    """Yield edges towards a given target node."""
    if isinstance(target, Node):
        target = target.name
    for source in self._reversed_edges[target]:
        yield from self._reversed_edges[target][source]

get_adjacency_matrix

1
2
3
4
5
6
7
get_adjacency_matrix(
    nodes: Optional[Union[List[Node], List[str]]] = None,
    inherit: bool = False,
    loops: bool = False,
    only: Optional[List[str]] = None,
    convention: Convention = Convention.IR_FAD,
) -> Union[ndarray, List[List[float]]]

Convert graph data into a numerical adjacency matrix.

Parameters:

Name Type Description Default
nodes Optional[Union[List[Node], List[str]]]

Optional list of nodes for which to get the adjacency matrix.

None
inherit bool

Whether to count weights between children of the given nodes.

False
loops bool

Whether to calculate self-loops from a node to itself.

False
only Optional[List[str]]

Optional subset of edge weights to consider. See Edge for default edge weight implementation.

None

Returns:

Type Description
Union[ndarray, List[List[float]]]

Adjacency matrix as a 2D numpy array if numpy is present. Otherwise it will

Union[ndarray, List[List[float]]]

return a 2D nested list.

Source code in ragraph/graph.py
def get_adjacency_matrix(
    self,
    nodes: Optional[Union[List[Node], List[str]]] = None,
    inherit: bool = False,
    loops: bool = False,
    only: Optional[List[str]] = None,
    convention: Convention = Convention.IR_FAD,
) -> Union["np.ndarray", List[List[float]]]:
    """Convert graph data into a numerical adjacency matrix.

    Arguments:
        nodes: Optional list of nodes for which to get the adjacency matrix.
        inherit: Whether to count weights between children of the given nodes.
        loops: Whether to calculate self-loops from a node to itself.
        only: Optional subset of edge weights to consider. See [`Edge`][ragraph.edge.Edge] for
            default edge weight implementation.

    Returns:
        Adjacency matrix as a 2D numpy array if numpy is present. Otherwise it will
        return a 2D nested list.
    """
    from ragraph.io.matrix import to_matrix

    return to_matrix(
        self,
        rows=nodes,
        cols=nodes,
        inherit=inherit,
        loops=loops,
        only=only,
        convention=convention,
    )

get_ascii_art

1
2
3
4
5
6
7
get_ascii_art(
    nodes: Optional[List[Node]] = None,
    edge: str = "X",
    diag: str = "■",
    show: bool = False,
    convention: Convention = Convention.IR_FAD,
) -> Optional[str]

Get a unicode ASCII art representation of a binary adjacency matrix for a given set of nodes.

Parameters:

Name Type Description Default
nodes Optional[List[Node]]

Nodes to include in the art.

None
edge str

Mark to use when there's an edge between nodes. (X by default)

'X'
diag str

Mark to use on the diagonal.

'■'
show bool

Whether to directly print the ASCII art.

False

Returns:

Type Description
Optional[str]

ASCII art representation of this graph.

Source code in ragraph/graph.py
def get_ascii_art(
    self,
    nodes: Optional[List[Node]] = None,
    edge: str = "X",
    diag: str = "\u25A0",
    show: bool = False,
    convention: Convention = Convention.IR_FAD,
) -> Optional[str]:
    """Get a unicode ASCII art representation of a binary adjacency matrix for a
    given set of nodes.

    Arguments:
        nodes: Nodes to include in the art.
        edge: Mark to use when there's an edge between nodes. (X by default)
        diag: Mark to use on the diagonal.
        show: Whether to directly print the ASCII art.

    Returns:
        ASCII art representation of this graph.
    """
    if nodes is None:
        nodes = self.leafs
    if not nodes:
        raise ValueError("Empty graph, cannot create ASCII art.")

    namelens = [len(n.name) for n in nodes]
    maxlen = max(namelens, default=0)
    dim = len(nodes)

    ddd = "\u2500\u2500\u2500"  # ---
    pad = maxlen * " "

    # Grid lines
    topline = pad + "\u250C" + (dim - 1) * (ddd + "\u252C") + ddd + "\u2510"
    sepline = pad + "\u251C" + (dim - 1) * (ddd + "\u253C") + ddd + "\u2524"
    endline = pad + "\u2514" + (dim - 1) * (ddd + "\u2534") + ddd + "\u2518"

    lines = [topline]

    for i, row in enumerate(nodes):
        line = (maxlen - namelens[i]) * " " + row.name + "\u2525"
        for j, col in enumerate(nodes):
            if i == j:
                mark = diag
            else:
                if convention == Convention.IR_FAD:
                    src = col
                    tgt = row
                elif convention == Convention.IC_FBD:
                    src = row
                    tgt = col
                else:
                    raise ValueError("Unknown matrix convention type.")
                if self._directed_edges.get(src.name, dict()).get(tgt.name, False):
                    mark = edge
                else:
                    mark = " "
            line += " {} \u2502".format(mark)

        lines.append(line)
        lines.append(sepline)
    lines[-1] = endline
    txt = "\n".join(lines)
    if show:
        print(txt)
        return None
    else:
        return txt

get_edge_selection

1
2
3
4
get_edge_selection(
    nodes: Optional[List[Node]] = None,
    edge_kinds: Optional[List[str]] = None,
) -> List[Edge]

Select specific edges from this graph.

Parameters:

Name Type Description Default
nodes Optional[List[Node]]

The list of nodes between which the edges must be selected.

None
edge_kinds Optional[List[str]]

The kinds of edges to be selected.

None

Returns:

Type Description
List[Edge]

List of selected edges.

Source code in ragraph/graph.py
def get_edge_selection(
    self, nodes: Optional[List[Node]] = None, edge_kinds: Optional[List[str]] = None
) -> List[Edge]:
    """Select specific edges from this graph.

    Arguments:
        nodes: The list of nodes between which the edges must be selected.
        edge_kinds: The kinds of edges to be selected.

    Returns:
        List of selected edges.
    """

    nlist = nodes or self.get_node_selection(edge_kinds=edge_kinds)
    ekinds = edge_kinds or self.edge_kinds
    ekinds = sorted(set(ekinds))

    return [e for e in self.edges_between_all(sources=nlist, targets=nlist) if e.kind in ekinds]

get_edges_by_kind

get_edges_by_kind(kind: str) -> List[Edge]

Get edges by kind.

Source code in ragraph/graph.py
def get_edges_by_kind(self, kind: str) -> List[Edge]:
    """Get edges by kind."""
    return [edge for edge in self.edge_gen if edge.kind == kind]

get_graph_slice

1
2
3
4
5
get_graph_slice(
    nodes: Optional[Iterable[Node]] = None,
    edges: Optional[Iterable[Edge]] = None,
    inherit: bool = False,
) -> Graph

Return the graph slice with only the given nodes and the edges between those.

Parameters:

Name Type Description Default
nodes Optional[Iterable[Node]]

Optional set of nodes to include in the new deep copy. Defaults to all.

None
edges Optional[Iterable[Edge]]

Optional set of edges to include in the new deep copy. Defaults to all between selected nodes.

None
inherit bool

Whether to include child nodes of the given nodes and edges between them (unless edges are explicitly specified).

False
Note

This returns a deepcopy, so the copied nodes and edges do NOT share any changes! All parent and child nodes not in the list are excluded.

Source code in ragraph/graph.py
def get_graph_slice(
    self,
    nodes: Optional[Iterable[Node]] = None,
    edges: Optional[Iterable[Edge]] = None,
    inherit: bool = False,
) -> "Graph":
    """Return the graph slice with only the given nodes and the edges between those.

    Arguments:
        nodes: Optional set of nodes to include in the new deep copy.
            Defaults to all.
        edges: Optional set of edges to include in the new deep copy.
            Defaults to all between selected nodes.
        inherit: Whether to include child nodes of the given nodes and edges
            between them (unless edges are explicitly specified).

    Note:
        This returns a deepcopy, so the copied nodes and edges do NOT share any
        changes! All parent and child nodes not in the list are excluded.
    """
    nodes = self.nodes if nodes is None else list(nodes)
    edges = self.edges_between_all(nodes, nodes, inherit) if edges is None else list(edges)

    # Get a deepcopy to prevent changing nodes in the current graph instance.
    graph = deepcopy(Graph(nodes=nodes, edges=edges, add_children=inherit))

    # Reinstate relationships only between given nodes.
    graph_nodes = graph.nodes
    for node in graph_nodes:
        if node.parent:
            node.parent = graph[node.parent.name] if node.parent in graph.nodes else None
        node.children = [c for c in node.children if c in graph.nodes]
        if node.is_bus and not node.parent:
            node.is_bus = False

    return graph

get_hierarchy_dict

1
2
3
4
get_hierarchy_dict(
    roots: Optional[List[Node]] = None,
    levels: Optional[int] = None,
)

Return a dictionary of the (partial) hierarchical node structure.

Parameters:

Name Type Description Default
roots Optional[List[Node]]

Root nodes of the hierarchy to calculate.

None
levels Optional[int]

Number of levels to include below the roots.

None
Source code in ragraph/graph.py
def get_hierarchy_dict(self, roots: Optional[List[Node]] = None, levels: Optional[int] = None):
    """Return a dictionary of the (partial) hierarchical node structure.

    Arguments:
        roots: Root nodes of the hierarchy to calculate.
        levels: Number of levels to include below the roots.
    """
    if roots is None:
        roots = self.roots

    if levels is None:
        return {root.name: self.get_hierarchy_dict(root.children) for root in roots}
    elif levels > 0:
        return {
            root.name: self.get_hierarchy_dict(root.children, levels=levels - 1)
            for root in roots
        }

    return {root.name: dict() for root in roots}

get_kinds

1
2
3
get_kinds(
    elements: Union[Iterable[Node], Iterable[Edge]]
) -> List[str]

Get an alphabetically sorted list of unique node/edge kinds in the graph.

Source code in ragraph/graph.py
def get_kinds(self, elements: Union[Iterable[Node], Iterable[Edge]]) -> List[str]:
    """Get an alphabetically sorted list of unique node/edge kinds in the graph."""
    return sorted(set(elem.kind for elem in elements), key=str)

get_mapping_matrix

1
2
3
4
5
6
7
8
get_mapping_matrix(
    rows: Optional[Union[List[Node], List[str]]] = None,
    cols: Optional[Union[List[Node], List[str]]] = None,
    inherit: bool = False,
    loops: bool = False,
    only: Optional[List[str]] = None,
    convention: Convention = Convention.IR_FAD,
) -> Union[ndarray, List[List[float]]]

Convert graph data into a numerical mapping matrix.

Parameters:

Name Type Description Default
rows Optional[Union[List[Node], List[str]]]

Nodes representing the matrix rows.

None
cols Optional[Union[List[Node], List[str]]]

Nodes representing the matrix columns if different from the rows.

None
inherit bool

Whether to count weights between children of the given nodes.

False
loops bool

Whether to calculate self-loops from a node to itself.

False
only Optional[List[str]]

Optional subset of edge weights to consider. See Edge for default edge weight implementation.

None

Returns:

Type Description
Union[ndarray, List[List[float]]]

Adjacency matrix as a 2D numpy array if numpy is present. Otherwise it will

Union[ndarray, List[List[float]]]

return a 2D nested list.

Source code in ragraph/graph.py
def get_mapping_matrix(
    self,
    rows: Optional[Union[List[Node], List[str]]] = None,
    cols: Optional[Union[List[Node], List[str]]] = None,
    inherit: bool = False,
    loops: bool = False,
    only: Optional[List[str]] = None,
    convention: Convention = Convention.IR_FAD,
) -> Union["np.ndarray", List[List[float]]]:
    """Convert graph data into a numerical mapping matrix.

    Arguments:
        rows: Nodes representing the matrix rows.
        cols: Nodes representing the matrix columns if different from the rows.
        inherit: Whether to count weights between children of the given nodes.
        loops: Whether to calculate self-loops from a node to itself.
        only: Optional subset of edge weights to consider. See [`Edge`][ragraph.edge.Edge] for
            default edge weight implementation.

    Returns:
        Adjacency matrix as a 2D numpy array if numpy is present. Otherwise it will
        return a 2D nested list.
    """
    from ragraph.io.matrix import to_matrix

    return to_matrix(
        self,
        rows=rows,
        cols=cols,
        inherit=inherit,
        loops=loops,
        only=only,
        convention=convention,
    )

get_node_and_edge_selection

1
2
3
4
5
6
get_node_and_edge_selection(
    node_kinds: Optional[List[str]] = None,
    edge_kinds: Optional[List[str]] = None,
    depth: int = 2,
    selection_mode: str = "dependent",
) -> Tuple[List[Node], List[Edge]]

Select specific nodes and edges from this graph in a structured order.

Parameters:

Name Type Description Default
node_kinds Optional[List[str]]

The kind of nodes to be selected.

None
edge_kinds Optional[List[str]]

The kind of edges to be selected.

None
depth int

The maximum depth of node to be selected.

2
selection_mode str

The selection mode. Either 'dependent' or 'independent'.

'dependent'
Note

The selection mode argument determines how nodes of different kinds are selected. If the selection mode is set to 'dependent', the first node kind in the node_kinds list is considered to be the 'lead node kind'. Nodes of different kind than the lead node kind, are only selected if they have a dependency with at least one of the selected nodes of the lead node kind. If the selection mode is set to 'independent' this dependency condition is dropped.

Source code in ragraph/graph.py
def get_node_and_edge_selection(
    self,
    node_kinds: Optional[List[str]] = None,
    edge_kinds: Optional[List[str]] = None,
    depth: int = 2,
    selection_mode: str = "dependent",
) -> Tuple[List[Node], List[Edge]]:
    """Select specific nodes and edges from this graph in a structured order.

    Arguments:
        node_kinds: The kind of nodes to be selected.
        edge_kinds: The kind of edges to be selected.
        depth: The maximum depth of node to be selected.
        selection_mode: The selection mode. Either 'dependent' or 'independent'.

    Note:
        The selection mode argument determines how nodes of different kinds are
        selected. If the selection mode is set to 'dependent', the first node kind
        in the `node_kinds` list is considered to be the 'lead node kind'.
        Nodes of different kind than the lead node kind, are only selected if they
        have a dependency with at least one of the selected nodes of the lead node
        kind. If the selection mode is set to 'independent' this dependency
        condition is dropped.
    """

    nodes = self.get_node_selection(
        node_kinds=node_kinds,
        edge_kinds=edge_kinds,
        depth=depth,
        selection_mode=selection_mode,
    )

    edges = self.get_edge_selection(nodes=nodes, edge_kinds=edge_kinds)

    return nodes, edges

get_node_selection

1
2
3
4
5
6
get_node_selection(
    node_kinds: Optional[List[str]] = None,
    edge_kinds: Optional[List[str]] = None,
    depth: int = 2,
    selection_mode: str = "dependent",
) -> List[Node]

Select specific nodes from this graph in a structured order.

Parameters:

Name Type Description Default
node_kinds Optional[List[str]]

The kind of nodes to be selected.

None
edge_kinds Optional[List[str]]

The kind of edges to be selected.

None
depth int

The maximum depth of node to be selected.

2
selection_mode str

The selection mode. Either 'dependent' or 'independent'.

'dependent'
Note

The selection mode argument determines how nodes of different kinds are selected. If the selection mode is set to 'dependent', the first node kind in the node_kinds list is considered to be the 'lead node kind'. Nodes of different kind than the lead node kind, are only selected if they have a dependency with at least one of the selected nodes of the lead node kind. If the selection mode is set to 'independent' this dependency condition is dropped.

Source code in ragraph/graph.py
def get_node_selection(
    self,
    node_kinds: Optional[List[str]] = None,
    edge_kinds: Optional[List[str]] = None,
    depth: int = 2,
    selection_mode: str = "dependent",
) -> List[Node]:
    """Select specific nodes from this graph in a structured order.

    Arguments:
        node_kinds: The kind of nodes to be selected.
        edge_kinds: The kind of edges to be selected.
        depth: The maximum depth of node to be selected.
        selection_mode: The selection mode. Either 'dependent' or 'independent'.

    Note:
        The selection mode argument determines how nodes of different kinds are
        selected. If the selection mode is set to 'dependent', the first node kind
        in the `node_kinds` list is considered to be the 'lead node kind'.
        Nodes of different kind than the lead node kind, are only selected if they
        have a dependency with at least one of the selected nodes of the lead node
        kind. If the selection mode is set to 'independent' this dependency
        condition is dropped.
    """
    nkinds = node_kinds or self.node_kinds
    ekinds = edge_kinds or self.edge_kinds
    return utils.select_nodes(self, nkinds, ekinds, depth, selection_mode)

get_nodes_by_kind

get_nodes_by_kind(kind: str) -> List[Node]

Get nodes/edges by kind.

Source code in ragraph/graph.py
def get_nodes_by_kind(self, kind: str) -> List[Node]:
    """Get nodes/edges by kind."""
    return [node for node in self.node_gen if node.kind == kind]

get_weight_labels

1
2
3
get_weight_labels(
    elements: Union[Iterable[Node], Iterable[Edge]]
) -> List[str]

Get an alphabetically sorted list of unique labels of node/edge weights.

Source code in ragraph/graph.py
def get_weight_labels(self, elements: Union[Iterable[Node], Iterable[Edge]]) -> List[str]:
    """Get an alphabetically sorted list of unique labels of node/edge weights."""
    labels: Set[str] = set()
    for elem in elements:
        labels.update(elem.weights.keys())
    return sorted(labels, key=str)

sources_of

1
2
3
sources_of(
    target: Union[str, Node]
) -> Generator[Node, None, None]

Yield source nodes of a given target node.

Source code in ragraph/graph.py
def sources_of(self, target: Union[str, Node]) -> Generator[Node, None, None]:
    """Yield source nodes of a given target node."""
    if isinstance(target, Node):
        target = target.name
    yield from [
        self._nodes[source]
        for source in self._reversed_edges[target]
        if self._reversed_edges[target][source]
    ]

targets_of

1
2
3
targets_of(
    source: Union[str, Node]
) -> Generator[Node, None, None]

Yield target nodes of a given source node.

Source code in ragraph/graph.py
def targets_of(self, source: Union[str, Node]) -> Generator[Node, None, None]:
    """Yield target nodes of a given source node."""
    if isinstance(source, Node):
        source = source.name
    yield from [
        self._nodes[target]
        for target in self._directed_edges[source]
        if self._directed_edges[source][target]
    ]

GraphView

GraphView(
    graph: Graph,
    view_func: Optional[
        Callable[
            [Any],
            Tuple[
                Union[Iterable[Node], Dict[str, Node]],
                Union[Iterable[Edge], Dict[str, Edge]],
            ],
        ]
    ] = None,
    view_kwargs: Optional[Dict[str, Any]] = None,
)

Bases: Graph

A view on a Graph object. It works exactly like the Graph object, except for that the nodes and edges are filtered according to the view function.

Parameters:

Name Type Description Default
graph Graph

Graph to provide a view on.

required
view_func Optional[Callable[[Any], Tuple[Union[Iterable[Node], Dict[str, Node]], Union[Iterable[Edge], Dict[str, Edge]]]]]

View function that filters the Graph's nodes and edges.

None
view_kwargs Optional[Dict[str, Any]]

Optional arguments to pass to the view function.

None
Source code in ragraph/graph.py
def __init__(
    self,
    graph: Graph,
    view_func: Optional[
        Callable[
            [Any],
            Tuple[
                Union[Iterable[Node], Dict[str, Node]],
                Union[Iterable[Edge], Dict[str, Edge]],
            ],
        ]
    ] = None,
    view_kwargs: Optional[Dict[str, Any]] = None,
):
    Metadata.__init__(
        self,
        name=graph.name,
        kind=graph.kind,
        labels=graph.labels,
        weights=graph.weights,
        annotations=graph.annotations,
    )
    self._graph = graph
    self._view_func = view_func
    self._view_kwargs = view_kwargs

    self._vnodes: Optional[Dict[str, Node]] = None
    self._vnode_uuid_dict: Optional[Dict[UUID, Node]] = None
    self._vedges: Optional[Dict[str, Edge]] = None
    self._vedge_uuid_dict: Optional[Dict[UUID, Edge]] = None
    self._vdirected_edges: Optional[defaultdict] = None
    self._vreversed_edges: Optional[defaultdict] = None
    self.update()

graph property writable

graph: Graph

Graph to provide a view on.

view_func property writable

1
2
3
4
5
6
7
8
9
view_func: Optional[
    Callable[
        [Any],
        Tuple[
            Union[Iterable[Node], Dict[str, Node]],
            Union[Iterable[Edge], Dict[str, Edge]],
        ],
    ]
]

View function that filters the Graph's nodes and edges.

view_kwargs property writable

view_kwargs: Dict[str, Any]

Optional arguments to pass to the view function.

proxy

proxy()

Undo any filtering and just set this view to proxy the Graph 1:1.

Source code in ragraph/graph.py
def proxy(self):
    """Undo any filtering and just set this view to proxy the Graph 1:1."""
    self._vnodes, self._vedges = self.graph._nodes, self.graph._edges
    self._vnode_uuid_dict = self.graph.node_uuid_dict
    self._vedge_uuid_dict = self.graph.edge_uuid_dict
    self._vdirected_edges = self.graph._directed_edges
    self._vreversed_edges = self.graph._reversed_edges

reset

reset()

Reset the private properties after the original graph might have changed.

Source code in ragraph/graph.py
def reset(self):
    """Reset the private properties after the original graph might have changed."""
    self._vnodes = None
    self._vnode_uuid_dict = None
    self._vedges = None
    self._vedge_uuid_dict = None
    self._vdirected_edges = None
    self._vreversed_edges = None

update

update(**kwargs)

Apply the set view function with arguments.

Source code in ragraph/graph.py
def update(self, **kwargs):
    """Apply the set view function with arguments."""
    # Just proxy if no view function is set.
    if self.view_func is None:
        return self.proxy()

    # Get view filter results.
    view_kwargs = kwargs or self.view_kwargs
    nodes, edges = self.view_func(self.graph, **view_kwargs)

    # Cast to dicts.
    ndict = nodes if isinstance(nodes, dict) else {n.name: n for n in nodes}
    edict = edges if isinstance(edges, dict) else {e.name: e for e in edges}

    # Populate directed/reversed edges.
    directed_edges = defaultdict(lambda: defaultdict(list))
    reversed_edges = defaultdict(lambda: defaultdict(list))
    for e in edict.values():
        directed_edges[e.source.name][e.target.name].append(e)
        reversed_edges[e.target.name][e.source.name].append(e)

    # Update private views.
    self._vnodes, self._vedges = ndict, edict
    self._vnode_uuid_dict = {n.uuid: n for n in ndict.values()}
    self._vedge_uuid_dict = {e.uuid: e for e in edict.values()}
    self._vdirected_edges, self._vreversed_edges = directed_edges, reversed_edges

    # Save view kwargs after view succeeded.
    self._view_kwargs = view_kwargs