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

Commit 2a163877 authored by Christophe Benz's avatar Christophe Benz

wip

parent c1592e8e
calculateur_impots.egg-info
calculateur_impots/dummy
state.json
......@@ -2,28 +2,37 @@
import importlib
import itertools
from .formulas_helpers import * # noqa
# Formulas helpers
saisies = None # Global variable used by formulas, filled by `evaluate_formulas`
saisies = None # Global variable used by formulas, filled by `calculate_formulas`
# Functional helpers
def is_positive(n):
return n > 0
def is_zero(n):
return n == 0
def mapcat(function, sequence):
return itertools.chain.from_iterable(map(function, sequence))
# Core functions
def calculate_formulas(variables_saisies=None):
def evaluate_formulas(variables_saisies=None):
"""
Evaluate formulas and return results.
Defer import of .generated.formulas module to inject `variables_saisies` in global variable.
"""
if variables_saisies is None:
variables_saisies = {}
global saisies
saisies = variables_saisies
importlib.import_module(name='.generated.formulas', package='calculateur_impots')
def get_variable_type(variable_name):
"""Return variable type given its name or None if `variable_name` is not a variable."""
from .generated.variables_definitions import variable_definition_by_name
return variable_definition_by_name.get(variable_name, {}).get('type')
# -*- coding: utf-8 -*-
def is_positive(n):
return n > 0
def is_zero(n):
return n == 0
This diff is collapsed.
......@@ -52,7 +52,7 @@ def main():
if args.saisies is not None \
else None
core.calculate_formulas(variables_saisies=variables_saisies)
core.evaluate_formulas(variables_saisies=variables_saisies)
calculees = [variable_name for variable_name in iter_variables_calculees(args.calculees)]
print(calculees)
......
# -*- coding: utf-8 -*-
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 ..core import get_variable_type, mapcat
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 pprint
# Main visitor
def visit_node(node, parenthesised=False):
"""
Main transpilation function which calls the specific transpilation functions below.
They are subfunctions to ensure they are never called directly.
"""
transpilation_function_name = 'visit_' + node['type']
if transpilation_function_name not in globals():
error_message = '"def {}(node):" is not defined, node = {}'.format(
transpilation_function_name,
pprint.pformat(node, width=120),
)
raise NotImplementedError(error_message)
transpilation_function = globals()[transpilation_function_name]
result = transpilation_function(node)
return result
# Specific visitors
# def visit_regle(node):
# import ipdb; ipdb.set_trace()
# -*- coding: utf-8 -*-
"""
Functions used to transpile the JSON AST nodes to Python source code.
Each function `visit_abc` handles a node of type 'abc' and returns either a string or None.
These functions are called by transpiler.core.
"""
# from functools import reduce
import itertools
import json
import logging
import pprint
import textwrap
from ..core import get_variable_type, mapcat
from .base import sanitized_variable_name
from .unloop_helpers import iter_unlooped_nodes
log = logging.getLogger(__name__)
# Helpers
def visit_infix_expression(node, operators={}):
def merge(*iterables):
for values in itertools.zip_longest(*iterables, fillvalue=UnboundLocalError):
for index, value in enumerate(values):
if value != UnboundLocalError:
yield index, value
tokens = (
visit_node(operand_or_operator)
if index == 0
else operators.get(operand_or_operator, operand_or_operator)
for index, operand_or_operator in merge(node['operands'], node['operators'])
)
return ' '.join(map(str, tokens))
def iter_find_nodes(node, type, skip_type=None):
"""
Iterates over all nodes matching `type` recursively in `node`.
"""
if isinstance(node, dict):
if node['type'] == type:
yield node
elif skip_type is None or node['type'] != skip_type:
yield from iter_find_nodes(node=list(node.values()), skip_type=skip_type, type=type)
elif isinstance(node, list):
for child_node in node:
if isinstance(child_node, (list, dict)):
yield from iter_find_nodes(node=child_node, skip_type=skip_type, type=type)
# Main visitor
deep_level = 0
def visit_node(node, parenthesised=False):
"""
Main transpilation function which calls the specific transpilation functions below.
They are subfunctions to ensure they are never called directly.
"""
global deep_level
transpilation_function_name = 'visit_' + node['type']
if transpilation_function_name not in globals():
error_message = '"def {}(node):" is not defined, node = {}'.format(
transpilation_function_name,
pprint.pformat(node, width=120),
)
raise NotImplementedError(error_message)
nb_prefix_chars = len('DEBUG:' + __name__) + 1
transpilation_function_short_name = transpilation_function_name[len('visit_'):]
prefix = '> {}{}'.format(
transpilation_function_short_name + ' ' * (nb_prefix_chars - len(transpilation_function_short_name) - 3) + ':',
' ' * deep_level * 4,
)
node_str = textwrap.indent(json.dumps(node, indent=4), prefix=prefix)[nb_prefix_chars:].lstrip()
log.debug(
'{}{}({})'.format(
' ' * deep_level * 4 + '={}=> '.format(deep_level),
transpilation_function_name,
node_str,
)
)
transpilation_function = globals()[transpilation_function_name]
deep_level += 1
result = transpilation_function(node)
assert result is not None
deep_level -= 1
unparenthesised_node_types = ('float', 'function_call', 'integer', 'string', 'symbol')
if parenthesised:
assert isinstance(result, str), result
if node['type'] not in unparenthesised_node_types:
result = '({})'.format(result)
log.debug(
'{}<={}= {}'.format(
' ' * deep_level * 4,
deep_level,
textwrap.indent(result, prefix='> ')[1:].lstrip() if isinstance(result, str) else str(result),
)
)
assert deep_level >= 0, deep_level
return result
# Specific visitors
def visit_boolean_expression(node):
return visit_infix_expression(node, operators={'et': 'and', 'ou': 'or'})
def visit_comparaison(node):
return '{} {} {}'.format(
visit_node(node['left_operand']),
{'=': '=='}.get(node['operator'], node['operator']),
visit_node(node['right_operand']),
)
def visit_variable_const(node):
return '{} = {}'.format(node['name'], node['value'])
def visit_dans(node):
return '{} {} {}'.format(
visit_node(node['expression'], parenthesised=True),
'not in' if node.get('negative_form') else 'in',
node['enumeration'],
)
def visit_enumeration_values(node):
return str(tuple(node['values']))
def visit_float(node):
return str(node['value'])
def visit_formula(node):
# state['current_formula'] = []
expression_source = visit_node(node['expression'])
# dependencies = set(map(
# lambda node: sanitized_variable_name(node['value']),
# iter_find_nodes(
# node=node,
# skip_type='loop_expression', # visit_loop_expression will handle its dependencies.
# type='symbol',
# ),
# ))
formula_name = sanitized_variable_name(node['name'])
# state['formulas_dependencies'][formula_name] = dependencies
source = '{} = {}'.format(formula_name, expression_source)
# state['formulas_sources'][formula_name] = source
# state['current_formula'] = None
return (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,
)
def visit_integer(node):
return str(node['value'])
def visit_interval(node):
return 'interval({}, {})'.format(node['first'], node['last'])
def visit_loop_expression(node):
# def iter_unlooped_loop_expressions_nodes_and_dependencies():
# for unlooped_node in iter_unlooped_nodes(
# loop_variables_nodes=node['loop_variables'],
# node=node['expression'],
# ):
# dependencies = set(map(
# lambda node: sanitized_variable_name(node['value']),
# iter_find_nodes(node=unlooped_node, type='symbol'),
# ))
# yield unlooped_node, dependencies
# unlooped_loop_expressions_nodes, dependencies = zip(*iter_unlooped_loop_expressions_nodes_and_dependencies())
# dependencies = reduce(or_, dependencies)
unlooped_loop_expressions_nodes = iter_unlooped_nodes(
loop_variables_nodes=node['loop_variables'],
node=node['expression'],
)
source = '[{}]'.format(
', '.join(map(visit_node, unlooped_loop_expressions_nodes)),
)
# state['current_formula'].append(dependencies)
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'],
node=node['formula'],
unloop_keys=['name'],
))
def visit_product_expression(node):
return visit_infix_expression(node)
def visit_regle(node):
return map(
lambda node1: visit_node(node1) if node1['type'] == 'pour_formula' else [visit_node(node1)],
node['formulas'],
)
def visit_sum_expression(node):
return visit_infix_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)
def visit_ternary_operator(node):
return '{} if {} else {}'.format(
visit_node(node['value_if_true'], parenthesised=True),
visit_node(node['condition'], parenthesised=True),
visit_node(node['value_if_false'], parenthesised=True) if 'value_if_false' in node else 0,
)
def visit_verif(node):
pass
# import ipdb; ipdb.set_trace()
# -*- coding: utf-8 -*-
"""
Unloop helpers are functions which deal with M language loops.
They take a node containing a loop and return as many "unlooped" nodes as the number of combination of loop variables.
"""
import copy
import itertools
from ..core import mapcat
def enumeration_node_to_sequence(enumeration_node):
if enumeration_node['type'] == 'enumeration_values':
return enumeration_node['values']
elif enumeration_node['type'] == 'interval':
return range(enumeration_node['first'], enumeration_node['last'] + 1)
else:
raise NotImplementedError('Unknown type for enumeration_node = {}'.format(enumeration_node))
def iter_unlooped_nodes(node, loop_variables_nodes, unloop_keys=None):
loop_variables_names, sequences = zip(*map(
lambda loop_variable_node: (
loop_variable_node['name'],
mapcat(
enumeration_node_to_sequence,
loop_variable_node['enumerations'],
),
),
loop_variables_nodes,
))
loop_variables_values_list = itertools.product(*sequences)
for loop_variables_values in loop_variables_values_list:
value_by_loop_variable_name = dict(zip(loop_variables_names, loop_variables_values))
yield unlooped(
node=node,
unloop_keys=unloop_keys,
value_by_loop_variable_name=value_by_loop_variable_name,
)
def unlooped(node, value_by_loop_variable_name, unloop_keys=None):
"""
Replace loop variables names by values given by `value_by_loop_variable_name` in symbols recursively found
in `node`.
"""
new_node = copy.deepcopy(node)
update_symbols(
node=new_node,
value_by_loop_variable_name=value_by_loop_variable_name,
)
if unloop_keys is not None:
for key in unloop_keys:
for loop_variable_name, loop_variable_value in value_by_loop_variable_name.items():
new_node[key] = new_node[key].replace(loop_variable_name, str(loop_variable_value), 1)
return new_node
def update_symbols(node, value_by_loop_variable_name):
"""
This function mutates `node` and returns nothing. Better use the `unlooped` function.
"""
if isinstance(node, dict):
if node['type'] == 'symbol':
for loop_variable_name, loop_variable_value in value_by_loop_variable_name.items():
node['value'] = node['value'].replace(loop_variable_name, str(loop_variable_value), 1)
else:
update_symbols(
node=list(node.values()),
value_by_loop_variable_name=value_by_loop_variable_name,
)
elif isinstance(node, list):
for child_node in node:
update_symbols(
node=child_node,
value_by_loop_variable_name=value_by_loop_variable_name,
)
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