Skip to main content

causal_graph

to_list​

def to_list(var: Any) -> List[Any]

Helper to make sure a var is always a list.

reset_cached_attributes_decorator​

def reset_cached_attributes_decorator(func: Callable) -> Callable

Decorator to reset cached attributes of CausalGraph.

Whenever a function is called that changes the graph, we need to reset these attributes.

Skeleton​

from cai_causal_graph.causal_graph import Skeleton

class Skeleton(CanDictSerialize, CanDictDeserialize)

A utility class to obtain the skeleton of the causal graph including nodes, edges and adjacency matrix.

Methods​

__init__​

def __init__(graph: CausalGraph)

Arguments:

__eq__​

def __eq__(other: object, deep: bool = False) -> bool

Check if equal to another Skeleton.

Checks if all nodes and edges are equal.

Arguments:

  • other: The other Skeleton to compare to.
  • deep: If True, also does deep equality checks on all the nodes and edges. Default is False.

Returns:

True if equal, False otherwise.

__ne__​

def __ne__(other: object) -> bool

Check if the skeleton is not equal to another skeleton.

__iter__​

def __iter__() -> Iterator

Return item,value tuples of the content of self.to_dict().

nodes​

@property
def nodes() -> List[Node]

Return a list of all nodes.

get_node​

def get_node(identifier: NodeLike) -> Node

Return node that matches the identifier.

get_node_names​

def get_node_names() -> List[str]

Return a list of each node's identifier.

node_exists​

def node_exists(identifier: NodeLike) -> bool

Check if node exists.

edges​

@property
def edges() -> List[Edge]

Return a list of all edges. All edges will be of the type cai_causal_graph.type_definitions.EdgeType.UNDIRECTED_EDGE because this class represents a skeleton, which only has undirected edges and no directed edges.

get_edge​

def get_edge(source: str, destination: str) -> Edge

Return edge by source and destination.

As all edges are undirected, the order of source and destination does not matter.

edge_exists​

def edge_exists(source: str, destination: str) -> bool

Return true if edge exists.

get_edge_pairs​

def get_edge_pairs() -> List[PAIR_T]

Return all edge pairs in the current graph.

get_edge_by_pair​

def get_edge_by_pair(pair: PAIR_T) -> Edge

Return edge by pair identifier.

is_edge_by_pair​

def is_edge_by_pair(pair: PAIR_T) -> bool

Check if a given edge exists by pair identifier.

get_neighbors​

def get_neighbors(node: NodeLike) -> List[str]

Get node identifiers for all neighbor nodes for a specific node.

get_neighbor_nodes​

def get_neighbor_nodes(node: NodeLike) -> List[Node]

Get all neighbor nodes for a specific node.

is_empty​

def is_empty() -> bool

Return True if there are no nodes and edges. False otherwise.

adjacency_matrix​

@property
def adjacency_matrix() -> numpy.ndarray

Return the adjacency matrix of the Skeleton instance.

from_adjacency_matrix​

@classmethod
def from_adjacency_matrix(
        cls,
        adjacency: numpy.ndarray,
        node_names: Optional[List[Union[NodeLike, int]]] = None,
        graph_class: Union[Type[CausalGraph], None] = None) -> Skeleton

Instantiate a Skeleton object from an adjacency matrix.

Arguments:

  • adjacency: The adjacency matrix.
  • node_names: A list of the node names. These should be in the order of the rows and columns of the adjacency matrix.
  • graph_class: The appropriate subclass of CausalGraph to use. If None, which is the default, CausalGraph will be used. This is to ensure the new Skeleton object has the correct node class type.

Returns:

A new Skeleton based on the provided adjacency matrix.

to_dict​

def to_dict(include_meta: bool = True) -> dict

Serialize the Skeleton instance to a dictionary.

Arguments:

  • include_meta: Whether to include meta information about the skeleton in the dictionary. Default is True.

Returns:

The dictionary representation of the Skeleton instance.

to_networkx​

def to_networkx() -> networkx.Graph

Return a networkx.Graph corresponding to the Skeleton instance.

to_numpy​

def to_numpy() -> Tuple[numpy.ndarray, List[str]]

Return a numpy array that represents the adjacency matrix of the Skeleton instance.

To avoid confusion, this method also returns a list of variables that corresponds to the order of columns in the numpy array.

to_gml_string​

def to_gml_string() -> str

Return a Graph Modelling Language (GML) string representative of the Skeleton instance.

from_dict​

@classmethod
def from_dict(cls,
              d: dict,
              graph_class: Union[Type[CausalGraph], None] = None) -> Skeleton

Instantiate a Skeleton object from a dictionary.

