Read about our upcoming Code of Conduct on this issue

Commit 9f1592d3 authored by Pierre Augier's avatar Pierre Augier
Browse files

Analyses for blocks_if

parent 8c9b607f27ae
from pathlib import Path
# import ast
import gast as ast
import beniget
import astunparse
from pprint import pprint
from capturex import CaptureX, make_code_external
from transonic.analyses.capturex import CaptureX, make_code_external
from transonic.analyses.util import get_annotations, print_ast, print_dump, filter_code_typevars
from transonic.analyses import compute_ancestors_chains, get_boosted_dicts
from transonic.analyses.util import (
print_dumped,
print_unparsed,
get_annotations,
filter_code_typevars,
)
from transonic.analyses.blocks_if import get_block_definitions
path_examples = Path("examples")
files = sorted(path_examples.glob("*.py"))
with open(files[1]) as file:
with open(files[4]) as file:
code = file.read()
print("ast.parse")
module = ast.parse(code)
print("compute DefUseChains")
duc = beniget.DefUseChains()
duc.visit(module)
print("compute Ancestors")
ancestors = beniget.Ancestors()
ancestors.visit(module)
print("compute UseDefChains")
udc = beniget.UseDefChains(duc)
functions_boosted = {}
classes_boosted = {}
methods_boosted = {}
# we first need to find the node where transonic.boost is defined...
def add_definition(definition_node):
tmp_boosted = None
key = definition_node.name
if isinstance(definition_node, ast.FunctionDef):
parent = ancestors.parent(definition_node)
if isinstance(parent, ast.ClassDef):
tmp_boosted = methods_boosted
key = (parent.name, key)
else:
tmp_boosted = functions_boosted
elif isinstance(definition_node, ast.ClassDef):
tmp_boosted = classes_boosted
if tmp_boosted is not None:
tmp_boosted[key] = definition_node
for node in module.body:
if isinstance(node, ast.Import):
for alias in node.names:
if alias.name == "transonic":
transonic_def_node = alias
transonic_def = duc.chains[transonic_def_node]
for user in transonic_def.users():
for user1 in user.users():
if (
isinstance(user1.node, ast.Attribute)
and user1.node.attr == "boost"
):
definition_node = ancestors.parent(user1.node)
add_definition(definition_node)
elif isinstance(node, ast.ImportFrom):
if node.module == "transonic":
for alias in node.names:
if alias.name == "boost":
boost_def_node = alias
boost_def = duc.chains[boost_def_node]
for user in boost_def.users():
definition_node = ancestors.parent(user.node)
add_definition(definition_node)
print("functions:")
pprint(functions_boosted)
print("methods:")
pprint(methods_boosted)
print("classes:")
pprint(classes_boosted)
boosteds = {"functions": functions_boosted, "methods": methods_boosted, "classes": classes_boosted}
objects = []
for boosted in boosteds.values():
for definition in boosted.values():
objects.append(definition)
print("compute ancestors and chains")
ancestors, duc, udc = compute_ancestors_chains(module)
print("filter_code_typevars")
code_dependance_annotations = filter_code_typevars(module, duc, ancestors)
print(code_dependance_annotations)
print("find boosted objects")
boosted_dicts = get_boosted_dicts(module, ancestors, duc)
pprint(boosted_dicts)
print("compute code dependance")
def_nodes = [
def_node
for boosted_dict in boosted_dicts.values()
for def_node in boosted_dict.values()
]
# remove the decorator (boost) to compute the code dependance
for def_node in def_nodes:
def_node.decorator_list = []
capturex = CaptureX(
objects,
def_nodes,
module,
ancestors=ancestors,
defuse_chains=duc,
usedef_chains=udc,
ancestors=ancestors,
consider_annotations=False,
)
code_dependance = make_code_external(capturex.external)
print(code_dependance)
print("compute the annotations")
namespace = {}
exec(code_dependance_annotations, namespace)
annotations = {}
for kind, boosted in boosteds.items():
for kind, boosted_dict in boosted_dicts.items():
annotations[kind] = {}
for key, definition in boosted.items():
for key, definition in boosted_dict.items():
annotations[kind][key] = get_annotations(definition, namespace)
pprint(annotations)
print("get_block_definitions")
blocks = get_block_definitions(code, module, ancestors, duc, udc)
pprint(blocks)
\ No newline at end of file
import gast as ast
import beniget
def compute_ancestors_chains(module_node):
ancestors = beniget.Ancestors()
ancestors.visit(module_node)
duc = beniget.DefUseChains()
duc.visit(module_node)
udc = beniget.UseDefChains(duc)
return ancestors, duc, udc
def get_boosted_dicts(module, ancestors, duc):
kinds = ("functions", "methods", "classes")
boosted_dicts = {kind: {} for kind in kinds}
def add_definition(definition_node):
boosted_dict = None
key = definition_node.name
if isinstance(definition_node, ast.FunctionDef):
parent = ancestors.parent(definition_node)
if isinstance(parent, ast.ClassDef):
boosted_dict = boosted_dicts["methods"]
key = (parent.name, key)
else:
boosted_dict = boosted_dicts["functions"]
elif isinstance(definition_node, ast.ClassDef):
boosted_dict = boosted_dicts["classes"]
if boosted_dict is not None:
boosted_dict[key] = definition_node
# we first need to find the node where transonic.boost is defined...
for node in module.body:
if isinstance(node, ast.Import):
for alias in node.names:
if alias.name == "transonic":
transonic_def_node = alias
transonic_def = duc.chains[transonic_def_node]
for user in transonic_def.users():
for user1 in user.users():
if (
isinstance(user1.node, ast.Attribute)
and user1.node.attr == "boost"
):
definition_node = ancestors.parent(user1.node)
add_definition(definition_node)
elif isinstance(node, ast.ImportFrom):
if node.module == "transonic":
for alias in node.names:
if alias.name == "boost":
boost_def_node = alias
boost_def = duc.chains[boost_def_node]
for user in boost_def.users():
definition_node = ancestors.parent(user.node)
add_definition(definition_node)
return boosted_dicts
\ No newline at end of file
import gast as ast
from transonic.analyses.util import gather_rawcode_comments
class BlockDefinition:
def __init__(self, **kwargs):
self.kwargs = kwargs
self.__dict__.update(**kwargs)
def __repr__(self):
return repr(self.kwargs)
def get_block_definitions(code, module, ancestors, duc, udc):
blocks = []
node_transonic_obj = None
for node in module.body:
if isinstance(node, ast.ImportFrom) and node.module == "transonic":
for alias in node.names:
if alias.name == "Transonic":
Transonic_node = alias
node_transonic_obj = None
for user in duc.chains[Transonic_node].users():
parent = ancestors.parent(user.node)
if isinstance(parent, ast.Call):
grandparent = ancestors.parent(parent)
if len(grandparent.targets) != 1:
continue
node_transonic_obj = grandparent.targets[0]
if node_transonic_obj is not None:
break
if node_transonic_obj is None:
return blocks
def_ = duc.chains[node_transonic_obj]
nodes_using_ts = [user.node for user in def_.users()]
for user in def_.users():
for user1 in user.users():
if isinstance(user1.node, ast.Attribute):
attribute = user1.node
if attribute.attr == "is_transpiled":
parent = ancestors.parent(attribute)
if isinstance(parent, ast.If):
# it could be the begining of a block
if_node = parent
if len(parent.body) != 1:
# no it's not a block definition
continue
node = parent.body[0]
call = node.value
if isinstance(node, ast.Expr):
results = []
elif isinstance(node, ast.Assign):
results = [target.id for target in node.targets]
else:
# no it's not a block definition
continue
attribute = call.func
ts_node = attribute.value
if (
ts_node not in nodes_using_ts
or attribute.attr != "use_block"
):
# no it's not a block definition
continue
name_block = call.args[0].s
rawcode, comments = gather_rawcode_comments(if_node, code)
# if we are here, it's a block definition
blocks.append(
BlockDefinition(
name=name_block,
results=results,
ast_code=if_node.orelse,
comments=comments,
rawcode=rawcode,
)
)
return blocks
......@@ -8,9 +8,9 @@ class CaptureX(ast.NodeVisitor):
self,
functions,
module_node,
ancestors=None,
defuse_chains=None,
usedef_chains=None,
ancestors=None,
consider_annotations=True,
):
......
from textwrap import dedent
import gast as ast
import astunparse
def print_ast(code):
module = ast.parse(code)
print(astunparse.dump(module))
return module
def print_dump(node):
def print_dumped(source):
if isinstance(source, str):
module = ast.parse(source)
if len(module.body) == 1:
node = module.body[0]
else:
node = module
else:
node = source
print(astunparse.dump(node))
......@@ -101,4 +105,55 @@ def filter_code_typevars(module, duc, ancestors):
elif isinstance(node, (ast.Assign, ast.AugAssign)):
kept.append(node)
return astunparse.unparse(module_filtered)
\ No newline at end of file
return astunparse.unparse(module_filtered)
class AnalyseLines(ast.NodeVisitor):
def __init__(self, main_node):
if isinstance(main_node, ast.Module):
self.line_start = 1
else:
self.line_start = main_node.lineno
self.line_last = self.line_start
self.visit(main_node)
def generic_visit(self, node):
if hasattr(node, "lineno"):
if node.lineno > self.line_last:
self.line_last = node.lineno
super().generic_visit(node)
def get_lines(self):
return self.line_start, self.line_last
def get_code(self, source):
lines = source.split("\n")
stop = self.line_last
nb_lines = len(lines)
if nb_lines != stop:
# find next not empty and not comment line
ind = stop - 1
next_line = ""
while ind + 1 < nb_lines and next_line == "":
ind += 1
next_line = lines[ind].strip()
if next_line.startswith("#"):
next_line = ""
if any(
next_line.startswith(character) for character in (")", "]", "}")
):
stop = ind + 1
return "\n".join(lines[self.line_start - 1 : stop])
def gather_rawcode_comments(node, code_module):
analysis = AnalyseLines(node)
rawcode = dedent(analysis.get_code(code_module))
comments = dedent("\n".join(
line for line in rawcode.split("\n") if line.strip().startswith("#")
))
return rawcode, comments
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment