Commit c6404867 authored by Christophe Benz's avatar Christophe Benz

Update transpiler

parent 25e1b760
# -*- coding: utf-8 -*-
import json
import math
import os
import pkg_resources
from m_language_parser import dependencies_helpers
from toolz.curried import first
from . import core
from .generated.variables_definitions import definition_by_variable_name
"""Functions used by formulas"""
......@@ -46,19 +46,19 @@ somme = sum
result_by_formula_name_cache = {}
def cache_result(formula_function):
def cached(formula_function):
formula_name = formula_function.__name__
def new_formula_function():
def cached_formula():
if formula_name not in result_by_formula_name_cache:
result = formula_function()
# if result == 10561.16:
# print(formula_name + ' == 10561.16')
# import ipdb; ipdb.set_trace()
result_by_formula_name_cache[formula_name] = result
return result_by_formula_name_cache[formula_name]
return new_formula_function
return cached_formula
# Dependencies helpres
def inspect():
......@@ -92,35 +92,41 @@ def inspect_cache(predicate=None):
]
def load_dependencies_by_formula_name():
m_language_parser_dir_path = pkg_resources.get_distribution('m_language_parser').location
variables_dependencies_file_path = os.path.join(m_language_parser_dir_path, 'json', 'data',
'formulas_dependencies.json')
with open(variables_dependencies_file_path) as variables_dependencies_file:
variables_dependencies_str = variables_dependencies_file.read()
dependencies_by_formula_name = json.loads(variables_dependencies_str)
return dependencies_by_formula_name
def has_dependency(formula_name, dependency_formula_names, dependencies_by_formula_name):
def walk_dependencies(formula_name, dependencies):
if formula_name in dependencies:
def walk_dependencies(formula_name, visited_dependencies):
if formula_name in visited_dependencies:
return False
dependencies.add(formula_name)
visited_dependencies.add(formula_name)
if formula_name in dependency_formula_names:
return True
formula_dependencies = dependencies_by_formula_name.get(formula_name)
if formula_dependencies is not None:
for dependency_name in formula_dependencies:
if walk_dependencies(
dependencies=dependencies,
visited_dependencies=visited_dependencies,
formula_name=dependency_name,
):
return True
return False
return walk_dependencies(
dependencies=set(),
formula_name=formula_name,
visited_dependencies=set(),
)
def find_restituee_variables_depending_on(formula_names):
dependencies_by_formula_name = dependencies_helpers.load_dependencies_by_formula_name()
# formulas_dependencies_by_formula_name = dependencies_helpers.filter_formulas_dependencies(
# definition_by_variable_name=definition_by_variable_name,
# dependencies_by_formula_name=dependencies_by_formula_name,
# )
dependencies_by_formula_name = load_dependencies_by_formula_name()
restituee_variable_names = core.find_restituee_variables()
return list(filter(
lambda restituee_variable_name: has_dependency(
......
......@@ -15,7 +15,7 @@ import pprint
import textwrap
from m_language_parser.unloop_helpers import iter_unlooped_nodes
from toolz import concatv, interpose, mapcat
from toolz import concatv, interpose
from . import core
......@@ -138,18 +138,23 @@ def visit_formula(node):
formula_name = node['name']
sanitized_formula_name = core.sanitized_variable_name(formula_name)
expression_source = visit_node(node['expression'])
docstring_lines = []
description = core.get_variable_description(formula_name)
if description is not None:
docstring_lines.append(description)
if 'pour_formula' in node:
docstring_lines.append('From {}'.format(node['pour_formula']['name']))
source = """\
@cache_result
def {}():
@cached
def {formula_name}():
\"\"\"
{}{}
{docstring}
\"\"\"
return {}
return {expression}
""".format(
sanitized_formula_name,
core.get_variable_description(formula_name),
'\n From {}'.format(node['pour_formula']['name']) if 'pour_formula' in node else '',
expression_source,
docstring=textwrap.indent('\n'.join(docstring_lines), prefix=4 * ' '),
expression=expression_source,
formula_name=sanitized_formula_name,
)
return (sanitized_formula_name, source)
......@@ -199,8 +204,11 @@ def visit_product_expression(node):
def visit_regle(node):
return mapcat(
lambda node1: visit_node(node1) if node1['type'] == 'pour_formula' else [visit_node(node1)],
return map(
lambda node1: {
'applications': node['applications'],
'sources': visit_node(node1) if node1['type'] == 'pour_formula' else [visit_node(node1)],
},
node['formulas'],
)
......@@ -215,7 +223,7 @@ def visit_symbol(node):
if core.is_saisie_variable(symbol) \
else (
'base_variables[{!r}]'.format(symbol)
if core.is_base_variable(symbol)
if core.is_base_variable(symbol) and not core.is_restituee_variable(symbol)
else (
'{}()'.format(core.sanitized_variable_name(symbol))
if core.is_calculee_variable(symbol)
......
......@@ -7,13 +7,12 @@ import logging
import os
import sys
from toolz.curried import drop, map, pipe, take_nth
from toolz.curried.operator import attrgetter
from toolz.curried import map
from calculateur_impots import core
from calculateur_impots.generated import formulas
from calculateur_impots.generated.variables_definitions import definition_by_variable_name
# from calculateur_impots.generated.verif_regles import verif_regles
from calculateur_impots.generated.verif_regles import verif_regles
# Globals
......@@ -101,7 +100,11 @@ def main():
saisie_variables['V_0DA'] = 1980
# src/tgvH.m:14317:V_0AC : saisie famille classe = 0 priorite = 20 categorie_TL = 10 nat_code = 0 alias 0AC : "Case a cocher : situation de famille Celibataire" type BOOLEEN ; # noqa
saisie_variables['V_0AC'] = 1
# saisie_variables['V_0AC'] = 1
# Avoid division by 0
saisie_variables['CARPENBAV'] = 1
saisie_variables['CARPENBAC'] = 1
if args.saisie_variables is not None:
saisie_variables.update(iter_saisie_variables(args.saisie_variables))
......@@ -119,8 +122,7 @@ def main():
# saisie_variables=saisie_variables,
# )
# FIXME Rename "compute"
formulas_functions = formulas.compute(
formulas_functions = formulas.get_formulas(
base_variables=base_variables,
saisie_variables=saisie_variables,
)
......
......@@ -19,10 +19,9 @@ import pprint
import sys
import textwrap
from m_language_parser import dependencies_helpers
from toolz.curried import concat, concatv, filter, map, mapcat, pipe, sorted, valmap
from toolz.curried import concat, concatv, filter, first, map, mapcat, pipe, sorted, valmap
from calculateur_impots import core, python_source_visitors
from calculateur_impots import core, formulas_helpers, python_source_visitors
# Globals
......@@ -40,20 +39,14 @@ generated_dir_path = os.path.abspath(os.path.join(script_dir_path, '..', 'genera
def formulas_file_source(formulas_sources):
return as_lines(concatv(
(
'from ..formulas_helpers import *',
'from .constants import *\n',
'def compute(base_variables, saisie_variables):',
),
map(
lambda formula_source: textwrap.indent(formula_source, prefix=4 * ' '),
formulas_sources,
),
(
' return locals()',
),
))
return """\
from ..formulas_helpers import *
from .constants import *
def get_formulas(base_variables, saisie_variables):
{}
return locals()
""".format(textwrap.indent('\n'.join(formulas_sources), prefix=4 * ' '))
def as_lines(sequence):
......@@ -104,20 +97,11 @@ def iter_ast_json_file_names(filenames, excluded_filenames=None):
def load_regles_file(json_file_name):
regles_nodes = read_ast_json_file(json_file_name)
regles_nodes = filter(
lambda node: 'batch' in node['applications'] or 'iliad' in node['applications'],
regles_nodes,
)
formula_name_and_source_pairs = mapcat(python_source_visitors.visit_node, regles_nodes)
return formula_name_and_source_pairs
return mapcat(python_source_visitors.visit_node, regles_nodes)
def load_verifs_file(json_file_name):
verifs_nodes = read_ast_json_file(json_file_name)
verifs_nodes = filter(
lambda node: 'batch' in node['applications'] or 'iliad' in node['applications'],
verifs_nodes,
)
verif_functions_sources = map(python_source_visitors.visit_node, verifs_nodes)
return verif_functions_sources
......@@ -164,66 +148,78 @@ def main():
source='definition_by_variable_name = {}\n'.format(pprint.pformat(definition_by_variable_name, width=120)),
)
# # Transpile verification functions
#
# verif_functions_sources = list(
# mapcat(load_verifs_file, iter_ast_json_file_names(filenames=['coc*.json', 'coi*.json']))
# )
# verif_regles_source = as_lines((
# 'from ..formulas_helpers import *',
# 'from . import chap_ini_formulas\n',
# 'def get_error(code):',
# ' return code',
# 'def verif_regles(base_variables, saisie_variables):',
# ' results = chap_ini_formulas.compute(base_variables=base_variables, saisie_variables=saisie_variables)',
# ' globals().update(results)',
# textwrap.indent(''.join(verif_functions_sources), prefix=4 * ' '),
# ))
# write_source_file(
# file_name='verif_regles.py',
# source=verif_regles_source,
# )
# Read formulas order
# ordered_formulas_names = read_json_file(json_file_name=os.path.join('data', 'ordered_formulas.json'))
# Transpile verification functions
verif_functions_sources = list(
mapcat(load_verifs_file, iter_ast_json_file_names(filenames=['coc*.json', 'coi*.json']))
)
verif_regles_source = """\
from ..formulas_helpers import *
def get_error(code):
return code
def verif_regles(base_variables, saisie_variables):
import ipdb; ipdb.set_trace()
globals().update(results)
{}
""".format(textwrap.indent(''.join(verif_functions_sources), prefix=4 * ' '))
write_source_file(
file_name='verif_regles.py',
source=verif_regles_source,
)
# Transpile formulas
formula_name_and_source_pairs = list(mapcat(
regles_sources_dicts = list(mapcat(
load_regles_file,
iter_ast_json_file_names(filenames=['chap-*.json', 'res-ser*.json']),
))
source_by_formula_name = {}
for formula_name, source in formula_name_and_source_pairs:
if formula_name not in definition_by_variable_name:
log.warning('Formula "{}" has no definition.'.format(formula_name))
else:
applications = definition_by_variable_name[formula_name]['applications']
if formula_name not in source_by_formula_name or 'batch' in applications:
if formula_name in source_by_formula_name and 'batch' in applications:
log.warning('Formula "{}" already met from another application, '
'but this one of "batch" is prefered => keep the source of this one ({}).'.format(
formula_name, source))
source_by_formula_name[formula_name] = source
def iter_formula_name_and_source_pairs(preferred_application='batch'):
visited_applications_by_variable_name = {}
for regle_sources_dicts in regles_sources_dicts:
applications = regle_sources_dicts['applications']
assert applications is not None
for variable_name, source in regle_sources_dicts['sources']:
visited_applications = visited_applications_by_variable_name.get(variable_name)
if visited_applications is not None:
is_double_defined_in_preferred_application = preferred_application in applications and \
preferred_application in visited_applications
assert not is_double_defined_in_preferred_application, (variable_name, visited_applications)
log.debug(
'Variable "{}" already visited and had another application, '
'but this one of "{}" is prefered => keep the source ({}) '
'and the applications({}) of this one.'.format(
variable_name, preferred_application, source, applications))
visited_applications_by_variable_name[variable_name] = applications
if preferred_application in applications or visited_applications is None:
yield variable_name, source
source_by_formula_name = dict(iter_formula_name_and_source_pairs())
def get_formula_source(formula_name):
sanitized_formula_name = core.sanitized_variable_name(formula_name)
return source_by_formula_name[sanitized_formula_name] \
if sanitized_formula_name in source_by_formula_name \
else ('# WARNING: the variable "{name}" is used in a formula at least, but is not defined.\n' +
else ('# WARNING: the formula "{name}" is used in a formula (or more), but is not defined.\n'
'def {name}(): return 0\n').format(name=sanitized_formula_name)
# write_source_file(
# file_name='formulas.py',
# source=formulas_file_source(map(get_formula_source, ordered_formulas_names)),
# )
dependencies_by_formula_name = dependencies_helpers.load_dependencies_by_formula_name()
formula_names = set(dependencies_by_formula_name.keys()) | set(filter(
lambda variable_name: core.is_calculee_variable(variable_name) and not core.has_tag(variable_name, 'base'),
definition_by_variable_name,
))
dependencies_by_formula_name = formulas_helpers.load_dependencies_by_formula_name()
formula_names = pipe(
concatv(
dependencies_by_formula_name.keys(),
concat(dependencies_by_formula_name.values()),
definition_by_variable_name,
),
filter(
lambda variable_name: core.is_calculee_variable(variable_name) and \
not core.is_constant(variable_name) and (
not core.is_base_variable(variable_name) or core.is_restituee_variable(variable_name),
),
),
set,
)
write_source_file(
file_name='formulas.py',
source=formulas_file_source(map(get_formula_source, formula_names)),
......
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