Commit d1403d21 authored by Christophe Benz's avatar Christophe Benz

Get rid of global variables

parent a08b58d7
# -*- coding: utf-8 -*-
import json
import math
import os
import pkg_resources
from toolz.curried import first
from . import core
"""Functions used by formulas"""
# M language functions
# M language functions translated to Python
arr = round
......@@ -46,97 +39,15 @@ somme = sum
# Cache decorator
# FIXME Do not use a global variable.
result_by_formula_name_cache = {}
def cached(formula_function):
formula_name = formula_function.__name__
def cached_formula():
if formula_name not in result_by_formula_name_cache:
result = formula_function()
result_by_formula_name_cache[formula_name] = result
return result_by_formula_name_cache[formula_name]
return cached_formula
# Dependencies helpres
def inspect():
ok_variable_tuples = inspect_cache(predicate=lambda x: x == 2461)
ok_variable_names = list(map(first, ok_variable_tuples))
print('restituee variables in batch depending on one of {}: {}'.format(
ok_variable_names,
'\n'.join(
'{} {!r}'.format(
formula_name,
core.get_variable_description(formula_name),
)
for formula_name in filter(
core.is_batch,
find_restituee_variables_depending_on(ok_variable_names),
)
) + '\n',
))
def inspect_cache(predicate=None):
return [
(
formula_name,
result,
core.get_variable_description(formula_name),
core.get_variable_applications(formula_name),
)
for formula_name, result in sorted(result_by_formula_name_cache.items())
if (predicate(result) if predicate is not None else True)
]
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, visited_dependencies):
if formula_name in visited_dependencies:
return False
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(
visited_dependencies=visited_dependencies,
formula_name=dependency_name,
):
return True
return False
return walk_dependencies(
formula_name=formula_name,
visited_dependencies=set(),
)
def find_restituee_variables_depending_on(formula_names):
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(
formula_name=restituee_variable_name,
dependency_formula_names=formula_names,
dependencies_by_formula_name=dependencies_by_formula_name,
),
restituee_variable_names,
))
def cached(result_by_formula_name_cache):
def cached_decorator(formula_function):
formula_name = formula_function.__name__
def cached_formula():
if formula_name not in result_by_formula_name_cache:
result = formula_function()
result_by_formula_name_cache[formula_name] = result
return result_by_formula_name_cache[formula_name]
return cached_formula
return cached_decorator
This diff is collapsed.
This source diff could not be displayed because it is too large. You can view the blob instead.
......@@ -26,6 +26,21 @@ log = logging.getLogger(__name__)
# Helpers
def make_formula_source(formula_name, expression, cached=True, description=None):
sanitized_name = core.sanitized_variable_name(formula_name)
return """\
{decorator}def {sanitized_name}():
{docstring}return {expression}
formulas[{formula_name!r}] = {sanitized_name}
""".format(
decorator='@cached(cache)\n' if cached else '',
docstring='' if description is None else '"""{}"""\n '.format(description),
expression=expression,
formula_name=formula_name,
sanitized_name=sanitized_name,
)
def visit_infix_expression(node, operators={}):
def interleave(*iterables):
for values in itertools.zip_longest(*iterables, fillvalue=UnboundLocalError):
......@@ -136,27 +151,15 @@ def visit_float(node):
def visit_formula(node):
formula_name = node['name']
sanitized_formula_name = core.sanitized_variable_name(formula_name)
sanitized_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 = """\
@cached
def {formula_name}():
\"\"\"
{docstring}
\"\"\"
return {expression}
""".format(
docstring=textwrap.indent('\n'.join(docstring_lines), prefix=4 * ' '),
source = make_formula_source(
description=description,
expression=expression_source,
formula_name=sanitized_formula_name,
formula_name=formula_name,
)
return (sanitized_formula_name, source)
return (sanitized_name, source)
def visit_function_call(node):
......@@ -216,12 +219,13 @@ def visit_sum_expression(node):
def visit_symbol(node):
symbol = node['value']
sanitized_name = core.sanitized_variable_name(symbol)
return 'saisie_variables.get({!r}, 0)'.format(symbol) \
if core.is_saisie_variable(symbol) \
else (
'{}()'.format(core.sanitized_variable_name(symbol))
'formulas[{!r}]()'.format(symbol)
if core.is_calculee_variable(symbol)
else core.sanitized_variable_name(symbol)
else sanitized_name
)
......
......@@ -11,7 +11,7 @@ import sys
from toolz.curried import map, pipe, unique
from calculateur_impots import core, formulas_helpers
from calculateur_impots import core
from calculateur_impots.generated import formulas, verifs
# from calculateur_impots.generated.variables_definitions import definition_by_variable_name
......@@ -60,6 +60,7 @@ def main():
parser.add_argument('--calculees', dest='calculee_variables', metavar='variable', nargs='+',
help='Variables calculées')
parser.add_argument('--saisies', dest='saisie_variables', metavar='nom=valeur', nargs='+', help='Variables saisies')
parser.add_argument('--no-verifs', action='store_true', default=False, help='Skip verifs step')
args = parser.parse_args()
logging.basicConfig(
level=logging.DEBUG if args.debug else (logging.INFO if args.verbose else logging.WARNING),
......@@ -73,84 +74,40 @@ def main():
log.warning(not_restituee_message)
# parser.error(not_restituee_message)
# Required variables_saisies: V_ANREV, V_REGCO (tag "contexte"?)
# src/tgvH.m:968:APPLI_AVIS : saisie contexte classe = 0 priorite = 10 categorie_TL = 20 restituee alias AVIS : "Application Avis" ; # noqa
# src/tgvH.m:970:APPLI_COLBERT : saisie contexte classe = 0 priorite = 0 categorie_TL = 0 restituee alias V_COLBERT : "Appli_Colbert" type BOOLEEN ; # noqa
# src/tgvH.m:973:APPLI_OCEANS : saisie contexte classe = 0 priorite = 0 categorie_TL = 0 restituee alias V_OCEANS : "Appli_Oceans" ; # noqa
# src/tgvH.m:2299:CODE_2042 : saisie contexte classe = 0 priorite = 0 categorie_TL = 0 alias V_CODE_2042 : "nouv cor: code majoration de la 2042 fourni par l interface" ; # noqa
# src/tgvH.m:5470:ITREDFRI : saisie contexte classe = 0 priorite = 0 categorie_TL = 0 restituee alias V_REDFRI : "ITRED de la decla finale avant application 1731 bis " type REEL ; # noqa
# src/tgvH.m:14401:V_ANCSDED : saisie contexte classe = 0 priorite = 10 categorie_TL = 20 alias ANCSDED : "Annee de revenu pour variation CSG" ; # noqa
# src/tgvH.m:14826:V_ROLCSG : saisie contexte classe = 0 priorite = 10 categorie_TL = 20 alias ROLCSG : "numero de role CSG" ; # noqa
# src/tgvH.m:14523:V_CALCULIR : saisie contexte classe = 0 priorite = 51 categorie_TL = 10 alias CALCULIR : "= 0 si saisie 2042 ILIAD , = 1 si CALCULIR sous ILIAD" ; # noqa
# src/tgvH.m:14524:V_CALCULMAJO : saisie contexte classe = 0 priorite = 51 categorie_TL = 10 alias CALCULMAJO : "pour appel Denature rappel" ; # noqa
# src/tgvH.m:14588:V_ETCVL : saisie contexte classe = 0 priorite = 51 categorie_TL = 10 alias ETCVL : "ISF : Variable relative a l etat civil " type BOOLEEN ; # noqa
# src/tgvH.m:14682:V_REGANT : saisie contexte classe = 0 priorite = 51 categorie_TL = 10 alias REGANT : "Valeur de REGCO evt -1 calculee " ; # noqa
# src/tgvH.m:14683:V_REGCO : saisie contexte classe = 0 priorite = 51 categorie_TL = 10 alias REGCO : "Valeur de REGCO calculee (Cf. VALREGCO)" ; # noqa
# src/tgvH.m:14880:V_TRCT : saisie contexte classe = 0 priorite = 51 categorie_TL = 10 alias TRCT : "Variable relative a la telecorrection" ; # noqa
saisie_variables = {}
# src/tgvH.m:14604:V_IND_TRAIT : saisie contexte classe = 0 priorite = 10 categorie_TL = 20 alias IND_TRAIT : "indicateur de nature de trait. primitif ou correctif" ; # noqa
# 0 = primitif ?
# 4 = calcul correctif mais utilisant d'autres calculs que le 5 ?
# 5 = calcul correctif
saisie_variables['V_IND_TRAIT'] = 0
# src/tgvH.m:14211:VALREGCO : saisie revenu classe = 0 priorite = 20 categorie_TL = 10 cotsoc = 5 ind_abat = 0 acompte = 1 avfisc = 0 rapcat = 13 sanction = 0 nat_code = 1 alias 8ZA : "Regime d'imposition des non residents - Valeur 1, 2 ou 4" ; # noqa
# saisie_variables['VALREGCO'] = 1
# src/tgvH.m:14404:V_ANREV : saisie contexte classe = 0 priorite = 10 categorie_TL = 20 alias ANREV : "Annee des revenus" type REEL ; # noqa
saisie_variables['V_ANREV'] = 2014
# src/tgvH.m:14339:V_0DA : saisie famille classe = 0 priorite = 20 categorie_TL = 10 nat_code = 0 restituee alias 0DA : "Annee de naissance du declarant" type DATE_AAAA ; # noqa
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
# Avoid division by 0
# saisie_variables['CARPENBAV'] = 1
# saisie_variables['CARPENBAC'] = 1
saisie_variables = {
'V_ANREV': 2014,
}
if args.saisie_variables is not None:
saisie_variables.update(iter_saisie_variables(args.saisie_variables))
log.debug('saisie_variables: {}'.format(saisie_variables))
# base_variables_value_by_name = {} # Override base variables values
# base_variables = {
# variable_name: base_variables_value_by_name.get(variable_name, 0)
# for variable_name, variable_definition in definition_by_variable_name.items()
# if core.is_base_variable(variable_name)
# }
# errors = verifs.get_errors(
# base_variables=base_variables,
# saisie_variables=saisie_variables,
# )
errors = None
if errors is not None:
errors_definitions = load_errors_definitions()
definition_by_error_name = pipe(errors_definitions, map(lambda d: (d['name'], d)), dict)
print('errors: {}'.format(
json.dumps(
[
(error, definition_by_error_name.get(error, {}).get('description', 'No error description found'))
for error in unique(errors) # Keep order
],
indent=4,
),
))
result_by_formula_name_cache = {}
formulas_functions = formulas.get_formulas(
# base_variables=base_variables,
base_variables=None,
cache=result_by_formula_name_cache,
saisie_variables=saisie_variables,
)
if not args.no_verifs:
errors = verifs.get_errors(
formulas=formulas_functions,
saisie_variables=saisie_variables,
)
if errors is not None:
errors_definitions = load_errors_definitions()
definition_by_error_name = pipe(errors_definitions, map(lambda d: (d['name'], d)), dict)
print('errors: {}'.format(
json.dumps(
[
(
error,
definition_by_error_name.get(error, {}).get('description', 'No error description found'),
)
for error in unique(errors) # Keep order
],
indent=4,
),
))
for calculee_variable_name in args.calculee_variables:
sanitized_calculee_variable = core.sanitized_variable_name(calculee_variable_name)
# requested_formula_result = formulas_functions[sanitized_calculee_variable]()
......
......@@ -8,6 +8,7 @@ Transpile (roughly means convert) a JSON AST file to Python source code.
from operator import itemgetter
from string import Template
import argparse
import copy
import glob
......@@ -15,6 +16,7 @@ import itertools
import json
import logging
import os
import pkg_resources
import pprint
import sys
import textwrap
......@@ -35,38 +37,13 @@ script_dir_path = os.path.dirname(os.path.abspath(__file__))
generated_dir_path = os.path.abspath(os.path.join(script_dir_path, '..', 'generated'))
# Source code helper functions
def formulas_file_source(formulas_sources):
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 * ' '))
# Functions to write source code files
def as_lines(sequence):
return ''.join(concat(zip(sequence, itertools.repeat('\n'))))
def read_ast_json_file(json_file_name):
nodes = read_json_file(os.path.join('ast', json_file_name))
assert isinstance(nodes, list)
return nodes
def read_json_file(json_file_name):
json_file_path = os.path.abspath(os.path.join(args.json_dir, json_file_name))
log.info('Loading "{}"...'.format(json_file_path))
with open(json_file_path) as json_file:
json_str = json_file.read()
return json.loads(json_str)
def write_source_file(file_name, source):
header = """\
# -*- coding: utf-8 -*-
......@@ -79,7 +56,7 @@ def write_source_file(file_name, source):
log.info('Output file "{}" written with success'.format(file_path))
# Load files functions
# Functions to read JSON data files
def iter_ast_json_file_names(filenames, excluded_filenames=None):
......@@ -95,6 +72,16 @@ def iter_ast_json_file_names(filenames, excluded_filenames=None):
yield json_file_name
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 load_regles_file(json_file_name):
return pipe(
read_ast_json_file(json_file_name),
......@@ -111,6 +98,20 @@ def load_verifs_file(json_file_name):
)
def read_ast_json_file(json_file_name):
nodes = read_json_file(os.path.join('ast', json_file_name))
assert isinstance(nodes, list)
return nodes
def read_json_file(json_file_name):
json_file_path = os.path.abspath(os.path.join(args.json_dir, json_file_name))
log.info('Loading "{}"...'.format(json_file_path))
with open(json_file_path) as json_file:
json_str = json_file.read()
return json.loads(json_str)
# Main
......@@ -158,20 +159,17 @@ def main():
verif_sources = list(
mapcat(load_verifs_file, iter_ast_json_file_names(filenames=['coc*.json', 'coi*.json']))
)
verifs_source = """\
from . import formulas
from ..formulas_helpers import *
from .constants import *
def get_errors(base_variables, saisie_variables):
formulas_functions = formulas.get_formulas(base_variables, saisie_variables)
globals().update(formulas_functions)
# Special variables not defined in tgvH
V_9VV = 1
verifs_source = Template("""\
from ..formulas_helpers import arr, cached, inf, interval, null, positif, positif_ou_nul, present, somme
from .constants import * # noqa
def get_errors(formulas, saisie_variables):
errors = []
{}
$verifs
return errors or None
""".format(textwrap.indent('\n'.join(verif_sources), prefix=4 * ' '))
""").substitute(verifs=textwrap.indent('\n'.join(verif_sources), prefix=4 * ' '))
write_source_file(
file_name='verifs.py',
source=verifs_source,
......@@ -185,29 +183,53 @@ def get_errors(base_variables, saisie_variables):
)))
def get_formula_source(formula_name):
# not core.has_tag('base', formula_name) and \
# 'corrective' not in definition_by_variable_name.get(formula_name, {}).get('regle_tags', []) \
sanitized_name = core.sanitized_variable_name(formula_name)
return source_by_formula_name[sanitized_name] \
if sanitized_name in source_by_formula_name \
else 'def {}(): return 0\n'.format(sanitized_name)
else python_source_visitors.make_formula_source(
cached=False,
formula_name=formula_name,
expression='0',
)
def should_be_in_formulas_file(variable_name):
return not core.is_saisie_variable(variable_name) and not core.is_constant(variable_name)
dependencies_by_formula_name = formulas_helpers.load_dependencies_by_formula_name()
dependencies_by_formula_name = load_dependencies_by_formula_name()
formula_names = pipe(
concatv(
dependencies_by_formula_name.keys(),
concat(dependencies_by_formula_name.values()),
definition_by_variable_name.keys(),
),
filter(should_be_in_formulas_file),
set,
filter(should_be_in_formulas_file),
sorted,
)
write_source_file(
file_name='formulas.py',
source=formulas_file_source(map(get_formula_source, formula_names)),
source=Template("""\
import inspect
from ..formulas_helpers import arr, cached, inf, interval, null, positif, positif_ou_nul, present, somme
from .constants import * # noqa
def get_formulas(cache, saisie_variables):
formulas = {}
$formulas
return {
key: val
for key, val in locals().items()
if inspect.isfunction(val)
}
""").substitute(
formulas=textwrap.indent(
'\n'.join(map(get_formula_source, formula_names)),
prefix=4 * ' ',
),
),
)
return 0
......
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