analogvnn.graph.AcyclicDirectedGraph#

Module Contents#

Classes#

AcyclicDirectedGraph

The base class for all acyclic directed graphs.

class analogvnn.graph.AcyclicDirectedGraph.AcyclicDirectedGraph(graph_state: analogvnn.graph.ModelGraphState.ModelGraphState = None)[source]#

Bases: abc.ABC

The base class for all acyclic directed graphs.

Variables:
  • graph (nx.MultiDiGraph) – The graph.

  • graph_state (ModelGraphState) – The graph state.

  • _is_static (bool) – If True, the graph is not changing during runtime and will be cached.

  • _static_graphs (Dict[GRAPH_NODE_TYPE, List[Tuple[GRAPH_NODE_TYPE, List[GRAPH_NODE_TYPE]]]]) – The static graphs.

  • INPUT (GraphEnum) – GraphEnum.INPUT

  • OUTPUT (GraphEnum) – GraphEnum.OUTPUT

  • STOP (GraphEnum) – GraphEnum.STOP

graph: networkx.MultiDiGraph[source]#
graph_state: analogvnn.graph.ModelGraphState.ModelGraphState[source]#
_is_static: bool[source]#
_static_graphs: Dict[analogvnn.graph.GraphEnum.GRAPH_NODE_TYPE, List[Tuple[analogvnn.graph.GraphEnum.GRAPH_NODE_TYPE, List[analogvnn.graph.GraphEnum.GRAPH_NODE_TYPE]]]][source]#
INPUT[source]#
OUTPUT[source]#
STOP[source]#
save[source]#

Alias for render.

abstract __call__(*args, **kwargs)[source]#

Performs pass through the graph.

Parameters:
  • *args – Arguments

  • **kwargs – Keyword arguments

Raises:

NotImplementedError – since method is abstract

add_connection(*args: analogvnn.graph.GraphEnum.GRAPH_NODE_TYPE)[source]#

Add a connection between nodes.

Parameters:

*args – The nodes.

Returns:

self.

Return type:

AcyclicDirectedGraph

add_edge(u_of_edge: analogvnn.graph.GraphEnum.GRAPH_NODE_TYPE, v_of_edge: analogvnn.graph.GraphEnum.GRAPH_NODE_TYPE, in_arg: Union[None, int, bool] = None, in_kwarg: Union[None, str, bool] = None, out_arg: Union[None, int, bool] = None, out_kwarg: Union[None, str, bool] = None)[source]#

Add an edge to the graph.

Parameters:
  • u_of_edge (GRAPH_NODE_TYPE) – The source node.

  • v_of_edge (GRAPH_NODE_TYPE) – The target node.

  • in_arg (Union[None, int, bool]) – The input argument.

  • in_kwarg (Union[None, str, bool]) – The input keyword argument.

  • out_arg (Union[None, int, bool]) – The output argument.

  • out_kwarg (Union[None, str, bool]) – The output keyword argument.

Returns:

self.

Return type:

AcyclicDirectedGraph

static check_edge_parameters(in_arg: Union[None, int, bool], in_kwarg: Union[None, str, bool], out_arg: Union[None, int, bool], out_kwarg: Union[None, str, bool]) Dict[str, Union[None, int, str, bool]][source]#

Check the edge’s in and out parameters.

Parameters:
  • in_arg (Union[None, int, bool]) – The input argument.

  • in_kwarg (Union[None, str, bool]) – The input keyword argument.

  • out_arg (Union[None, int, bool]) – The output argument.

  • out_kwarg (Union[None, str, bool]) – The output keyword argument.

Returns:

Dict of valid edge’s in and out parameters.

Return type:

Dict[str, Union[None, int, str, bool]]

Raises:

ValueError – If in and out parameters are invalid.

static _create_edge_label(in_arg: Union[None, int, bool] = None, in_kwarg: Union[None, str, bool] = None, out_arg: Union[None, int, bool] = None, out_kwarg: Union[None, str, bool] = None, **kwargs) str[source]#

Create the edge’s label.

Parameters:
  • in_arg (Union[None, int, bool]) – The input argument.

  • in_kwarg (Union[None, str, bool]) – The input keyword argument.

  • out_arg (Union[None, int, bool]) – The output argument.

  • out_kwarg (Union[None, str, bool]) – The output keyword argument.

Returns:

The edge’s label.

Return type:

str

compile(is_static: bool = True)[source]#

Compile the graph.

Parameters:

is_static (bool) – If True, the graph will be compiled as a static graph.

Returns:

The compiled graph.

Return type:

AcyclicDirectedGraph

Raises:

ValueError – If the graph is not acyclic.

static _reindex_out_args(graph: networkx.MultiDiGraph) networkx.MultiDiGraph[source]#

Reindex the output arguments.

Parameters:

graph (nx.MultiDiGraph) – The graph.

Returns:

The graph with re-indexed output arguments.

Return type:

nx.MultiDiGraph

_create_static_sub_graph(from_node: analogvnn.graph.GraphEnum.GRAPH_NODE_TYPE) List[Tuple[analogvnn.graph.GraphEnum.GRAPH_NODE_TYPE, List[analogvnn.graph.GraphEnum.GRAPH_NODE_TYPE]]][source]#

Create a static sub graph connected to the given node.

Parameters:

from_node (GRAPH_NODE_TYPE) – The node.

Returns:

The static sub graph.

Return type:

List[Tuple[GRAPH_NODE_TYPE, List[GRAPH_NODE_TYPE]]]

parse_args_kwargs(input_output_graph: Dict[analogvnn.graph.GraphEnum.GRAPH_NODE_TYPE, analogvnn.graph.ArgsKwargs.InputOutput], module: analogvnn.graph.GraphEnum.GRAPH_NODE_TYPE, predecessors: List[analogvnn.graph.GraphEnum.GRAPH_NODE_TYPE]) analogvnn.graph.ArgsKwargs.ArgsKwargs[source]#

Parse the arguments and keyword arguments.

Parameters:
  • input_output_graph (Dict[GRAPH_NODE_TYPE, InputOutput]) – The input output graph.

  • module (GRAPH_NODE_TYPE) – The module.

  • predecessors (List[GRAPH_NODE_TYPE]) – The predecessors.

Returns:

The arguments and keyword arguments.

Return type:

ArgsKwargs

render(*args, real_label: bool = False, **kwargs) str[source]#

Save the source to file and render with the Graphviz engine.

Parameters:
  • *args – Arguments to pass to graphviz render function.

  • real_label – If True, the real label will be used instead of the label.

  • **kwargs – Keyword arguments to pass to graphviz render function.

Returns:

The (possibly relative) path of the rendered file.

Return type:

str