`elasticai.creator.graph`#

Package Contents#

Classes#

`BaseGraph`	All iterators in this class are guaranteed to have a fixed order. That means, the order of iteration is only allowed to change when the data Tucture is altered. If the content of two GraphDelegates is the same and they were built in the same way, then their iteration order is the same as well.
`Graph`
`GraphRewriter`	Rewrite graphs based on a pattern, interface, and replacement.
`NameRegistry`
`RewriteResult`

Functions#

`find_all_subgraphs`
`find_subgraph`
`bfs_iter_down`
`bfs_iter_up`
`dfs_iter`
`rewrite`	Return new rewritten graph and a mapping of replacements to new nodes.
`get_rewriteable_matches`	Yield all matches that do produce dangling edges and do not overlap with previous matches.
`produces_dangling_edge`	Check if there are dangling edges attached to non-interface nodes.

API#

class elasticai.creator.graph.BaseGraph[source]#

Bases: elasticai.creator.graph.graph.Graph[elasticai.creator.graph.base_graph.T]

All iterators in this class are guaranteed to have a fixed order. That means, the order of iteration is only allowed to change when the data Tucture is altered. If the content of two GraphDelegates is the same and they were built in the same way, then their iteration order is the same as well.

Caution

This class is not thread-safe.

Note

We are not providing methods for removal of nodes or edges on purpose. If you need to remove nodes or edges, you should create a new GraphDelegate. Manipulation of the graph should usually be done in a dedicated build phase.

Initialization

We keep successor and predecessor nodes just to allow for easier implementation. Currently, this implementation is not optimized for performance.

property successors: collections.abc.Mapping[elasticai.creator.graph.base_graph.T, set[elasticai.creator.graph.base_graph.T]]#

property predecessors: collections.abc.Mapping[elasticai.creator.graph.base_graph.T, set[elasticai.creator.graph.base_graph.T]]#

static from_dict(d: dict[elasticai.creator.graph.base_graph.T, collections.abc.Iterable[elasticai.creator.graph.base_graph.T]])[source]#

as_dict() → dict[elasticai.creator.graph.base_graph.T, set[elasticai.creator.graph.base_graph.T]][source]#

add_edge(_from: elasticai.creator.graph.base_graph.T, _to: elasticai.creator.graph.base_graph.T) → Self[source]#

add_node(node: elasticai.creator.graph.base_graph.T) → Self[source]#

property nodes: collections.abc.Set[elasticai.creator.graph.base_graph.T]#

iter_nodes() → collections.abc.Iterator[elasticai.creator.graph.base_graph.T][source]#: Iterator over nodes in a fixed but unspecified order.

iter_edges() → collections.abc.Iterator[tuple[elasticai.creator.graph.base_graph.T, elasticai.creator.graph.base_graph.T]][source]#: Iterator over edges in a fixed but unspecified order.

get_edges() → collections.abc.Iterator[tuple[elasticai.creator.graph.base_graph.T, elasticai.creator.graph.base_graph.T]][source]#: Iterator over edges in a fixed but unspecified order.

get_successors(node: elasticai.creator.graph.base_graph.T) → collections.abc.Iterator[elasticai.creator.graph.base_graph.T][source]#: Iterator over node successors in a fixed but unspecified order.

get_predecessors(node: elasticai.creator.graph.base_graph.T) → collections.abc.Iterator[elasticai.creator.graph.base_graph.T][source]#: Iterator over node predecessors in a fixed but unspecified order.

new() → Self[source]#

copy() → Self[source]#

class elasticai.creator.graph.Graph[source]#

Bases: typing.Protocol[elasticai.creator.graph.graph.T]

abstract property nodes: collections.abc.Set[elasticai.creator.graph.graph.T]#

abstract iter_edges() → collections.abc.Iterator[tuple[elasticai.creator.graph.graph.T, elasticai.creator.graph.graph.T]][source]#

abstract add_node(node: elasticai.creator.graph.graph.T) → Self[source]#

abstract add_edge(src: elasticai.creator.graph.graph.T, dst: elasticai.creator.graph.graph.T) → Self[source]#

abstract property predecessors: collections.abc.Mapping[elasticai.creator.graph.graph.T, set[elasticai.creator.graph.graph.T]]#

abstract property successors: collections.abc.Mapping[elasticai.creator.graph.graph.T, set[elasticai.creator.graph.graph.T]]#

abstract new() → Graph[T][source]#

abstract copy() → Graph[T][source]#

