import pickle
import logging as log
from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable, Dict, List, Optional
from zoti_graph.appgraph import AppGraph
from zoti_graph.core import Port
from zoti_graph.io import dump_node_info, draw_graphviz, draw_tree
from zoti_graph.exceptions import ScriptError
[docs]@dataclass(eq=False, repr=False)
class TransSpec:
"""Transformation specification wrapper. Extends the transformatiion
with some control flags and arguments, e.g., useful when
debugging.
"""
func: Callable
"""the transformation function"""
clean: List[str] = field(default_factory=list)
"""list of names of previous transformations, whose byproducts should
be *completely removed* from the handler's state.
"""
dump_tree: Optional[Dict] = None
"""keyword-arguments sent to `zoti_graph.io.draw_tree()
<../zoti-graph/api-reference>`_. If left ``None``, the tree
structure will not be dumped.
"""
dump_graphviz: Optional[Dict] = None
"""keyword-arguments sent to `zoti_graph.io.draw_graph()
<../zoti-graph/api-reference>`_. If left ``None``, the graph
structure will not be dumped.
"""
dump_nodes: bool = False
"""dump node info as text after transformation for debugging"""
dump_prefix: Optional[str] = None
"""overrides the :class:`Script` member with the same name."""
dump_title: Optional[str] = None
"""optional title for the dumped file. If left ``None`` it will be
replaced by the function name.
"""
[docs]class Script:
"""Transformation script handler. It storgit ses an application graph and
(possibly) a data type handler and contains utilities for
executing rules.
:param G: a fully-constructed application graph
:param T: (optional) a data type handler
:param dump_prefix: path where intermediate results will be written to
Apart from altering the application graph as side effects,
transformation rules are able to return byproducts. These
byproducts are gradually stored and are accessible as class
members baring the name of the applied rule. E.g., after applying
a transformation ``TransSpec(foo)`` where::
def foo(G, **kwargs):
# do something on G
return 'bar'
the current script will have a new member ``foo`` containing the
string ``'bar'``. Existing members with the same name are
overriden.
"""
def __init__(self, G: AppGraph, T=None, dump_prefix="."):
self._dump_prefix = dump_prefix
self.G = G
self.T = T
[docs] @classmethod
def from_pickle(cls, path):
"""Loads a binary object containing a previously pickled script handler."""
with open(path, "rb") as f:
loaded = pickle.load(f)
assert hasattr(loaded, "G")
assert hasattr(loaded, "T")
return loaded
[docs] def pickle(self, path):
"""Dumps the current state of the handler into a binary object."""
with open(path, "wb") as f:
pickle.dump(self, f)
[docs] def sanity(self, rules: List[Callable]):
"""Utility for checking a batch of sanity rules on different elements
of the stored graph (see
`zoti_graph.appgraph.AppGraph.sanity()
<../zoti-graph/api-reference>`_) based on their name
formation:
* ``port_[name]`` are applied only on ports;
* ``edge_[name]`` are applied only on edges;
* ``node_[name]`` are applied only on regular nodes;
* in all other cases it applies the rule on the entire graph
(i.e., the root node).
"""
log.info(f"*** Verifying sanity rules for graph {self.G.root}***")
port_rules = [r for r in rules if r.__name__.startswith("port")]
node_rules = [r for r in rules if r.__name__.startswith("node")]
edge_rules = [r for r in rules if r.__name__.startswith("edge")]
graph_rules = [r for r in rules if r not in
port_rules + node_rules + edge_rules]
def _check(collection, *element):
for rule in collection:
self.G.sanity(rule, *element)
for edge in self.G.only_graph().edges:
_check(edge_rules, *edge)
log.info(f" - passed {[f.__name__ for f in edge_rules]}")
for node in self.G.ir.nodes:
if isinstance(self.G.entry(node), Port):
_check(port_rules, node,)
else:
_check(node_rules, node)
log.info(
f" - passed {[f.__name__ for f in port_rules + node_rules]}")
_check(graph_rules, self.G.root)
log.info(f" - passed {[f.__name__ for f in graph_rules]}")