Skip to main content

graph_components

Node​

from cai_causal_graph.graph_components import Node

class Node(HasIdentifier, HasMetadata, CanDictSerialize)

A utility class that manages the state of a node.

Methods​

__init__​

def __init__(identifier: str,
             meta: Optional[Dict[str, Any]] = None,
             variable_type: NodeVariableType = NodeVariableType.UNSPECIFIED)

Arguments:

  • identifier: String that uniquely identifies the node within the causal graph.
  • meta: The metadata of the node. Default is None. If passed, meta is shallow-copied.
  • variable_type: The variable type that the node represents. The choices are available through the NodeVariableType enum. Default is NodeVariableType.UNSPECIFIED.

__hash__​

def __hash__() -> int

Return a hash value of the node identifier.

__eq__​

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

Check if a node is equal to another node.

When deep is False (default), this method checks equality between the node identifiers. When deep is True, variable type and metadata is also checked. Inbound and outbound edges are never checked.

Arguments:

  • other: The other node to compare to.
  • deep: If True, then the variable type and metadata are also checked, in addition to the identifier. Default is False.

__ne__​

def __ne__(other: object) -> bool

Check if the node is not equal to another node.

invalidate​

def invalidate()

Set this node to be invalid after deleting it.

node_name​

@property
def node_name() -> str

Alias for identifier.

identifier​

@property
def identifier() -> str

Return the node identifier.

get_identifier​

def get_identifier() -> str

Return the node identifier.

variable_type​

@property
def variable_type() -> NodeVariableType

Return the variable type of the node.

variable_type​

@variable_type.setter
def variable_type(new_type: Union[NodeVariableType, str])

Set the variable type of the node.

Arguments:

  • new_type: New variable type.

metadata​

@property
def metadata() -> dict

Return the node metadata.

get_metadata​

def get_metadata() -> dict

Return the node metadata.

get_inbound_edges​

def get_inbound_edges() -> List[Edge]

Get all inbound (directed) edges to the node.

get_outbound_edges​

def get_outbound_edges() -> List[Edge]

Get all outbound (directed) edges to the node.

count_inbound_edges​

def count_inbound_edges() -> int

Count the number of inbound (directed) edges to the node.

count_outbound_edges​

def count_outbound_edges() -> int

Count the number of outbound (directed) edges to the node.

is_source_node​

def is_source_node() -> bool

Return whether the node is a source node (no incoming edges).

is_sink_node​

def is_sink_node() -> bool

Return whether the node is a sink node (no outgoing edges).

identifier_from​

@staticmethod
def identifier_from(node_like: NodeLike) -> str

Return the node identifier from a node-like object instance.

__repr__​

def __repr__() -> str

Return a string description of the object.

details​

def details() -> str

Return a detailed string description of the object.

to_dict​

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

Serialize a Node instance to a dictionary.

Returned dictionary contains a shallow-copy of the metadata of this node (if include_meta is True).

Arguments:

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

Returns:

The dictionary representation of the Node instance.

from_dict​

@classmethod
def from_dict(cls, node_dict: dict) -> Node

Return a Node instance from a dictionary.

TimeSeriesNode​

from cai_causal_graph.graph_components import TimeSeriesNode

class TimeSeriesNode(Node)

Time series node.

A node in a time series causal graph will have additional metadata and attributes that provides the time information of the node together with the variable name.

The two additional metadata are:

  • cai_causal_graph.type_definitions.TIME_LAG: the time difference with respect to the reference time 0
  • cai_causal_graph.type_definitions.VARIABLE_NAME: the name of the variable (without the lag information)

Methods​

__init__​

def __init__(identifier: Optional[NodeLike] = None,
             time_lag: Optional[int] = None,
             variable_name: Optional[str] = None,
             meta: Optional[Dict[str, Any]] = None,
             variable_type: NodeVariableType = NodeVariableType.UNSPECIFIED)

Initialize the time series node.

Arguments:

  • identifier: String that uniquely identifies the node within the causal graph. If the identifier is provided, the time_lag and variable_name will be extracted from the identifier. Default is None.
  • time_lag: The time lag of the node. If time_lag is provided, then variable_name must be provided to set the identifier. If both time_lag and variable_name are provided, the identifier must be None. Default is None.
  • variable_name: The variable name of the node. If variable_name is provided, then time_lag must be provided to set the identifier. If both time_lag and variable_name are provided, the identifier must be None. Default is None.
  • meta: The metadata of the node. Default is None. If passed, meta is shallow-copied.
  • variable_type: The variable type that the node represents. The choices are available through the NodeVariableType enum. Default is NodeVariableType.UNSPECIFIED.

get_metadata_schema​

@classmethod
def get_metadata_schema(cls) -> List[MetaField]

Return the metadata schema of TimeSeriesNode.

See cai_causal_graph.interfaces.HasMetadata.get_metadata_schema for more information on the how the metadata schema is used.

