Graph DAG Module

The DAG (Directed Acyclic Graph) class provides functionality for working with fully directed acyclic graphs, commonly used to represent causal structures.

Classes

DAG

Directed Acyclic Graph class for representing fully oriented causal structures.

Features:

• Fully directed edges only
• Strict acyclicity enforcement
• Topological ordering
• String representation (bnlearn format)
• Causal model representation

Usage:

from causaliq_core.graph import DAG

# Create a DAG
nodes = ['X', 'Y', 'Z']
edges = [
    ('X', '->', 'Y'),    # X causes Y
    ('Y', '->', 'Z'),    # Y causes Z
]
dag = DAG(nodes, edges)

# Get topological ordering
for node in dag.ordered_nodes():
    print(f"Node: {node}")

# Get string representation
print(dag.to_string())  # e.g. [X][Y|X][Z|Y]

NotDAGError

Exception raised when attempting to create an invalid DAG (e.g., contains cycles).

Reference

Classes:

• NotDAGError –

  Indicate graph is not a DAG when one is expected.

• DAG –

  Directed Acyclic Graph (DAG).

Classes

NotDAGError

Indicate graph is not a DAG when one is expected.

DAG

DAG(nodes: List[str], edges: List[Tuple[str, str, str]])

Directed Acyclic Graph (DAG).

Parameters:

• nodes

  (List[str]) –

  Nodes present in the graph.

• edges

  (List[Tuple[str, str, str]]) –

  Edges which define the graph connections as list of tuples: (node1, dependency symbol, node2).

Attributes:

• nodes (List[str]) –

  Graph nodes in alphabetical order.

• edges (Dict[Tuple[str, str], EdgeType]) –

  Graph edges {(node1, node2): EdgeType}.

• is_directed (bool) –

  Always True for DAGs.

• parents (Dict[str, List[str]]) –

  Parents of node {node: [parents]}.

Raises:

• TypeError –

  If nodes and edges not both lists.

• ValueError –

  If node or edge invalid.

• NotDAGError –

  If graph is not a DAG.

Parameters:

• nodes

  (List[str]) –

  Nodes present in the graph.

• edges

  (List[Tuple[str, str, str]]) –

  Edges which define the graph connections.

Raises:

• NotDAGError –

  If graph is not a DAG.

Methods:

• ordered_nodes –

  Generator which returns nodes in a topological order.

• to_string –

  Compact (bnlearn) string representation of DAG e.g. [A][B][C|A:B].

• __str__ –

  Return a human-readable description of the DAG.

• rename –

  Rename nodes in place according to name map.

• partial_order –

  Return partial topological ordering for the directed part of a

• is_DAG –

  Return whether graph is a Directed Acyclic Graph (DAG).

• is_PDAG –

  Return whether graph is a Partially Directed Acyclic Graph (PDAG).

• undirected_trees –

  Return undirected trees present in graph.

• components –

  Return components present in graph.

• number_components –

  Return number of components (including unconnected nodes) in graph.

• to_adjmat –

  Return an adjacency matrix representation of the graph.

• __eq__ –

  Test if graph is identical to this one.

• compress –

  Compress graph to compact binary representation.

• decompress –

  Decompress graph from compact binary representation.

• edge_reversible –

  Return whether specified edge is in CPDAG and is reversible.

Functions

ordered_nodes

ordered_nodes() -> Generator[str, None, None]

Generator which returns nodes in a topological order.

Yields:

• str –

  Next node in topological order.

Raises:

• RuntimeError –

  If the graph has cycles (shouldn't happen for a DAG).

to_string

to_string() -> str

Compact (bnlearn) string representation of DAG e.g. [A][B][C|A:B].

Returns:

• str –

  Description of graph.

__str__

__str__() -> str

Return a human-readable description of the DAG.

Returns:

• str –

  Description of DAG.

rename

rename(name_map: Dict[str, str]) -> None

Rename nodes in place according to name map.

Parameters:

• name_map (Dict[str, str]) –

  Name mapping {name: new name}. Must have mapping for every node.

Raises:

• TypeError –

  With bad arg type.

• ValueError –

  With bad arg values e.g. unknown node names.

partial_order classmethod

partial_order(
    parents: Dict[str, List[str]],
    nodes: Optional[Union[List[str], Set[str]]] = None,
    new_arc: Optional[Tuple[str, str]] = None,
) -> Optional[List[Set[str]]]

Return partial topological ordering for the directed part of a graph.

The graph is specified by list of parents for each node.

Parameters:

• parents (Dict[str, List[str]]) –

  Parents of each node {node: [parents]}.

• nodes (Optional[Union[List[str], Set[str]]], default: None ) –

  Optional complete list of nodes including parentless ones for use if parents argument doesn't include them already.

• new_arc (Optional[Tuple[str, str]], default: None ) –

  A new arc (n1, n2) to be added before order is evaluated. If the opposing arc is implied in parents then it is removed so that arc reversal is also supported. This argument facilitates seeing whether an arc addition or reversal would create a cycle.

Returns:

• Optional[List[Set[str]]] –

  Nodes in a partial topological order as list of sets or None if

• Optional[List[Set[str]]] –

  there is no ordering which means the graph is cyclic.

is_DAG

is_DAG() -> bool

Return whether graph is a Directed Acyclic Graph (DAG).

Returns:

• bool –

  True if graph is a DAG, False otherwise.

is_PDAG

is_PDAG() -> bool

Return whether graph is a Partially Directed Acyclic Graph (PDAG).

Returns:

• bool –

  True if graph is a PDAG, False otherwise.

undirected_trees

undirected_trees() -> List[Set[Union[Tuple[str, str], Tuple[str, None]]]]

Return undirected trees present in graph.

Returns:

• List[Set[Union[Tuple[str, str], Tuple[str, None]]]] –

  List of trees, each tree a set of tuples representing edges in tree

• List[Set[Union[Tuple[str, str], Tuple[str, None]]]] –

  (n1, n2) or a single isolated node (n1, None).

components

components() -> List[List[str]]

Return components present in graph.

Uses tree search algorithm to span the undirected graph to identify nodes in individual trees which are the spanning tree of each component.

Returns:

• List[List[str]] –

  List of lists, each a list of sorted nodes in component.

number_components

number_components() -> int

Return number of components (including unconnected nodes) in graph.

Returns:

• int –

  Number of components.

to_adjmat

to_adjmat() -> DataFrame

Return an adjacency matrix representation of the graph.

Returns:

• DataFrame –

  Adjacency matrix as a pandas DataFrame.

__eq__

__eq__(other: object) -> bool

Test if graph is identical to this one.

Parameters:

• other (object) –

  Graph to compare with self.

Returns:

• bool –

  True if other is identical to self.

compress

compress() -> bytes

Compress graph to compact binary representation.

Format:

• 2 bytes: number of nodes (uint16, big-endian)
• For each node: 2 bytes name length + UTF-8 encoded name
• 2 bytes: number of edges (uint16, big-endian)
• For each edge: 2 bytes source index + 2 bytes target index + 1 byte edge type

Returns:

• bytes –

  Compact binary representation of the graph.

Raises:

• ValueError –

  If graph has more than 65535 nodes or edges.

decompress classmethod

decompress(data: bytes) -> SDG

Decompress graph from compact binary representation.

Parameters:

• data (bytes) –

  Binary data from SDG.compress().

Returns:

• SDG –

  Reconstructed SDG instance.

Raises:

• TypeError –

  If data is not bytes.

• ValueError –

  If data is invalid or corrupted.

edge_reversible

edge_reversible(edge: Tuple[str, str]) -> bool

Return whether specified edge is in CPDAG and is reversible.

Parameters:

• edge (Tuple[str, str]) –

  Edge to examine, (node1, node2).

Returns:

• bool –

  Whether present and reversible, or not.

Raises:

• TypeError –

  If edge argument has bad type.