<filename>tax_utils.py


__can_cache__ = True
__tabulate_avail__ = True
__plotly_avail__ = True
__config_name__ = 'taxes.ini'


import json
import os
import configparser
import requests
import numpy as np
import pandas as pd

try:
    import caching3 as cache
except ImportError as err:
    # pdb.set_trace()
    __can_cache__ = False
    print("Failed to import caching lib: %s" % err.msg)

try:
    import plotly.io as pio
except ImportError as err:
    __plotly_avail__ = True
    print("Failed to import plotting lib: %s" % err.msg)


def get_request(url='', refresh=False, headers_=None,
                method='get', payload_=None):
    """
    simple wrapper to avoid querying multiple times
    meaningless if cache library not available

    """
    # print __file__
    if __can_cache__:
        cache_key = cache.create_cache_key(
            locals(), cache.current_function_name())
        ret_val = cache.extract_cache_data(cache_key, refresh)
    else:
        ret_val = None
    # pdb.set_trace()
    if ret_val is None:
        if headers_ is None:
            if method == 'get':
                ret_val = requests.get(url)
            elif method == 'post':
                if payload_ is not None:
                    # print("got to the right place")
                    ret_val = requests.post(
                        url, data=json.dumps(payload_), headers=headers_)
                else:
                    ret_val = requests.post(url)
            else:
                raise NotImplementedError(
                    "Method == '%s' not implemented" % method)
        else:
            if method == 'get':
                ret_val = requests.get(url, headers=headers_)
            elif method == 'post':
                if payload_ is not None:
                    # print("got to the right place")
                    ret_val = requests.post(url, data=json.dumps(
                        payload_, cls=NpEncoder), headers=headers_)
                else:
                    ret_val = requests.post(url)
            else:
                raise Exception("Unknown method == '%s'" % method)

        if __can_cache__:
            cache.set_gp_cache(cache_key, ret_val)
    return ret_val
class NpEncoder(json.JSONEncoder):
    """
    Needed for when we produce the test cases (since they get created with np.int64s instead of ints, which JSONEncoder can't handle)
    """

    def default(self, obj):
        if isinstance(obj, np.integer):
            return int(obj)
        if isinstance(obj, np.floating):
            return float(obj)
        if isinstance(obj, np.ndarray):
            return obj.tolist()
        return super(NpEncoder, self).default(obj)

def tax_parameters(jurisdiction='NO', tax_year=None):
    """
    this reads in the parameters from the config and makes them easily usable from other functions
    """

    if not os.path.exists(__config_name__):
        raise Exception(
            "You need a file called '%s' with all relevant tax parameters in it!" % (__config_name__))

    if tax_year is None:
        tax_year = pd.to_datetime('today').year
    key = '%s-%d' % (jurisdiction, tax_year)
    config = configparser.ConfigParser()
    config.read(__config_name__)
    try:
        return config[key]
    except BaseException:
        raise Exception("Tax year = %d not implemented for %s jurisdiction" % (tax_year, jurisdiction))

def tax_round(in_float):
    """
    For values exactly halfway between rounded decimal values, NumPy
    rounds to the nearest even value. Thus 1.5 and 2.5 round to 2.0,
    -0.5 and 0.5 round to 0.0, etc.
    """
    if in_float == int(in_float):
        return int(in_float)

    _lo = np.floor(in_float)
    _hi = np.ceil(in_float)
    dist_to_hi = abs(in_float - _hi)
    dist_to_lo = abs(in_float - _lo)
    if dist_to_hi <= dist_to_lo:
        return int(_hi)
    # elif dist_to_lo < dist_to_hi:
    return int(_lo)
def big_fmt(val):
    """
    format with thousandsd separator
    """
    if pd.isnull(val):
        return 'n/a'

    if isinstance(val, str):
        return val

    return '{:,}'.format(int(round(val, 0))).strip()


def display_df(frame, title=None, header_size=1):
    """
    Show the dataframe in the browser
    """
    if not __plotly_avail__:
        raise Exception("You don't have plotly installed")
    raw = frame.to_html(index=False)
    if title is not None:
        raw = '<h%d>%s</h%d>%s' % (header_size, title, header_size, raw)
    pio._base_renderers.open_html_in_browser(raw)

def check_config_cases_vs_web(verbose=False, rtol=1e-5, atol=1e-8, case_file='no_test_cases.ini', official_func=None):
    """
    Mostly to catch fat fingers when punching in test cases

    example usage:


    import tax_norway as tn
    import tax_utils as tut
    tut.check_config_cases_vs_web(official_func=tn.produce_tax_results)
    """
    cases = configparser.ConfigParser()
    cases.read(case_file)
    case_numbers = sorted(map(int, list(set(cases.keys()) - set(['DEFAULT']))))
    with requests.Session() as sesh:
        for idx in case_numbers:
            assert check_test_case_config_vs_web(
                case_idx=idx, session=sesh, verbose=verbose, atol=atol, rtol=rtol, official_func=official_func), "Tax for case %s is off!" % idx

    print("All tax figures in config tested successfully!")

def check_test_case_config_vs_web(
        case_idx=0, verbose=True, refresh=True, session=None, rtol=1e-5, atol=1e-8, case_file='no_test_cases.ini', official_func=None):
    """
    this checks the config tax number vs the web number
    """
    input_struct = {}
    cases = configparser.ConfigParser()
    cases.read(case_file)
    
    for k, value in cases['%d' % case_idx].items():
        if k != 'tax':
            input_struct[k] = value

    for k, value in input_struct.items():
        if value.isdigit():
            input_struct[k] = float(value)
        elif value.replace('.', '').isdigit():
            input_struct[k] = float(value)
        elif value.replace('e', '').isdigit():
            input_struct[k] = eval(value)
    # pdb.set_trace()
    # input_struct['return_fields'] = ['all']
    input_struct['refresh'] = refresh
    official = official_func(**input_struct, session=session)

    exp = official.tax.sum()
    obs = config_tax(case_idx=case_idx, case_file=case_file)
    if verbose:
        print("Correct tax is %.1f" % exp)
        print("Config tax is %.1f" % obs)
    res_ok = np.abs(obs - exp) <= (atol + rtol * np.abs(exp))
    return res_ok



def config_tax(case_idx=0, correct_tax_field='tax', case_file='no_test_cases.ini'):
    """
    what is the 'correct' tax as per the config
    """
    # print("getting inputs from %s with case_idx = %d"%(case_file, case_idx))
    input_struct = inputs_for_case_number(case_idx, case_file=case_file, include_correct_tax=True)
    return input_struct.get(correct_tax_field, None)

def check_test_case_config_vs_calculation(
        case_idx=0, verbose=True, atol=1e-8, rtol=1e-5, return_breakdown=False, case_file='no_test_cases.ini', calc_func=None, correct_tax_field='tax'):
    """
    this checks the config tax number vs the our calculated number
    """

    input_struct = inputs_for_case_number(case_idx, case_file=case_file, include_correct_tax=True)
    
    config_tax_ = input_struct[correct_tax_field]
    del input_struct[correct_tax_field]
    calculated = calc_func(**input_struct)

    if return_breakdown:
        return calculated

    
    obs = calculated
    exp = config_tax_
    if verbose:
        print(
            "[%d]: Calculated tax = %.1f, config tax = %.1f" %
            (case_idx, obs, exp))
    res_ok = np.abs(obs - exp) <= (atol + rtol * np.abs(exp))
    return res_ok

def inputs_for_case_number(case_idx=0, case_file='no_test_cases.ini', include_correct_tax=False, correct_tax_field='tax'):
    """
    helper function
    """
    if case_idx is None:
        return {}
    input_struct = {}
    cases = configparser.ConfigParser()
    cases.read(case_file)
    # pdb.set_trace()
    # print(case_idx)  
    for k, value in cases['%d' % case_idx].items():
        if k != correct_tax_field:
            input_struct[k] = value
        elif k == correct_tax_field and include_correct_tax:
            input_struct[k] = value
   

    for k, value in input_struct.items():
        if value.isdigit():
            input_struct[k] = float(value)
        elif value.replace('.', '').isdigit():
            input_struct[k] = float(value)
        elif value.replace('e', '').isdigit():
            input_struct[k] = eval(value)

    
    return input_struct
    

