`elasticai.creator.graph.graph_rewriting`#

Module Contents#

Classes#

`RewriteResult`
`Matcher`
`GraphRewriter`	Rewrite graphs based on a pattern, interface, and replacement.
`SingleNewNameGenerator`

Functions#

`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.
`rewrite`	Return new rewritten graph and a mapping of replacements to new nodes.

Data#

`TG`
`TR`
`TI`
`T`
`TP`

API#

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

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

class elasticai.creator.graph.graph_rewriting.Matcher[source]#

Bases: typing.Protocol

__call__(*, pattern: elasticai.creator.graph.graph.Graph, graph: elasticai.creator.graph.graph.Graph) → dict[str, str][source]#

elasticai.creator.graph.graph_rewriting.TG#: ‘TypeVar(…)’

elasticai.creator.graph.graph_rewriting.TR#: ‘TypeVar(…)’

elasticai.creator.graph.graph_rewriting.TI#: ‘TypeVar(…)’

elasticai.creator.graph.graph_rewriting.T#: ‘TypeVar(…)’

elasticai.creator.graph.graph_rewriting.TP#: ‘TypeVar(…)’

elasticai.creator.graph.graph_rewriting.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.

elasticai.creator.graph.graph_rewriting.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.

elasticai.creator.graph.graph_rewriting.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.

class elasticai.creator.graph.graph_rewriting.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)

class elasticai.creator.graph.graph_rewriting.SingleNewNameGenerator(registry: elasticai.creator.graph.name_generation.NameRegistry)[source]#

Initialization

get_name(name: str) → str[source]#

elasticai.creator.graph.graph_rewriting#

Module Contents#

Classes#

Functions#

Data#

API#

`elasticai.creator.graph.graph_rewriting`#