time_lag​

@property
def time_lag() -> int

Return the time lag of the node from the metadata.

variable_name​

@property
def variable_name() -> str

Return the variable name of the node from the metadata.

__eq__​

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

Check if a node is equal to another node.

When deep is False (default), this method checks equality between the node identifiers, variable names, and time lags. When deep is True, variable type and metadata is also checked. Inbound and outbound edges are never checked.

Arguments:

  • other: The other node to compare with.
  • deep: If True, then the variable type and metadata are also checked, in addition to the identifier, variable name, and time lag. Default is False.

__hash__​

def __hash__() -> int

Return a hash value of the node identifier.

to_dict​

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

Serialize a TimeSeriesNode instance to a dictionary.

Arguments:

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

Returns:

The dictionary representation of the TimeSeriesNode instance.

Edge​

from cai_causal_graph.graph_components import Edge

class Edge(HasIdentifier, HasMetadata, CanDictSerialize)

A utility class that manages the state of an edge.

Methods​

__init__​

def __init__(source: Node,
             destination: Node,
             edge_type: EdgeType = EdgeType.DIRECTED_EDGE,
             meta: Optional[Dict[str, Any]] = None)

Arguments:

  • source: The Node from which the edge will originate.
  • destination: The Node at which the edge will terminate.
  • 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. Default is None. If passed, meta is shallow-copied.

__hash__​

def __hash__() -> int

Return a hash value of the edge identifier.

__eq__​

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

Check if the edge is equal to another edge.

When deep is False (default), this method checks equality between the source node identifiers, destination node identifiers, and edge types. When deep is True, the metadata is also checked as well as a deep equality check on the nodes.

For some edge types, particularly cai_causal_graph.type_definitions.EdgeType.UNDIRECTED_EDGE, cai_causal_graph.type_definitions.EdgeType.BIDIRECTED_EDGE, and cai_causal_graph.type_definitions.EdgeType.UNKNOWN_EDGE, there is inherently no direction. Therefore, edges of those types with the source and destination nodes switched are considered equal.

Arguments:

  • other: The other edge to compare with.
  • deep: If True, then a deep equality check is done on the nodes and the metadata is also checked, in addition to the edge types. Default is False.

__ne__​

def __ne__(other: object) -> bool

Check if the edge is not equal to another edge.

invalidate​

def invalidate()

Set this edge to be invalid after deleting it.

source​

@property
def source() -> Node

Return the source node.

destination​

@property
def destination() -> Node

Return the destination node.

identifier​

@property
def identifier() -> str

Return the edge identifier.

get_identifier​

def get_identifier() -> str

Return the edge identifier.

descriptor​

@property
def descriptor() -> str

Return the edge descriptor.

metadata​

@property
def metadata() -> dict

Return the edge metadata.

get_metadata​

def get_metadata() -> dict

Return the edge metadata.

edge_type​

@property
def edge_type() -> EdgeType

Return the edge type.

get_edge_type​

def get_edge_type() -> EdgeType

Return the edge type.

Please note that to change the edge type, you must use the change_edge_type method defined on the causal graph.

get_edge_pair​

def get_edge_pair() -> Tuple[str, str]

Return a tuple of the source node and destination node identifiers.

__repr__​

def __repr__() -> str

Return a string description of the object.

details​

def details() -> str

Return a detailed string description of the object.

from_dict​

@classmethod
def from_dict(cls, edge_dict: dict) -> Edge

Deserialize a dictionary to a Edge instance.

Arguments:

  • edge_dict: The dictionary representation of the Edge instance. If the node class is not specified, it defaults to Node.

Returns:

The Edge instance.

to_dict​

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

Serialize a Edge instance to a dictionary.

Returned dictionary contains a shallow-copy of the metadata of this edge (if include_meta is True).

Arguments:

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

Returns:

The dictionary representation of the Edge instance.

TimeSeriesEdge​

from cai_causal_graph.graph_components import TimeSeriesEdge

class TimeSeriesEdge(Edge)

Class defining a time-series edge.

Time-series edge is equivalent to a Edge class, except that it only ensures that the destination of an edge is at the same or later time than its source.

This means that it is not possible to construct a directed time-series edge which does not respect time. Moreover, if an undirected edge is constructed by passing a source which is at a later time than the destination, the source and destination are swapped.

Attributes​

  • source: TimeSeriesNode
  • destination: TimeSeriesNode

Methods​

__init__​

def __init__(source: Node,
             destination: Node,
             edge_type: EdgeType = EdgeType.DIRECTED_EDGE,
             meta: Optional[Dict[str, Any]] = None)

Construct a TimeSeriesEdge.

Arguments:

  • source: The Node from which the edge will originate.
  • destination: The Node at which the edge will terminate.
  • 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. Default is None. If passed, meta is shallow-copied.