def all_case_numbers(case_file='no_test_cases.ini'):
    """
    helper function
    """
    cases = configparser.ConfigParser()
    cases.read(case_file)
    return sorted(map(int, list(set(cases.keys()) - set(['DEFAULT']))))

def check_calculations_vs_config(
        verbose=False, start_at_case=None, rtol=1e-5, atol=1e-8, stop_at_failure=True, case_file='no_test_cases.ini', calc_func=None):
    """
    check that our calculations tied with the hardcoded values int he config


    example usage:
    import tax_norway as tn
    import tax_utils as tut
    tut.check_calculations_vs_config(verbose=True, start_at_case=None, rtol=1e-5, stop_at_failure=True, calc_func=tn.tax_calculation)
    """
    cases = configparser.ConfigParser()
    cases.read(case_file)
    case_numbers = sorted(map(int, list(set(cases.keys()) - set(['DEFAULT']))))
    # pdb.set_trace()
    for idx in case_numbers:
        if start_at_case is not None and idx < start_at_case:
            continue
        if stop_at_failure:
            assert check_test_case_config_vs_calculation(
                case_idx=idx, verbose=verbose, rtol=rtol, atol=atol, case_file=case_file, calc_func=calc_func), "Tax for case %s is off!" % idx
        else:
            if not check_test_case_config_vs_calculation(
                    case_idx=idx, verbose=verbose, rtol=rtol, atol=atol, case_file=case_file, calc_func=calc_func):
                print("Tax for case %s is off!" % idx)
                continue

    print("All tax figures in config tested successfully!")

def read_file_as_string(filename):
    """
    util func
    """
    with open(filename, 'r') as fil:
        return fil.read()

def syncsort(a_arr, b_arr):
    """
    sorts a in ascending order (and b will tag along, so each element of b is still associated with the right element in a)

    """
    a_arr, b_arr = (list(t) for t in zip(*sorted(zip(a_arr, b_arr))))
    return a_arr, b_arr

def get_request_from_session(session=None, url='', refresh=False, headers=None):
    """
    don't want the session object as part o fthe cache key
    """

    if __can_cache__:
        ck = cache.create_cache_key({'url':url})
        rv = cache.extract_cache_data(ck, refresh)
        if rv is not None:
            return rv
    rv = session.get(url, headers=headers)
    if __can_cache__:
        cache.set_gp_cache(ck, rv)
    return rv



def value_is_numeric_type(X):
    X = np.asanyarray(X)
    
    if (X.dtype.char in np.typecodes['AllFloat']) or (X.dtype.char in np.typecodes['Integer']):
        return True
    return False
def post_request_from_session(session=None, url='', refresh=False, headers=None, payload_=None):
    """
    don't want the session object as part o fthe cache key
    """

    if __can_cache__:
        ck = cache.create_cache_key({'url':url, 'pl':payload_})
        rv = cache.extract_cache_data(ck, refresh)
        if rv is not None:
            return rv
    rv = session.post(url, headers=headers, data=json.dumps(payload_, cls=NpEncoder))
    if __can_cache__:
        cache.set_gp_cache(ck, rv)
    return rv



<filename>tax_utils.py


__can_cache__ = True
__tabulate_avail__ = True
__plotly_avail__ = True
__config_name__ = 'taxes.ini'


import json
import os
import configparser
import requests
import numpy as np
import pandas as pd

try:
    import caching3 as cache
except ImportError as err:
    # pdb.set_trace()
    __can_cache__ = False
    print("Failed to import caching lib: %s" % err.msg)

try:
    import plotly.io as pio
except ImportError as err:
    __plotly_avail__ = True
    print("Failed to import plotting lib: %s" % err.msg)