elasticai.creator.graph.find_all_subgraphs(pattern: elasticai.creator.graph.graph.Graph[elasticai.creator.graph.subgraph_matching.TP], graph: elasticai.creator.graph.graph.Graph[elasticai.creator.graph.subgraph_matching.T], node_constraint: elasticai.creator.graph._types.NodeConstraintFn[elasticai.creator.graph.subgraph_matching.TP, elasticai.creator.graph.subgraph_matching.T]) → list[dict[elasticai.creator.graph.subgraph_matching.TP, elasticai.creator.graph.subgraph_matching.T]][source]#

elasticai.creator.graph.find_subgraph(pattern: elasticai.creator.graph.graph.Graph[elasticai.creator.graph.subgraph_matching.TP], graph: elasticai.creator.graph.graph.Graph[elasticai.creator.graph.subgraph_matching.T], node_constraint: elasticai.creator.graph._types.NodeConstraintFn[elasticai.creator.graph.subgraph_matching.TP, elasticai.creator.graph.subgraph_matching.T]) → dict[elasticai.creator.graph.subgraph_matching.TP, elasticai.creator.graph.subgraph_matching.T][source]#

class elasticai.creator.graph.GraphRewriter(*, pattern: elasticai.creator.graph.graph.Graph[str], interface: elasticai.creator.graph.graph.Graph[str], replacement: elasticai.creator.graph.graph.Graph[str], match: elasticai.creator.graph.graph_rewriting.Matcher | elasticai.creator.graph.graph_rewriting._SeqMatcher, lhs: collections.abc.Mapping[str, str], rhs: collections.abc.Mapping[str, str])[source]#

Rewrite graphs based on a pattern, interface, and replacement.

The terminology here is based on the double pushout approach to graph rewriting. The algorithm will find the first occurence of pattern in the graph using the match function. Then, it will create a new graph by replacing the pattern with the replacement. The structures specified by interface are excluded from this replacement. Instead the interface is used as the “glue” between the original graph and the replacement. The functions lhs and rhs are used to identify the interface nodes in the pattern and replacement respectively.

Important

The rhs function is required to be injective. This means that each interface node should be mapped to a unique replacement node.
The algorithm will stop on the first match. If you need to replace multiple matches you have to call the rewrite function multiple times.
The nodes from replacement will be automatically renamed to avoid conflicts with the nodes in the original graph, i.e., if node a exists already we will add a new node a_1.

Note

In many cases where you want the matcher to compare graph and pattern attributes, you will have to pass a custom matcher and change it before running the rewrite on a new graph. E.g. to perform two subsequent rewrites you might have to:

matcher = MyMatcher()
matcher.set_graph(graph)
rewriter = GraphRewriter(
    pattern=pattern,
    interface=interface,
    replacement=replacement,
    match=matcher,
    lhs=lhs,
    rhs=rhs,
)
result = rewriter.rewrite(graph)
matcher.set_graph(result.new_graph)
result = rewriter.rewrite(result.new_graph)

Parameters:

pattern – The pattern to match in the graph.
interface – The interface where graph and replacement will be “glued” together.
replacement – The replacement for the pattern.
match – A function that takes a graph and a pattern and returns a dictionary of matches. See find_subgraphs for more information.
lhs – A dict that is used to map the interface to the pattern.
rhs – A dict that is used to map the interface to the replacement.

Initialization

rewrite(graph: elasticai.creator.graph.graph.Graph[str]) → elasticai.creator.graph.graph_rewriting.RewriteResult[source]#

Rewrite the graph based on the pattern, interface, and replacement.

Returns a RewriteResult object containing the new graph and two dicts mapping nodes from pattern to the original graph and nodes from replacement to the new graph. matches = self._match(graph, self._pattern)

elasticai.creator.graph.bfs_iter_down(successors: elasticai.creator.graph.graph_iterators.NodeNeighbourFn, predecessors: elasticai.creator.graph.graph_iterators.NodeNeighbourFn, start: elasticai.creator.graph.graph_iterators.HashableT) → collections.abc.Iterator[elasticai.creator.graph.graph_iterators.HashableT][source]#

elasticai.creator.graph.bfs_iter_up(predecessors: elasticai.creator.graph.graph_iterators.NodeNeighbourFn[elasticai.creator.graph.graph_iterators.HashableT], successors: elasticai.creator.graph.graph_iterators.NodeNeighbourFn[elasticai.creator.graph.graph_iterators.HashableT], start: elasticai.creator.graph.graph_iterators.HashableT) → collections.abc.Iterator[elasticai.creator.graph.graph_iterators.HashableT][source]#