Arguments:

  • d: The dictionary representation of the skeleton.
  • graph_class: The appropriate subclass of CausalGraph to use. If None, which is the default, CausalGraph will be used. This is to ensure the new Skeleton object has the correct node class type.

Returns:

A new Skeleton based on the provided dictionary.

from_networkx​

@classmethod
def from_networkx(
        cls,
        g: networkx.Graph,
        graph_class: Union[Type[CausalGraph], None] = None) -> Skeleton

Instantiate a Skeleton object from a networkx.Graph.

Arguments:

  • g: The networkx.Graph representing the skeleton.
  • graph_class: The appropriate subclass of CausalGraph to use. If None, which is the default, CausalGraph will be used. This is to ensure the new Skeleton object has the correct node class type.

Returns:

A new Skeleton based on the networkx.Graph.

from_gml_string​

@classmethod
def from_gml_string(
        cls,
        gml: str,
        graph_class: Union[Type[CausalGraph], None] = None) -> Skeleton

Instantiate a Skeleton object from a Graph Modelling Language (GML) string.

Arguments:

  • gml: The GML string representing the skeleton.
  • graph_class: The appropriate subclass of CausalGraph to use. If None, which is the default, CausalGraph will be used. This is to ensure the new Skeleton object has the correct node class type.

Returns:

A new Skeleton based on the provided GML string.

copy​

def copy() -> Skeleton

Copy a Skeleton instance.

__repr__​

def __repr__() -> str

Return a string description of the Skeleton instance.

details​

def details() -> str

Return a detailed string description of the Skeleton instance.

__hash__​

def __hash__() -> int

Return a hash representation of the Skeleton instance.

CausalGraph​

from cai_causal_graph.causal_graph import CausalGraph

class CausalGraph(HasIdentifier, HasMetadata, CanDictSerialize,
                  CanDictDeserialize)

A low-level class that uniquely defines the state of a causal graph.

Attributes​

  • _NodeCls: Type[Node]
  • _EdgeCls: Type[Edge]

Methods​

__init__​

def __init__(input_list: Optional[List[NodeLike]] = None,
             output_list: Optional[List[NodeLike]] = None,
             fully_connected: bool = True,
             meta: Optional[dict] = None)

The CausalGraph class manages and defines the state of a causal graph.

It encodes and allows for easy visualization of causal relationships, which are presented as edges between nodes where each node is a variable from a data set.

It can deal with a variety of different edge types such as directed, undirected, bidirected and unknown edges. Therefore, this class can represent a large variety of causal graph classes, such as (among others):

- Directed acyclic graph (DAG)
- Completed partially directed acyclic graph (CPDAG)
- Maximal ancenstral graph (MAG)
- Partial ancestral graph (PAG)

Example:

 from cai_causal_graph import CausalGraph

 # create an empty graph by not providing a `input_list` or `output_list`
 causal_graph = CausalGraph()

 # add nodes to the causal graph
 node_names = ['input1', 'input2', 'input3', 'target1', 'target2']
 causal_graph.add_nodes_from(node_names)

 # add a directed edge from 'input1' to 'output1'
 causal_graph.add_edge('input1', 'output1')
By default, any edges added will be directed edges, e.g. 'input1' -> 'target1' for the edge added above. It
is possible to specify different edge types via the `edge_type` argument. For the full list of edge types,
see `cai_causal_graph.type_definitions.EdgeType`. For instance, an undirected edge can be added:
 # add an undirected edge between 'input1' and 'input2'
 causal_graph.add_edge('input1', 'input2', edge_type=EdgeType.UNDIRECTED_EDGE)
It is straightforward to export an instantiated `cai_causal_graph.causal_graph.CausalGraph` to a
serializable dictionary,
 causal_graph_dict = causal_graph.to_dict()

Arguments:

  • input_list: List of objects coercable to Node. Each element is treated as an input node, if full_connected parameter is True. Otherwise, the nodes will simply be added to the graph with no edges.
  • output_list: List of objects coercable to Node. Each element is treated as an output node, if fully_connected parameter is True. Otherwise, the nodes will simply be added to the graph with no edges.
  • fully_connected: If set to True (default), create a fully-connected bipartite directed graph, with all inputs connected to all outputs. If no input_list and no output_list is provided, an empty graph will be created. If either or both are provided, but this is False, then the nodes will be added but not connected by edges.
  • meta: Any metadata defined on the graph. The keys must be strings, but no requirement is placed on the values of the dictionary. Default is None. If passed, meta is shallow-copied.

__copy__​

def __copy__() -> CausalGraph

Copy a CausalGraph instance.