def get_request(url='', refresh=False, headers_=None,
                method='get', payload_=None):
    """
    simple wrapper to avoid querying multiple times
    meaningless if cache library not available

    """
    # print __file__
    if __can_cache__:
        cache_key = cache.create_cache_key(
            locals(), cache.current_function_name())
        ret_val = cache.extract_cache_data(cache_key, refresh)
    else:
        ret_val = None
    # pdb.set_trace()
    if ret_val is None:
        if headers_ is None:
            if method == 'get':
                ret_val = requests.get(url)
            elif method == 'post':
                if payload_ is not None:
                    # print("got to the right place")
                    ret_val = requests.post(
                        url, data=json.dumps(payload_), headers=headers_)
                else:
                    ret_val = requests.post(url)
            else:
                raise NotImplementedError(
                    "Method == '%s' not implemented" % method)
        else:
            if method == 'get':
                ret_val = requests.get(url, headers=headers_)
            elif method == 'post':
                if payload_ is not None:
                    # print("got to the right place")
                    ret_val = requests.post(url, data=json.dumps(
                        payload_, cls=NpEncoder), headers=headers_)
                else:
                    ret_val = requests.post(url)
            else:
                raise Exception("Unknown method == '%s'" % method)

        if __can_cache__:
            cache.set_gp_cache(cache_key, ret_val)
    return ret_val
class NpEncoder(json.JSONEncoder):
    """
    Needed for when we produce the test cases (since they get created with np.int64s instead of ints, which JSONEncoder can't handle)
    """

    def default(self, obj):
        if isinstance(obj, np.integer):
            return int(obj)
        if isinstance(obj, np.floating):
            return float(obj)
        if isinstance(obj, np.ndarray):
            return obj.tolist()
        return super(NpEncoder, self).default(obj)

def tax_parameters(jurisdiction='NO', tax_year=None):
    """
    this reads in the parameters from the config and makes them easily usable from other functions
    """

    if not os.path.exists(__config_name__):
        raise Exception(
            "You need a file called '%s' with all relevant tax parameters in it!" % (__config_name__))

    if tax_year is None:
        tax_year = pd.to_datetime('today').year
    key = '%s-%d' % (jurisdiction, tax_year)
    config = configparser.ConfigParser()
    config.read(__config_name__)
    try:
        return config[key]
    except BaseException:
        raise Exception("Tax year = %d not implemented for %s jurisdiction" % (tax_year, jurisdiction))

def tax_round(in_float):
    """
    For values exactly halfway between rounded decimal values, NumPy
    rounds to the nearest even value. Thus 1.5 and 2.5 round to 2.0,
    -0.5 and 0.5 round to 0.0, etc.
    """
    if in_float == int(in_float):
        return int(in_float)

    _lo = np.floor(in_float)
    _hi = np.ceil(in_float)
    dist_to_hi = abs(in_float - _hi)
    dist_to_lo = abs(in_float - _lo)
    if dist_to_hi <= dist_to_lo:
        return int(_hi)
    # elif dist_to_lo < dist_to_hi:
    return int(_lo)
def big_fmt(val):
    """
    format with thousandsd separator
    """
    if pd.isnull(val):
        return 'n/a'

    if isinstance(val, str):
        return val

    return '{:,}'.format(int(round(val, 0))).strip()


def display_df(frame, title=None, header_size=1):
    """
    Show the dataframe in the browser
    """
    if not __plotly_avail__:
        raise Exception("You don't have plotly installed")
    raw = frame.to_html(index=False)
    if title is not None:
        raw = '<h%d>%s</h%d>%s' % (header_size, title, header_size, raw)
    pio._base_renderers.open_html_in_browser(raw)

def check_config_cases_vs_web(verbose=False, rtol=1e-5, atol=1e-8, case_file='no_test_cases.ini', official_func=None):
    """
    Mostly to catch fat fingers when punching in test cases

    example usage:


    import tax_norway as tn
    import tax_utils as tut
    tut.check_config_cases_vs_web(official_func=tn.produce_tax_results)
    """
    cases = configparser.ConfigParser()
    cases.read(case_file)
    case_numbers = sorted(map(int, list(set(cases.keys()) - set(['DEFAULT']))))
    with requests.Session() as sesh:
        for idx in case_numbers:
            assert check_test_case_config_vs_web(
                case_idx=idx, session=sesh, verbose=verbose, atol=atol, rtol=rtol, official_func=official_func), "Tax for case %s is off!" % idx

    print("All tax figures in config tested successfully!")