elasticai.creator.graph.dfs_iter(successors: elasticai.creator.graph.graph_iterators.NodeNeighbourFn, start: elasticai.creator.graph.graph_iterators.HashableT) → collections.abc.Iterator[elasticai.creator.graph.graph_iterators.HashableT][source]#

elasticai.creator.graph.rewrite(*, replacement: elasticai.creator.graph.graph.Graph[str], original: elasticai.creator.graph.graph.Graph[str], match: collections.abc.Mapping[str, str], lhs: collections.abc.Mapping[str, str], rhs: collections.abc.Mapping[str, str]) → tuple[elasticai.creator.graph.graph.Graph[str], dict[str, str]][source]#

Return new rewritten graph and a mapping of replacements to new nodes.

The terminology here is based on the double pushout approach to graph rewriting. The algorithm will find the first occurence of pattern in the graph using the match function. Then, it will create a new graph by replacing the pattern with the replacement. The structures specified by interface are excluded from this replacement. Instead the interface is used as the “glue” between the original graph and the replacement. The functions lhs and rhs are used to identify the interface nodes in the pattern and replacement respectively.

Important

The rhs function is required to be injective. This means that each interface node should be mapped to a unique replacement node.
The algorithm will stop on the first match. If you need to replace multiple matches you have to call the rewrite function multiple times.
The nodes from replacement will be automatically renamed to avoid conflicts with the nodes in the original graph, i.e., if node a exists already we will add a new node a_1.

Note

The nodes from the replacement graph that aren’t part of the interface will be automatically renamed to avoid conflicts with nodes in the original graph.
Interface nodes serve as connection points between the original graph and replacement.

Parameters:

replacement – The graph that will replace the matched pattern in the original graph.
graph – The original graph where the pattern will be replaced.
match – A dictionary mapping nodes from pattern to nodes in the original graph.
lhs – Maps interface nodes to their corresponding nodes in the pattern (Left Hand Side).
rhs – Maps interface nodes to their corresponding nodes in the replacement (Right Hand Side).

Returns:

A tuple containing the new graph after the rewrite operation and a dictionary mapping replacement node names to new node names.

Raises:

ValueError – If the rhs function is not injective (when different interface nodes map to the same replacement node).
DanglingEdgeError – if there is an edge between an unmatched node and a matched non-interface node.

elasticai.creator.graph.get_rewriteable_matches(original: elasticai.creator.graph.graph.Graph[elasticai.creator.graph.graph_rewriting.T], matches: collections.abc.Iterable[dict[elasticai.creator.graph.graph_rewriting.TP, elasticai.creator.graph.graph_rewriting.T]], interface_nodes: collections.abc.Iterable[elasticai.creator.graph.graph_rewriting.TP]) → collections.abc.Iterator[dict[elasticai.creator.graph.graph_rewriting.TP, elasticai.creator.graph.graph_rewriting.T]][source]#

Yield all matches that do produce dangling edges and do not overlap with previous matches.

The matches returned by this function are considerd safe to be rewritten in a single rewriting step in any order, without having to run an additional matching step and without producing dangling edges.

Parameters:

original – The original graph.
matches – The matches to check.
interface_nodes – All nodes in the pattern that are belong to the interface. These nodes are considered to be preserved during rewriting. Thus, the edges connected to these nodes are not considered dangling.

class elasticai.creator.graph.NameRegistry[source]#

Initialization

prepopulate(names)[source]#

get_unique_name(name)[source]#

class elasticai.creator.graph.RewriteResult(*, new_graph: elasticai.creator.graph.graph.Graph[str], pattern_to_original: dict[str, str], replacement_to_new: dict[str, str])[source]#: Initialization

elasticai.creator.graph.produces_dangling_edge(graph: elasticai.creator.graph.graph.Graph[elasticai.creator.graph.graph_rewriting.T], match: dict[elasticai.creator.graph.graph_rewriting.TP, elasticai.creator.graph.graph_rewriting.T], interface_nodes: collections.abc.Iterable[elasticai.creator.graph.graph_rewriting.TP]) → bool[source]#: Check if there are dangling edges attached to non-interface nodes.

exception elasticai.creator.graph.DanglingEdgeError(a, b)[source]#: Bases: Exception

elasticai.creator.graph#

Package Contents#

Classes#

Functions#

API#

`elasticai.creator.graph`#