__deepcopy__​

def __deepcopy__(memo) -> CausalGraph

Deep-copy a CausalGraph instance.

__getitem__​

def __getitem__(
        item: Union[NodeLike, Tuple[NodeLike, NodeLike]]) -> Union[Node, Edge]

Get a node (single identifier) or edge (tuple of identifiers).

__iter__​

def __iter__() -> Iterator

Return item,value tuples of the content of self.to_dict().

__eq__​

def __eq__(other: object, deep: bool = False) -> bool

Check if equal to another CausalGraph.

Checks if all nodes and edges are equal.

Arguments:

  • other: The other CausalGraph to compare to.
  • deep: If True, also does deep equality checks on all the nodes and edges. Default is False.

Returns:

True if equal, False otherwise.

__ne__​

def __ne__(other: object) -> bool

Check if the graph is not equal to another graph.

nodes​

@property
def nodes() -> List[Node]

Return a list of nodes.

edges​

@property
def edges() -> List[Edge]

Return a list of edges.

skeleton​

@property
def skeleton() -> Skeleton

Return a Skeleton instance of the causal graph.

adjacency_matrix​

@property
def adjacency_matrix() -> numpy.ndarray

Return the adjacency matrix of the causal graph.

The column order is dependent upon the ordering of node names. To obtain both, adjacency matrix and node names in corresponding order use the to_numpy method instead.

sepsets​

@property
def sepsets() -> dict

Return a dictionary of separation sets between variables, potentially obtained via conditional independence testing or via domain knowledge.

identifier​

@property
def identifier() -> str

Return a unique identifier for a graph by concatenating all node identifiers in topological order. If the causal graph is not a DAG, in which case the topological order is undefined, a sorted list of node names is returned.

get_identifier​

def get_identifier() -> str

Return the graph identifier.

is_dag​

def is_dag() -> bool

Check whether the CausalGraph instance is a Directed Acyclic Graph (DAG).

is_empty​

def is_empty() -> bool

Return True if there are no nodes and edges. False otherwise.

has_non_serializable_metadata​

@property
def has_non_serializable_metadata() -> bool

Return True whether the graph or any node or edge contains non-JSON serializable metadata.

By default, this returns False. Please note that by default CausalGraph does not perform any checks on the metadata. Instead, this method enables any derived classes to implement custom checks.

get_node​

def get_node(identifier: NodeLike) -> Node

Return a node based on its identifier.

get_nodes​

def get_nodes(
    identifier: Optional[Union[NodeLike,
                               List[NodeLike]]] = None) -> List[Node]

Return nodes matching the given identifier(s).

If an identifier is provided then only that node will be returned (if it exists; an empty list will be returned otherwise). If a list of identifiers is provided, then a corresponding list of nodes will be returned. If the identifier is left as None (default) then all nodes will be returned.

Arguments:

  • identifier: String or list of string identifier(s) of the node(s) to find, or None to return all nodes.

Returns:

List of matching nodes. If a list of identifiers was provided, the order of nodes matches that of the provided identifiers. Otherwise, they are sorted alphabetically by identifier.

get_node_names​

def get_node_names() -> List[str]

Return a list of each node's identifier.

get_inputs​

def get_inputs() -> List[Node]

Get all nodes without ancestors.

get_outputs​

def get_outputs() -> List[Node]

Get all nodes without descendants.

node_exists​

def node_exists(identifier: NodeLike) -> bool

Check whether the specified node exists.

add_node​

@reset_cached_attributes_decorator
def add_node(identifier: Optional[NodeLike] = None,
             *,
             variable_type: NodeVariableType = NodeVariableType.UNSPECIFIED,
             meta: Optional[dict] = None,
             node: Optional[Node] = None) -> Node

Add a node to the causal graph but do not connect it to anything.

Arguments:

  • identifier: String that uniquely identifies the node within the causal graph. If a node with that identifier already exists within the network, an exception will be raised. Can be None, if a node parameter is specified.
  • variable_type: The variable type that the node represents. The choices are available through the NodeVariableType enum.
  • meta: The meta values for the node. If passed explicitly, this metadata is not deepcopied.
  • node: A Node node to be used to construct a new node. All the properties of the provided node will be deep copied to the constructed node, including metadata and variable type. If provided, then all other parameters to the method must not be specified. Default is None.

Returns:

The created node.

add_nodes_from​

@reset_cached_attributes_decorator
def add_nodes_from(identifiers: List[NodeLike])

A convenience method to add multiple nodes. Only allows to set up nodes with default setup.

For more details how nodes are being set refer to CausalGraph.add_node() method.

Arguments:

  • identifiers: List of valid node identifiers.