def check_test_case_config_vs_web(
        case_idx=0, verbose=True, refresh=True, session=None, rtol=1e-5, atol=1e-8, case_file='no_test_cases.ini', official_func=None):
    """
    this checks the config tax number vs the web number
    """
    input_struct = {}
    cases = configparser.ConfigParser()
    cases.read(case_file)
    
    for k, value in cases['%d' % case_idx].items():
        if k != 'tax':
            input_struct[k] = value

    for k, value in input_struct.items():
        if value.isdigit():
            input_struct[k] = float(value)
        elif value.replace('.', '').isdigit():
            input_struct[k] = float(value)
        elif value.replace('e', '').isdigit():
            input_struct[k] = eval(value)
    # pdb.set_trace()
    # input_struct['return_fields'] = ['all']
    input_struct['refresh'] = refresh
    official = official_func(**input_struct, session=session)

    exp = official.tax.sum()
    obs = config_tax(case_idx=case_idx, case_file=case_file)
    if verbose:
        print("Correct tax is %.1f" % exp)
        print("Config tax is %.1f" % obs)
    res_ok = np.abs(obs - exp) <= (atol + rtol * np.abs(exp))
    return res_ok



def config_tax(case_idx=0, correct_tax_field='tax', case_file='no_test_cases.ini'):
    """
    what is the 'correct' tax as per the config
    """
    # print("getting inputs from %s with case_idx = %d"%(case_file, case_idx))
    input_struct = inputs_for_case_number(case_idx, case_file=case_file, include_correct_tax=True)
    return input_struct.get(correct_tax_field, None)

def check_test_case_config_vs_calculation(
        case_idx=0, verbose=True, atol=1e-8, rtol=1e-5, return_breakdown=False, case_file='no_test_cases.ini', calc_func=None, correct_tax_field='tax'):
    """
    this checks the config tax number vs the our calculated number
    """

    input_struct = inputs_for_case_number(case_idx, case_file=case_file, include_correct_tax=True)
    
    config_tax_ = input_struct[correct_tax_field]
    del input_struct[correct_tax_field]
    calculated = calc_func(**input_struct)

    if return_breakdown:
        return calculated

    
    obs = calculated
    exp = config_tax_
    if verbose:
        print(
            "[%d]: Calculated tax = %.1f, config tax = %.1f" %
            (case_idx, obs, exp))
    res_ok = np.abs(obs - exp) <= (atol + rtol * np.abs(exp))
    return res_ok

def inputs_for_case_number(case_idx=0, case_file='no_test_cases.ini', include_correct_tax=False, correct_tax_field='tax'):
    """
    helper function
    """
    if case_idx is None:
        return {}
    input_struct = {}
    cases = configparser.ConfigParser()
    cases.read(case_file)
    # pdb.set_trace()
    # print(case_idx)  
    for k, value in cases['%d' % case_idx].items():
        if k != correct_tax_field:
            input_struct[k] = value
        elif k == correct_tax_field and include_correct_tax:
            input_struct[k] = value
   

    for k, value in input_struct.items():
        if value.isdigit():
            input_struct[k] = float(value)
        elif value.replace('.', '').isdigit():
            input_struct[k] = float(value)
        elif value.replace('e', '').isdigit():
            input_struct[k] = eval(value)

    
    return input_struct
    

def all_case_numbers(case_file='no_test_cases.ini'):
    """
    helper function
    """
    cases = configparser.ConfigParser()
    cases.read(case_file)
    return sorted(map(int, list(set(cases.keys()) - set(['DEFAULT']))))

