webber.edges

  1"""
  2Helper class for edge and DAG validation logic.
  3"""
  4import typing as _T
  5import uuid as _uuid
  6import networkx as _nx
  7import enum as _enum
  8
  9__all__ = ["valid_node", "valid_nodes", "valid_dag", "validate_nodes", "label_node"]
 10
 11class Condition(_enum.IntEnum):
 12    """Represents edge condition for a node execution, based on outcome(s) of predecessor(s)."""
 13    Success = 0
 14    Failure = 1
 15    AnyCase = 3
 16
 17class dotdict(dict):
 18    """dot.notation access to dictionary attributes"""
 19    __getattr__ = dict.get
 20    __setattr__ = dict.__setitem__
 21    __delattr__ = dict.__delitem__
 22
 23class edgedict(dotdict):
 24    super(dotdict)
 25    def __init__(self, *E, **kwargs):
 26        super().__init__({'parent': E[0], 'child': E[1], 'id': E[:2]})
 27        self.update(kwargs)
 28
 29def continue_on_failure(edge: dict) -> bool:
 30    """Check edge condition for whether to continue on parent node's failure."""
 31    return edge['Condition'] in (Condition.Failure, Condition.AnyCase)
 32
 33def continue_on_success(edge: dict) -> bool:
 34    """Check edge condition for whether to continue on parent node's success."""
 35    return edge['Condition'] in (Condition.Success, Condition.AnyCase)
 36
 37def label_node(node: _T.Callable) -> str:
 38    """Generates unique identifiers for Python callables in a DAG."""
 39    return f"{node.__name__}__{_uuid.uuid4()}"
 40
 41def get_root(graph: _nx.DiGraph) -> list:
 42    """Given a network graph, return list of all nodes without incoming edges or dependencies."""
 43    return list(filter(
 44        lambda node: len(list(graph.predecessors(node))) < 1,
 45        graph.nodes.keys()
 46    ))
 47
 48# TODO: Refactor logic for DAG and node validation.
 49
 50def valid_node(node: _T.Union[str, _T.Callable]) -> bool:
 51    """Check whether given identifier represents a valid node (string or callable)."""
 52    return (isinstance(node,str) or callable(node))
 53
 54def valid_nodes(u_of_edge: _T.Union[str, _T.Callable], v_of_edge: _T.Union[str, _T.Callable]) -> bool:
 55    """Check whether parent and child nodes represent valid nodes (string or callable)."""
 56    return valid_node(u_of_edge) and valid_node(v_of_edge)
 57
 58def validate_nodes(u_of_edge: _T.Union[str, _T.Callable], v_of_edge: _T.Union[str, _T.Callable]) -> True:
 59    """
 60    Given parent and child identifiers, validate that both represent valid nodes.
 61    Otherwise raise exceptions.
 62    """
 63    if not valid_node(u_of_edge):
 64        err_msg = f"Outgoing node {u_of_edge} must be a string or a Python callable"
 65        raise TypeError(err_msg)
 66    
 67    if not valid_node(v_of_edge):
 68        err_msg = f"Incoming node {v_of_edge} must be a string or a Python callable"
 69        raise TypeError(err_msg)
 70    
 71    return True
 72
 73def valid_dag(graph: _nx.Graph) -> bool:
 74    """
 75    Given a network graph, return whether network is a valid DAG and that all node-keys are Python callables.
 76    Meant for internal use, DAG initialization.
 77    """
 78    return (
 79        isinstance(graph, _nx.Graph) and
 80        _nx.is_directed_acyclic_graph(graph) and
 81        not set(map(callable, list(graph.nodes.keys()))).issuperset({False})
 82    )
 83
 84
 85def validate_dag(graph: _nx.DiGraph) -> None:
 86    """
 87    Given a network graph, validate whether graph is a valid Webber DAG. Otherwise, raise exceptions.
 88    Meant for internal use, DAG initialization.
 89    """
 90    if not graph.is_directed():
 91        err_msg = f"Directed graph must be defined as type {_nx.DiGraph.__name__}"
 92        raise TypeError(err_msg)
 93
 94    if set(map(callable, list(graph.nodes.keys()))).issuperset({False}):
 95        err_msg = "All registered nodes must be callable Python functions."
 96        raise TypeError(err_msg)
 97
 98    if not _nx.is_directed_acyclic_graph(graph):
 99        err_msg = "Directed acyclic graph must be properly defined --" \
100                + "no cycles and one or more root nodes."
101        raise ValueError(err_msg)