add_fully_connected_nodes​

@reset_cached_attributes_decorator
def add_fully_connected_nodes(inputs: List[NodeLike], outputs: List[NodeLike])

Create directed edges between all inputs and all outputs.

delete_node​

@reset_cached_attributes_decorator
def delete_node(identifier: NodeLike)

Delete a node from the causal graph. This also deletes all edges connecting to this node.

Arguments:

  • identifier: String identifying the node to be deleted.

remove_node​

@reset_cached_attributes_decorator
def remove_node(identifier: NodeLike)

Remove a node from the causal graph. This also deletes all edges connecting to this node.

Arguments:

  • identifier: String identifying the node to be deleted.

replace_node​

@reset_cached_attributes_decorator
def replace_node(
        node_id: NodeLike,
        new_node_id: Optional[NodeLike] = None,
        *,
        variable_type: NodeVariableType = NodeVariableType.UNSPECIFIED,
        meta: Optional[dict] = None)

Replace an existing node by a new one, copying over all inbound and outbound edges and removing original node.

If any additional information, such as variable type or meta is specified, it is used for the new node. Otherwise, these are copied from the original node. If new node is not provided, original one is simply changed in-place.

Arguments:

  • node_id: original node to be replaced and removed
  • new_node_id: new node to replace the old one
  • variable_type: The variable type that the node represents. The choices are available through the NodeVariableType enum.
  • meta: optional meta to be used for the new node

get_edge​

def get_edge(source: NodeLike,
             destination: NodeLike,
             *,
             edge_type: Optional[EdgeType] = None) -> Edge

Return an edge based on its source, destination and (optional) edge_type.

get_edges​

def get_edges(source: Optional[NodeLike] = None,
              destination: Optional[NodeLike] = None,
              *,
              edge_type: Optional[EdgeType] = None) -> List[Edge]

Return edges matching the given identifiers.

Both source and destination can either be provided or left as None. If either is provided, only edges originating from or terminating at that node will be returned. Otherwise, that identifier will match against edges originating from or terminating at any node. Specifying edge_type yields edges that only match the desired edge type. Otherwise, if set to None, edges with any edge type are returned.

Arguments:

  • source: Identifier of the source node. If set to None (default), edges originating from any node are returned.
  • destination: Identifier of the destination node. If set to None (default), edges terminating at any node are returned.
  • edge_type: The edge type of the desired edges to be returned. If set to None (default), edges with any edge_type are returned.

Returns:

List of matching edges, sorted alphabetically by source identifier and then by destination identifier.

get_edge_by_pair​

def get_edge_by_pair(pair: Tuple[NodeLike, NodeLike],
                     edge_type: Optional[EdgeType] = None) -> Edge

Return an edge based on a tuple of (source, destination) and an edge_type.

is_edge_by_pair​

def is_edge_by_pair(pair: Tuple[NodeLike, NodeLike],
                    edge_type: Optional[EdgeType] = None) -> bool

Check if a given edge exists by pair identifier.

get_directed_edges​

def get_directed_edges() -> List[Edge]

Returns a list of directed edges, e.g. 'X' -> 'Y', in the causal graph.

get_undirected_edges​

def get_undirected_edges() -> List[Edge]

Returns a list of undirected edges, e.g. 'X' -- 'Y', in the causal graph.

get_bidirected_edges​

def get_bidirected_edges() -> List[Edge]

Returns a list of bidirectional edges, e.g. 'X' <-> 'Y', in the causal graph.

get_unknown_edges​

def get_unknown_edges() -> List[Edge]

Returns a list of edges that are unknown, e.g. 'X' oo 'Y', in the causal graph.

get_unknown_directed_edges​

def get_unknown_directed_edges() -> List[Edge]

Returns a list of edges that are unknown-directed, e.g. 'X' o> 'Y', in the causal graph.

get_unknown_undirected_edges​

def get_unknown_undirected_edges() -> List[Edge]

Returns a list of edges that are unknown-undirected, e.g. 'X' o- 'Y', in the causal graph.

edge_exists​

def edge_exists(source: NodeLike,
                destination: NodeLike,
                *,
                edge_type: Optional[EdgeType] = None) -> bool

Returns True if the edge exists. If edge_type is None (default), this ignores edge types.

get_edge_pairs​

def get_edge_pairs() -> List[PAIR_T]

Return all edge pairs in the current graph.

change_edge_type​

@reset_cached_attributes_decorator
def change_edge_type(source: NodeLike, destination: NodeLike,
                     new_edge_type: EdgeType)

Change an edge type for a specific edge.

