Si vous constatez la perte de tickets, de commentaires ou de projets, veuillez consulter la page https://status.framasoft.org/incident/626.

Commit f845442c authored by Christophe Benz's avatar Christophe Benz

Update

parent 7ad0fc75
......@@ -3,6 +3,8 @@
import importlib
from toolz import keyfilter
from .formulas_helpers import * # noqa
......@@ -17,11 +19,17 @@ def evaluate_formulas(variables_saisies=None):
Evaluate formulas and return results.
Defer import of .generated.formulas module to inject `variables_saisies` in global variable.
"""
from .generated.variables_definitions import variable_definition_by_name
if variables_saisies is None:
variables_saisies = {}
global saisies
saisies = variables_saisies
importlib.import_module(name='.generated.formulas', package='calculateur_impots')
formulas_module = importlib.import_module(name='.generated.formulas', package='calculateur_impots')
variables_calculees = keyfilter(
lambda key: key in variable_definition_by_name,
formulas_module.__dict__,
)
return variables_calculees
def get_variable_type(variable_name):
......
# -*- coding: utf-8 -*-
def is_positive(n):
return n > 0
import math
from toolz.curried.operator import eq, ge, gt, ne
import infix
def is_zero(n):
return n == 0
# M language functions
arr = round
inf = math.floor
null = eq(0)
positif = gt(0)
positif_ou_nul = ge(0)
present = ne(0)
somme = sum
# Division operator overload
def safe_divide(a, b):
return 0 if b == 0 else a / b
div = infix.or_infix(safe_divide)
#
# class wrap_value(object):
# def __init__(self, value):
# self.value = value
#
# def __add__(self, other):
# return wrap_value(self.value + other.value)
#
# def __sub__(self, other):
# return wrap_value(self.value - other.value)
#
# def __mul__(self, other):
# return wrap_value(self.value * other.value)
#
# def __lt__(self, other):
# return wrap_value(self.value < other.value)
#
# def __gt__(self, other):
# return wrap_value(self.value > other.value)
#
# def __truediv__(self, other):
# # if other == 0:
# # import ipdb; ipdb.set_trace()
# return wrap_value(0 if other.value == 0 else self.value / other.value)
# def wrap_function(func):
# def apply(val):
# return wrap_value(func(val.value))
# return apply
# arr = wrap_function(round)
# inf = wrap_function(math.floor)
# null = wrap_function(eq(0))
# positif = wrap_function(gt(0))
# positif_ou_nul = wrap_function(ge(0))
# present = wrap_function(ne(0))
# somme = wrap_function(sum)
This diff is collapsed.
This diff is collapsed.
......@@ -14,11 +14,10 @@ import logging
import pprint
import textwrap
from m_language_parser.unloop_helpers import iter_unlooped_nodes
from toolz import mapcat
from ..core import get_variable_type
from .base import get_visitor, sanitized_variable_name
from .unloop_helpers import iter_unlooped_nodes
from . import core
log = logging.getLogger(__name__)
......@@ -27,6 +26,11 @@ log = logging.getLogger(__name__)
# Helpers
def sanitized_variable_name(value):
# Python variables must not begin with a digit.
return '_' + value if value[0].isdigit() else value
def visit_infix_expression(node, operators={}):
def interleave(*iterables):
for values in itertools.zip_longest(*iterables, fillvalue=UnboundLocalError):
......@@ -52,8 +56,8 @@ deep_level = 0
def visit_node(node, parenthesised=False):
"""Main visitor which calls the specific visitors below."""
global deep_level
visitor = get_visitor(module_name=__name__, node=node)
visitor_name = 'visit_' + node['type']
visitor = globals().get(visitor_name)
if visitor is None:
error_message = '"def {}(node):" is not defined, node = {}'.format(
visitor_name,
......@@ -127,33 +131,15 @@ def visit_float(node):
def visit_formula(node):
expression_source = visit_node(node['expression'])
formula_name = sanitized_variable_name(node['name'])
source = '{} = {}'.format(formula_name, expression_source)
return (formula_name, source)
sanitized_formula_name = sanitized_variable_name(node['name'])
source = '{} = {}'.format(sanitized_formula_name, expression_source)
return (sanitized_formula_name, source)
def visit_function_call(node):
m_function_name = node['name']
python_function_name_by_m_function_name = {
'abs': 'abs',
'arr': 'round',
# 'inf': '',
'max': 'max',
'min': 'min',
'null': 'is_zero',
# 'positif_ou_nul': '',
'positif': 'is_positive',
# 'present': '',
'somme': 'sum',
}
# TODO When all functions will be handled, use this assertion.
# assert m_function_name in python_function_name_by_m_function_name, \
# 'Unknown M function name: "{}"'.format(m_function_name)
# python_function_name = python_function_name_by_m_function_name[m_function_name]
python_function_name = python_function_name_by_m_function_name.get(m_function_name, m_function_name)
return '{name}({arguments})'.format(
arguments=', '.join(map(visit_node, node['arguments'])),
name=python_function_name,
name=node['name'],
)
......@@ -161,10 +147,6 @@ def visit_integer(node):
return str(node['value'])
def visit_interval(node):
return 'interval({}, {})'.format(node['first'], node['last'])
def visit_loop_expression(node):
unlooped_loop_expressions_nodes = iter_unlooped_nodes(
loop_variables_nodes=node['loop_variables'],
......@@ -176,16 +158,6 @@ def visit_loop_expression(node):
return source
def visit_loop_variable(node):
return ' or '.join(
'{} in {}'.format(
node['name'],
visit_node(enumerations_node),
)
for enumerations_node in node['enumerations']
)
def visit_pour_formula(node):
return map(visit_node, iter_unlooped_nodes(
loop_variables_nodes=node['loop_variables'],
......@@ -195,7 +167,7 @@ def visit_pour_formula(node):
def visit_product_expression(node):
return visit_infix_expression(node)
return visit_infix_expression(node, operators={'/': '|div|'})
def visit_regle(node):
......@@ -212,7 +184,7 @@ def visit_sum_expression(node):
def visit_symbol(node):
symbol = node['value']
return 'saisies.get({!r}, 0)'.format(symbol) \
if get_variable_type(symbol) == 'variable_saisie' else sanitized_variable_name(symbol)
if core.get_variable_type(symbol) == 'variable_saisie' else sanitized_variable_name(symbol)
def visit_ternary_operator(node):
......
......@@ -8,6 +8,7 @@ import os
import sys
from calculateur_impots import core
from toolz import keyfilter
# Globals
......@@ -20,47 +21,52 @@ log = logging.getLogger(script_name)
script_dir_name = os.path.dirname(os.path.abspath(__file__))
def iter_variables_calculees(variables):
for variable_name, value in iter_name_and_value(variables):
if value is None:
variable_definition = core.find_definition(variable_name)
assert variable_definition['type'] == 'variable_calculee', \
'Calculation requested for variable "{}" but type != "variable_calculee"'.format(variable_name)
yield variable_name
def iter_variables_calculees(variables_names):
for variable_name in variables_names:
variable_type = core.get_variable_type(variable_name)
if variable_type != 'variable_calculee':
parser.error('Variable "{}" is not a variable calculee'.format(variable_name))
yield variable_name
def iter_variables_saisies(values):
for variable_name, value in map(lambda value: value.split('=', 1), values):
variable_definition = core.find_definition(variable_name)
assert variable_definition['type'] == 'variable_saisie', \
'Value provided for variable "{}" but type != "variable_saisie"'.format(variable_name)
yield variable_name, value
for value, words in map(lambda value: (value, value.strip('=').split('=', 1)), values):
if len(words) == 1:
parser.error('Missing value for variable saisie: "{}"'.format(value))
elif len(words) > 2:
parser.error('Invalid syntax for variable saisie: "{}"'.format(value))
variable_name, variable_value = words
try:
variable_value = float(variable_value)
except ValueError:
parser.error('Variable "{}" value is not a float'.format(value))
variable_type = core.get_variable_type(variable_name)
if variable_type != 'variable_saisie':
parser.error('Variable "{}" is not a variable saisie'.format(variable_name))
yield variable_name, variable_value
def main():
global args, parser
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('-v', '--verbose', action='store_true', default=False, help='Increase output verbosity')
parser.add_argument('-d', '--debug', action='store_true', default=False, help='Display debug messages')
parser.add_argument('--calculee', dest='calculees', default='IINET', metavar='variable', nargs='+',
help='Variables calculées')
parser.add_argument('--saisie', dest='saisies', metavar='variable', nargs='+', help='Variables saisies')
global args
parser.add_argument('calculees', default=['IINET'], metavar='variable', nargs='*', help='Variables calculées')
parser.add_argument('--saisie', dest='saisies', metavar='nom=valeur', nargs='+', help='Variables saisies')
args = parser.parse_args()
logging.basicConfig(level=logging.DEBUG if args.verbose or args.debug else logging.WARNING, stream=sys.stdout)
variables_saisies = {variable_name: value for variable_name, value in iter_variables_saisies(args.saisies)} \
variables_saisies = dict(iter_variables_saisies(args.saisies)) \
if args.saisies is not None \
else None
log.debug('variables_saisies: {}'.format(variables_saisies))
core.evaluate_formulas(variables_saisies=variables_saisies)
calculees = [variable_name for variable_name in iter_variables_calculees(args.calculees)]
print(calculees)
# result = {
# variable_name: simulation.calculate(variable_name)
# for variable_name in variables_calculees
# }
# print(result)
requested_variables_calculees = list(iter_variables_calculees(args.calculees))
variables_calculees = core.evaluate_formulas(variables_saisies=variables_saisies)
print(keyfilter(
lambda key: key in requested_variables_calculees,
variables_calculees,
))
return 0
......
......@@ -19,11 +19,9 @@ import pprint
import sys
import textwrap
from toolz.curried import concat, map, mapcat, pipe, valmap
from toolz.curried import concat, filter, map, mapcat, pipe, sorted, valmap
from calculateur_impots.transpiler import to_dependencies_visitors, to_python_source_visitors
from calculateur_impots.transpiler.base import sanitized_variable_name
from calculateur_impots.transpiler.dependencies_helpers import get_ordered_formulas_names
from calculateur_impots import python_source_visitors
# Globals
......@@ -45,12 +43,16 @@ def lines_to_python_source(sequence):
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.join(args.json_dir, json_file_name)
with open(json_file_path) as json_file:
json_str = json_file.read()
nodes = json.loads(json_str)
assert isinstance(nodes, list)
return nodes
return json.loads(json_str)
def write_source_file(file_name, source):
......@@ -68,9 +70,9 @@ def write_source_file(file_name, source):
# Load files functions
def iter_json_file_names(*pathnames):
def iter_ast_json_file_names(*pathnames):
for json_file_path in sorted(mapcat(
lambda pathname: glob.iglob(os.path.join(args.json_dir, pathname)),
lambda pathname: glob.iglob(os.path.join(args.json_dir, 'ast', pathname)),
pathnames,
)):
json_file_name = os.path.basename(json_file_path)
......@@ -82,25 +84,12 @@ def iter_json_file_names(*pathnames):
def load_regles_file(json_file_name):
log.info('Loading "{}"...'.format(json_file_name))
regles_nodes = read_ast_json_file(json_file_name)
batch_application_regles_nodes = list(filter(
batch_application_regles_nodes = filter(
lambda node: 'batch' in node['applications'],
regles_nodes,
))
formula_name_and_source_list = list(mapcat(to_python_source_visitors.visit_node, batch_application_regles_nodes))
formula_name_and_dependencies_list = list(mapcat(to_dependencies_visitors.visit_node, batch_application_regles_nodes))
return formula_name_and_source_list, formula_name_and_dependencies_list
def load_tgvH_file():
json_file_name = 'tgvH.json'
log.info('Loading "{}"...'.format(json_file_name))
nodes = read_ast_json_file(json_file_name)
variable_definition_by_name = {
sanitized_variable_name(node['name']): node
for node in nodes
if node['type'].startswith('variable_')
}
return variable_definition_by_name
)
formula_name_and_source_pairs = mapcat(python_source_visitors.visit_node, batch_application_regles_nodes)
return formula_name_and_source_pairs
# def load_verifs_file(json_file_name):
......@@ -116,7 +105,7 @@ def main():
parser.add_argument('-d', '--debug', action='store_true', default=False, help='Display debug messages')
parser.add_argument('--json', nargs='+', help='Parse only this JSON file and exit')
parser.add_argument('-v', '--verbose', action='store_true', default=False, help='Increase output verbosity')
parser.add_argument('json_dir', help='Directory containing the JSON AST files')
parser.add_argument('json_dir', help='Directory containing the JSON AST files and semantic data')
global args
args = parser.parse_args()
logging.basicConfig(
......@@ -129,49 +118,48 @@ def main():
if args.json is not None:
for json_file_name in args.json:
json_file_path = os.path.join(args.json_dir, json_file_name)
json_file_path = os.path.join(args.json_dir, 'ast', json_file_name)
if not os.path.exists(json_file_path):
parser.error('JSON file "{}" does not exist.'.format(json_file_path))
# Transpile variables definitions (before regles)
variable_definition_by_name = load_tgvH_file()
# Transpile constants
constants_file_name = os.path.join('semantic_data', 'constants.json')
constants = read_json_file(json_file_name=constants_file_name)
constants_source = pipe(
constants.items(),
sorted,
map(lambda item: '{} = {}'.format(*item)),
lines_to_python_source,
)
write_source_file(
file_name='constants.py',
source=constants_source,
)
# Transpile variables definitions
variables_definitions_file_name = os.path.join('semantic_data', 'variables_definitions.json')
variable_definition_by_name = read_json_file(json_file_name=variables_definitions_file_name)
write_source_file(
file_name='variables_definitions.py',
source='variable_definition_by_name = {}\n'.format(pprint.pformat(variable_definition_by_name, width=120)),
)
# Load regles
# Transpile formulas
vals = list(map(
formula_source_by_name = dict(list(mapcat(
load_regles_file,
iter_json_file_names('chap-*.json', 'res-ser*.json'),
))
formula_source_by_name, formula_dependencies_by_name = pipe(
zip(*vals),
map(concat),
map(dict),
)
formula_dependencies_by_name = valmap(list, formula_dependencies_by_name)
iter_ast_json_file_names('chap-*.json', 'res-ser*.json'),
)))
assert formula_source_by_name
dependencies_file_path = os.path.join(generated_dir_path, 'dependencies.json')
with open(dependencies_file_path, 'w') as dependencies_file:
dependencies_file.write(json.dumps(formula_dependencies_by_name, indent=2))
log.info('Output file "{}" written with success'.format(dependencies_file_path))
variables_dependencies_file_name = os.path.join('semantic_data', 'variables_dependencies.json')
variable_dependencies_by_name = read_json_file(json_file_name=variables_dependencies_file_name)
# Load verifs
# for json_file_name in iter_json_file_names('coc*.json', 'coi*.json'):
# load_verifs_file(json_file_name)
ordered_formulas_names_file_name = os.path.join('semantic_data', 'ordered_formulas.json')
ordered_formulas_names = read_json_file(json_file_name=ordered_formulas_names_file_name)
# Output transpiled sources to Python file.
variables_const = list(filter(lambda v: v['type'] == 'variable_const', variable_definition_by_name.values()))
ordered_formulas_names = get_ordered_formulas_names(formula_dependencies_by_name)
write_source_file(
file_name='constants.py',
source=lines_to_python_source(map(
to_python_source_visitors.visit_node,
sorted(variables_const, key=itemgetter('name')),
)),
)
write_source_file(
file_name='formulas.py',
source=lines_to_python_source(itertools.chain(
......@@ -180,12 +168,16 @@ def main():
'from .constants import *',
'\n',
),
map(
lambda formula_name: formula_source_by_name.get(formula_name, '{} = 0'.format(formula_name)),
pipe(
ordered_formulas_names,
map(python_source_visitors.sanitized_variable_name),
map(lambda formula_name: formula_source_by_name.get(formula_name, '{} = 0'.format(formula_name))),
),
)),
)
),
))
# for json_file_name in iter_ast_json_file_names('coc*.json', 'coi*.json'):
# load_verifs_file(json_file_name)
return 0
......
# -*- coding: utf-8 -*-
import inspect
import sys
def get_visitor(node, module_name):
visitor_name = 'visit_' + node['type']
module = sys.modules[module_name]
module_visitors = filter(lambda member: member[0].startswith('visit_'), inspect.getmembers(module))
visitor = dict(module_visitors).get(visitor_name)
return visitor
def sanitized_variable_name(value):
# Python variables must not begin with a digit.
return '_' + value if value[0].isdigit() else value
# -*- coding: utf-8 -*-
"""
Dependencies helpers are functions dealing with enumerating formulas in the right order.
"""
import logging
from toolz import mapcat
from ..core import get_variable_type
log = logging.getLogger(__name__)
# Helpers
def is_writable(formula_name):
return get_variable_type(formula_name) == 'variable_calculee'
# Main function
def get_ordered_formulas_names(formula_dependencies_by_name):
"""Return a list of formula names in a dependencies resolution order."""
writable_formula_dependencies_by_name = {
key: set(filter(is_writable, values))
for key, values in formula_dependencies_by_name.items()
if is_writable(key)
}
writable_formulas_names = set(mapcat(
lambda item: set([item[0]]) | set(item[1]), writable_formula_dependencies_by_name.items()
))
assert all(map(is_writable, writable_formulas_names))
ordered_formulas_names = []
expected_formulas_count = len(writable_formulas_names) # Store it since writable_formulas_names is mutated below.
while len(ordered_formulas_names) < expected_formulas_count:
# Iterate sorted set for stability between executions of the script.
for writable_formula_name in sorted(writable_formulas_names):
if all(map(
lambda dependency_name: dependency_name in ordered_formulas_names,
writable_formula_dependencies_by_name.get(writable_formula_name, []),
)):
ordered_formulas_names.append(writable_formula_name)
log.debug('{} ordered_formulas_names added (latest {})'.format(
len(ordered_formulas_names),
writable_formula_name,
))
writable_formulas_names -= set(ordered_formulas_names)
return ordered_formulas_names
# -*- coding: utf-8 -*-
import logging
from toolz.curried import concat, filter, mapcat, pipe
from .base import get_visitor, sanitized_variable_name
from .unloop_helpers import iter_unlooped_nodes
log = logging.getLogger(__name__)
# Main visitor
deep_level = 0
def visit_node(node):
"""Main visitor which calls the specific visitors below."""
global deep_level
visitor = get_visitor(module_name=__name__, node=node)
log.debug('{}:{}:{}'.format(deep_level, visitor.__name__, node))
deep_level += 1
result = list(visitor(node))
deep_level -= 1
log.debug('{}:{}: => {}'.format(deep_level, visitor.__name__, result))
assert deep_level >= 0, deep_level
return result
# Specific visitors
def visit_boolean_expression(node):
return mapcat(visit_node, node['operands'])
def visit_comparaison(node):
return mapcat(visit_node, (node['left_operand'], node['right_operand']))
def visit_dans(node):
return visit_node(node['expression'])
# def visit_enumeration_values(node):
def visit_float(node):
return []
def visit_formula(node):
sanitized_formula_name = sanitized_variable_name(node['name'])
dependencies = set(visit_node(node['expression']))
sanitized_dependencies = set(map(sanitized_variable_name, dependencies))
return (sanitized_formula_name, sanitized_dependencies)
def visit_function_call(node):
return mapcat(visit_node, node['arguments'])
def visit_integer(node):
return []
def visit_interval(node):
return []
def visit_loop_expression(node):
return mapcat(visit_node, iter_unlooped_nodes(
loop_variables_nodes=node['loop_variables'],
node=node['expression'],
))
def visit_pour_formula(node):
return map(visit_node, iter_unlooped_nodes(
loop_variables_nodes=node['loop_variables'],
node=node['formula'],
unloop_keys=['name'],
))
def visit_product_expression(node):
return mapcat(visit_node, node['operands'])
def visit_regle(node):