def check_calculations_vs_config(
        verbose=False, start_at_case=None, rtol=1e-5, atol=1e-8, stop_at_failure=True, case_file='no_test_cases.ini', calc_func=None):
    """
    check that our calculations tied with the hardcoded values int he config


    example usage:
    import tax_norway as tn
    import tax_utils as tut
    tut.check_calculations_vs_config(verbose=True, start_at_case=None, rtol=1e-5, stop_at_failure=True, calc_func=tn.tax_calculation)
    """
    cases = configparser.ConfigParser()
    cases.read(case_file)
    case_numbers = sorted(map(int, list(set(cases.keys()) - set(['DEFAULT']))))
    # pdb.set_trace()
    for idx in case_numbers:
        if start_at_case is not None and idx < start_at_case:
            continue
        if stop_at_failure:
            assert check_test_case_config_vs_calculation(
                case_idx=idx, verbose=verbose, rtol=rtol, atol=atol, case_file=case_file, calc_func=calc_func), "Tax for case %s is off!" % idx
        else:
            if not check_test_case_config_vs_calculation(
                    case_idx=idx, verbose=verbose, rtol=rtol, atol=atol, case_file=case_file, calc_func=calc_func):
                print("Tax for case %s is off!" % idx)
                continue

    print("All tax figures in config tested successfully!")

def read_file_as_string(filename):
    """
    util func
    """
    with open(filename, 'r') as fil:
        return fil.read()

def syncsort(a_arr, b_arr):
    """
    sorts a in ascending order (and b will tag along, so each element of b is still associated with the right element in a)

    """
    a_arr, b_arr = (list(t) for t in zip(*sorted(zip(a_arr, b_arr))))
    return a_arr, b_arr

def get_request_from_session(session=None, url='', refresh=False, headers=None):
    """
    don't want the session object as part o fthe cache key
    """

    if __can_cache__:
        ck = cache.create_cache_key({'url':url})
        rv = cache.extract_cache_data(ck, refresh)
        if rv is not None:
            return rv
    rv = session.get(url, headers=headers)
    if __can_cache__:
        cache.set_gp_cache(ck, rv)
    return rv



def value_is_numeric_type(X):
    X = np.asanyarray(X)
    
    if (X.dtype.char in np.typecodes['AllFloat']) or (X.dtype.char in np.typecodes['Integer']):
        return True
    return False
def post_request_from_session(session=None, url='', refresh=False, headers=None, payload_=None):
    """
    don't want the session object as part o fthe cache key
    """

    if __can_cache__:
        ck = cache.create_cache_key({'url':url, 'pl':payload_})
        rv = cache.extract_cache_data(ck, refresh)
        if rv is not None:
            return rv
    rv = session.post(url, headers=headers, data=json.dumps(payload_, cls=NpEncoder))
    if __can_cache__:
        cache.set_gp_cache(ck, rv)
    return rv



# pdb.set_trace() simple wrapper to avoid querying multiple times     meaningless if cache library not available # print __file__ # pdb.set_trace() # print("got to the right place") # print("got to the right place") Needed for when we produce the test cases (since they get created with np.int64s instead of ints, which JSONEncoder can't handle) this reads in the parameters from the config and makes them easily usable from other functions For values exactly halfway between rounded decimal values, NumPy     rounds to the nearest even value. Thus 1.5 and 2.5 round to 2.0,     -0.5 and 0.5 round to 0.0, etc. # elif dist_to_lo < dist_to_hi: format with thousandsd separator Show the dataframe in the browser Mostly to catch fat fingers when punching in test cases      example usage:       import tax_norway as tn     import tax_utils as tut     tut.check_config_cases_vs_web(official_func=tn.produce_tax_results) this checks the config tax number vs the web number # pdb.set_trace() # input_struct['return_fields'] = ['all'] what is the 'correct' tax as per the config # print("getting inputs from %s with case_idx = %d"%(case_file, case_idx)) this checks the config tax number vs the our calculated number helper function # pdb.set_trace() # print(case_idx) helper function check that our calculations tied with the hardcoded values int he config       example usage:     import tax_norway as tn     import tax_utils as tut     tut.check_calculations_vs_config(verbose=True, start_at_case=None, rtol=1e-5, stop_at_failure=True, calc_func=tn.tax_calculation) # pdb.set_trace() util func sorts a in ascending order (and b will tag along, so each element of b is still associated with the right element in a) don't want the session object as part o fthe cache key don't want the session object as part o fthe cache key