Changing an edge type may affect the topology of the causal graph, as it changes the relationship between source and destination nodes.

Please note that this method removes existing edge and replaces it with a new edge with an updated edge type, while keeping reference to all other attributes (e.g. metadata) unmodified.

Arguments:

  • source: Source of the edge for which the type should be changed.
  • destination: Destination of the edge for which the type should be changed.
  • new_edge_type: New edge type of the edge. If the new edge type matches existing edge type, no action is performed.

add_edge​

@reset_cached_attributes_decorator
def add_edge(source: Optional[NodeLike] = None,
             destination: Optional[NodeLike] = None,
             *,
             edge_type: EdgeType = EdgeType.DIRECTED_EDGE,
             meta: Optional[dict] = None,
             edge: Optional[Edge] = None,
             validate: bool = True) -> Edge

Add an edge from a source to a destination node with a specific edge type.

If these two nodes are already connected in any way, then an error will be raised. An error will also be raised if the new edge would create a cyclic connection of directed edges. In this case, the CausalGraph instance will be restored to its original state and the edge will not be added. It is possible to specify an edge type from source to destination as well, with the default being a forward directed edge, i.e., source -> destination.

Arguments:

  • source: String identifying the node from which the edge will originate. Can be None, if an edge parameter is specified.
  • destination: String identifying the node at which the edge will terminate. Can be None, if an edge parameter is specified.
  • edge_type: The type of the edge to be added. Default is cai_causal_graph.type_definitions.EdgeType.DIRECTED_EDGE. See EdgeType for the list of possible edge types.
  • meta: The meta values for the edge. If passed explicitly, this metadata is not deepcopied.
  • edge: A Edge edge to be used to construct a new edge. All the properties of the provided edge will be deep copied to the constructed edge, including metadata. If provided, then all other parameters to the method must not be specified. Default is None.
  • validate: Whether to perform validation checks. The validation checks will raise if any cycles are introduced to the graph by adding the edge. There is no guarantees about the behavior of the resulting graph if this is disabled specifically to introduce cycles. This should only be used to speed up this method in situations where it is known the new edge will not add cycles, for example when copying a graph. Default is True.

Returns:

The created edge object.

add_edges_from​

@reset_cached_attributes_decorator
def add_edges_from(pairs: List[Tuple[NodeLike, NodeLike]],
                   validate: bool = True)

A convenience method to add multiple edges by specifying tuples of source and destination node identifiers.

Only allows to set up edges with default setup. For more details on how edges are being set, refer to add_edge method.

Arguments:

  • pairs: List of valid edge pairs, defined as tuples of (source_identifier, destination_identifier).
  • validate: Whether to perform validation checks. The validation checks will raise if any cycles are introduced to the graph by adding the edge. There is no guarantees about the behavior of the resulting graph if this is disabled specifically to introduce cycles. This should only be used to speed up this method in situations where it is known the new edge will not add cycles, for example when copying a graph. Default is True.

add_edges_from_paths​

@reset_cached_attributes_decorator
def add_edges_from_paths(paths: Union[List[NodeLike], List[List[NodeLike]]],
                         validate: bool = True)

A convenience method to add multiple edges by specifying a single or a list of paths.

Importantly, the path can overlap with each other, or with existing causal graph structure. Meaning, if an edge specified on the path already exists, an error is not raised. For example, the following two paths parameters will add the same edges: [['a','b','c'],['b','c','d']] and [['a','b','c'],['c','d']].

However, conflicting paths will produce an error. For example, it is not possible to add the following two paths: [['a','b','c'],['c','b','y']], because the first path defines an edge ('b','c'), while the second defines the edge ('c','b').

Only allows to set up edges with default setup. For more details on how edges are being set, refer to add_edge method.

Arguments:

  • paths: A list of paths or a single path. A path is defined as a list of node identifiers, defining the causal path in a causal graph.
  • validate: Whether to perform validation checks. The validation checks will raise if any cycles are introduced to the graph by adding the edge. There is no guarantees about the behavior of the resulting graph if this is disabled specifically to introduce cycles. This should only be used to speed up this method in situations where it is known the new edge will not add cycles, for example when copying a graph. Default is True.

add_edge_by_pair​

@reset_cached_attributes_decorator
def add_edge_by_pair(pair: Tuple[NodeLike, NodeLike],
                     edge_type: EdgeType = EdgeType.DIRECTED_EDGE,
                     meta: Optional[dict] = None,
                     validate: bool = True)

Add edge by pair identifier (source, destination).

Arguments:

  • pair: Tuple to identify the source and destination of the edge.
  • edge_type: The type of edge to add. Default is a directed edge
  • meta: Any metadata to add to the edge. Default is None (no metadata).
  • validate: Whether to perform validation checks. The validation checks will raise if any cycles are introduced to the graph by adding the edge. There is no guarantees about the behavior of the resulting graph if this is disabled specifically to introduce cycles. This should only be used to speed up this method in situations where it is known the new edge will not add cycles, for example when copying a graph. Default is True.

remove_edge_by_pair​

@reset_cached_attributes_decorator
def remove_edge_by_pair(pair: Tuple[NodeLike, NodeLike],
                        edge_type: Optional[EdgeType] = None)

Remove edge by pair identifier (source, destination).

delete_edge​

@reset_cached_attributes_decorator
def delete_edge(source: NodeLike,
                destination: NodeLike,
                *,
                edge_type: Optional[EdgeType] = None)

Delete an edge from the causal graph.

This will raise an exception if the edge does not exist, or either of the nodes do not exist.

Arguments:

  • source: Identifier of the node from which the edge originates.
  • destination: Identifier of the node at which the edge terminates.
  • edge_type: The edge type of the edge to be deleted. Default is None, in which case the type is ignored.

remove_edge​

@reset_cached_attributes_decorator
def remove_edge(source: NodeLike,
                destination: NodeLike,
                *,
                edge_type: Optional[EdgeType] = None)

Remove a specific edge by source and destination node identifiers, as well as edge type.

replace_edge​

@reset_cached_attributes_decorator
def replace_edge(source: NodeLike,
                 destination: NodeLike,
                 new_source: NodeLike,
                 new_destination: NodeLike,
                 *,
                 edge_type: Optional[EdgeType] = None,
                 meta: Optional[dict] = None)

Replace an existing edge by a new one.

If any additional information, such as edge type or meta is specified, it is used for the new edge. Otherwise, these are copied from the original edge.

Note that if you simply wish to change the type of an existing edge, you should use the change_edge_type method instead.

Arguments:

  • source: Source of the edge to be replaced.
  • destination: Destination of the edge to be replaced.
  • new_source: Source of the new edge.
  • new_destination: Destination of the new edge.
  • edge_type: The edge type of the new edge. If None (default), the same edge type as the original edge will be used.
  • meta: Optional meta to be used for the new edge. If None (default)`, the meta of the original edge will be used.

get_neighbors​

def get_neighbors(node: NodeLike) -> List[str]

Get node identifiers for all neighbor nodes for a specific node.

Note: It does not matter what the edge type is, as long as there is an edge between node and another node, that other node is considered its neighbor.

get_neighbor_nodes​

def get_neighbor_nodes(node: NodeLike) -> List[Node]

Get all neighbor nodes for a specific node.

Note: It does not matter what the edge type is, as long as there is an edge between node and another node, that other node is considered its neighbor.

get_children_nodes​

def get_children_nodes(node: NodeLike) -> List[Node]

Get all children nodes for a specific node.

get_children​

def get_children(node: NodeLike) -> List[str]

Get node identifiers for all children nodes for a specific node.

get_children_graph​

def get_children_graph(node: NodeLike) -> CausalGraph

Get a sub causal graph that only includes the children of a specific node and the node itself.

This method returns a star graph, such that all edges are going from the provided node. Any edges between the children nodes are not present in the returned graph.

get_parent_nodes​

def get_parent_nodes(node: NodeLike) -> List[Node]

Get all parent nodes for a specific node.

get_parents​

def get_parents(node: NodeLike) -> List[str]

Get node identifiers for all parent nodes for a specific node.

get_parents_graph​

def get_parents_graph(node: NodeLike) -> CausalGraph

Get a sub causal graph that only includes the parents of a specific node and the node itself.

This method returns a star graph, such that all edges are going to the provided node. Any edges between the parents nodes are not present in the returned graph.

get_ancestors​

def get_ancestors(node: NodeLike) -> Set[str]

Get all ancestors of a node.

This method is only applicable if the graph is a valid DAG (i.e., all edges are directed).

get_ancestral_graph​

def get_ancestral_graph(node: NodeLike) -> CausalGraph

Get a sub causal graph that only includes the ancestors of a specific node and the node itself.

get_descendants​

def get_descendants(node: NodeLike) -> Set[str]

Get all descendants of a node.

This method is only applicable if the graph is a valid DAG (i.e., all edges are directed).

get_descendant_graph​

def get_descendant_graph(node: NodeLike) -> CausalGraph

Get a sub causal graph that only includes the descendants of a specific node and the node itself.

is_ancestor​

def is_ancestor(
        ancestor_node: NodeLike,
        descendant_node: Union[NodeLike, Set[NodeLike],
                               List[NodeLike]]) -> bool

Check whether there is a causal path between ancestor and descendant node(s).

This method is only applicable if the graph is a valid DAG (i.e., all edges are directed).

Arguments:

  • ancestor_node: node identifier-coercible object of a potential ancestor node.
  • descendant_node: a single or a set/list of node identifier-coercible objects.

Returns:

whether all nodes passed in the 'descendant_node' parameter are descendants of the ancestor node. Hence, if some nodes are descendants and some are not, False is returned.

is_descendant​

def is_descendant(
        descendant_node: NodeLike,
        ancestor_node: Union[NodeLike, Set[NodeLike], List[NodeLike]]) -> bool

Check whether there is a causal path between descendant and ancestor node(s).

This method is only applicable if the graph is a valid DAG (i.e., all edges are directed).

Arguments:

  • descendant_node: node identifier-coercible object of a potential descendant node.
  • ancestor_node: a single or a set/list of node identifier-coercible objects.

Returns:

whether all nodes passed in the 'ancestor_node' parameter are ancestors of the descendant node. Hence, if some nodes are ancestors and some are not, False is returned.

get_common_ancestors​

def get_common_ancestors(node_1: NodeLike, node_2: NodeLike) -> Set[str]

Get all common ancestors for two nodes.

This method is only applicable if the graph is a valid DAG (i.e., all edges are directed).

If one of the provided nodes is an ancestor of another, it will not appear in the returned set.

Arguments:

  • node_1: node identifier-coercible object specifying a node in the graph.
  • node_2: node identifier-coercible object specifying a node in the graph.

Returns:

set of all node identifiers which are ancestors of both nodes.

get_common_descendants​

def get_common_descendants(node_1: NodeLike, node_2: NodeLike) -> Set[str]

Get all common descendants for two nodes.

This method is only applicable if the graph is a valid DAG (i.e., all edges are directed).

If one of the provided nodes is a descendant of another, it will not appear in the returned set.

Arguments:

  • node_1: node identifier-coercible object specifying a node in the graph.
  • node_2: node identifier-coercible object specifying a node in the graph.

Returns:

set of all node identifiers which are descendants of both nodes.

get_d_separation_set​

def get_d_separation_set(node_1: NodeLike, node_2: NodeLike) -> Set[str]

Return a minimal d-separation set between two nodes.

This method is only applicable if the graph is a valid DAG (i.e., all edges are directed).

Arguments:

  • node_1: a single node-identifier coercible object.
  • node_2: a single node-identifier coercible object.

Returns:

a set of node identifiers for a minimal d-separation set between the provided nodes.

is_d_separated​

def is_d_separated(
    nodes_1: Union[Set[NodeLike], List[NodeLike], NodeLike],
    nodes_2: Union[Set[NodeLike], List[NodeLike], NodeLike],
    separation_set: Optional[Union[Set[NodeLike],
                                   List[NodeLike]]] = None) -> bool

Check whether given sets of nodes are d-separated given the separation set.

This method is only applicable if the graph is a valid DAG (i.e., all edges are directed).

Arguments:

  • nodes_1: a set/list of or a single node-identifier coercible object(s).
  • nodes_2: a set/list of or a single node-identifier coercible object(s).
  • separation_set: a set/list of node-identifier coercible objects.

Returns:

whether the separation set is d-separating for the provided nodes.

is_minimally_d_separated​

def is_minimally_d_separated(
        node_1: NodeLike,
        node_2: NodeLike,
        separation_set: Optional[Set[NodeLike]] = None) -> bool

Check whether given nodes are minimally d-separated given the separation set.

This method is only applicable if the graph is a valid DAG (i.e., all edges are directed).

Arguments:

  • node_1: a single node-identifier coercible object.
  • node_2: a single node-identifier coercible object.
  • separation_set: a set/list of node-identifier coercible objects.

Returns:

whether the separation set is d-separating for the provided nodes.

get_topological_order​

def get_topological_order(
        return_all: bool = False) -> Union[List[str], List[List[str]]]

Return either a single or all topological orders of the graph.

A topological order is a non-unique permutation of the nodes such that an edge from 'A' to 'B' implies that 'A' appears before 'B' in the topological sort order. Generating all possible topological orders may be expensive for large graphs.

It is only possible to get topological order if the graph is a valid DAG.

Arguments:

  • return_all: whether to generate all topological orders.

Returns:

either a list of strings identifying a single topological order, or a list of lists identifying all possible topological orders.

get_all_causal_paths​

def get_all_causal_paths(source: NodeLike,
                         destination: NodeLike) -> List[List[str]]

Get all causal paths between the provided source and destination.

This method is only applicable if the graph is a valid DAG (i.e., all edges are directed).

Arguments:

  • source: source node-identifier coercible object.
  • destination: destination node-identifier coercible object.

Returns:

a list of lists specifying the node identifiers on the path between source and destination (this is inclusive of the source and destination node identifiers). An empty list is returned if no paths are available.

directed_path_exists​

def directed_path_exists(source: NodeLike, destination: NodeLike) -> bool

Check whether there exists a directed path between the source and destination.

Unlike the CausalGraph.get_all_causal_paths method, this works for mixed causal graphs as well (note that non-directed edges are ignored).

Arguments:

  • source: source node-identifier coercible object.
  • destination: destination node-identifier coercible object.

Returns:

True if there exists a directed path between the source and destination, False otherwise.

coerce_to_nodelike​

@staticmethod
def coerce_to_nodelike(node: Union[NodeLike, int]) -> NodeLike

Coerce an object to be used for node construction.

If the provided node is already cai_causal_graph.type_definitions.NodeLike, it is simply returned. If provided node is an integer, it is converted to a string and returned. If anything else, it will raise a TypeError.

to_dict​

def to_dict(include_meta: bool = True) -> dict

Serialize a CausalGraph instance to a dictionary.

Arguments:

  • include_meta: Whether to include meta information about the graph in the dictionary. Default is True.

Returns:

The dictionary representation of the CausalGraph instance.

to_networkx​

def to_networkx() -> networkx.Graph

Return a networkx.Graph or networkx.DiGraph corresponding to the CausalGraph

instance.

Returns:

networkx.Graph representation of the CausalGraph. networkx.DiGraph is a subclass of networkx.Graph and will be returned when all edges in the CausalGraph instance are directed. When all edges in the CausalGraph instance are undirected, a networkx.Graph will be returned. When there is a mix of directed and undirected edges or there are more elaborate edge types, a cai_causal_graph.exceptions.CausalGraphErrors.GraphConversionError will be raised.

to_numpy​

def to_numpy() -> Tuple[numpy.ndarray, List[str]]

Return a numpy array that represents the adjacency matrix of the CausalGraph instance.

To avoid confusion, this method also returns a list of variables that corresponds to the order of columns in the numpy array.

to_gml_string​

def to_gml_string() -> str

Return a Graph Modelling Language (GML) string representative of the CausalGraph instance.

from_dict​

@classmethod
def from_dict(cls, d: dict, validate: bool = True) -> CausalGraph

Construct a CausalGraph instance from a Python dictionary.

The contents of the passed dictionary are deepcopied when constructing a new CausalGraph instance. This includes metadata of each node and edge, as well as the graph itself.

Arguments:

  • d: Dictionary to build a graph from.
  • validate: Whether to perform validation checks. The validation checks will raise if any cycles are introduced to the graph by adding the edge. This should only be disabled to speed up this method in situations where it is known that the serialized graph is valid. Default is True.

from_networkx​

@classmethod
def from_networkx(cls, g: networkx.Graph) -> CausalGraph

Construct a CausalGraph instance from a networkx.Graph instance.

from_skeleton​

@classmethod
def from_skeleton(cls, skeleton: Skeleton) -> CausalGraph

Construct a CausalGraph instance from a Skeleton instance.

from_gml_string​

@classmethod
def from_gml_string(cls, gml: str) -> CausalGraph

Return an instance of CausalGraph constructed from the provided Graph Modelling Language (GML) string.

from_adjacency_matrix​

@classmethod
def from_adjacency_matrix(
        cls,
        adjacency: numpy.ndarray,
        node_names: Optional[List[Union[NodeLike,
                                        int]]] = None) -> CausalGraph

Construct a CausalGraph instance from an adjacency matrix and optionally a list

of node names.

If a list of node names is provided it is used to identify the columns and rows in the adjacency matrix. If no node names are provided, these are autogenerated to be ['node_0','node_1'...].

Arguments:

  • adjacency: A square binary numpy adjacency array.
  • node_names: A list of strings, HasIdentifier, and/or integers which can be coerced to Node.

Returns:

A CausalGraph object.

copy​

def copy(include_meta: bool = True) -> CausalGraph

Return a copy of the CausalGraph instance.

Arguments:

  • include_meta: if True (default), the metadata will be copied as well.

Returns:

A copy of the CausalGraph instance.

__repr__​

def __repr__() -> str

Return a string description of the CausalGraph instance.

details​

def details() -> str

Return a detailed string description of the CausalGraph instance.

__hash__​

def __hash__() -> int

Return a hash representation of the CausalGraph instance.