# AUTOGENERATED! DO NOT EDIT! File to edit: nbs/00_tests/blocks/tst.preprocessing.ipynb (unless otherwise specified).

__all__ = ['test_one_hot_encoder', 'generate_input_for_window_generator', 'test_window_generator',
           'test_window_aggregator']

# Cell
import pytest
import pandas as pd
import os
import joblib
from IPython.display import display

from dsblocks.blocks.preprocessing import *

# Cell
#<EMAIL>.fixture (name='example_people_data')
#def example_people_data_fixture():
#    return example_people_data()

# Comes from preprocessing.ipynb, cell
<EMAIL>
def test_one_hot_encoder ():
    df = pd.DataFrame({'x1':['b','b','a','b','a'],
                   'x2':['e','c','d','e','f'],
                  })
    one_hot_encoder = OneHotEncoder(categories=[['a','b'],['c','d','e']])
    dfr = one_hot_encoder.transform (df)

    # show result
    print ('result:')
    display (dfr)

    # compare against pd.get_dummies, removing last column corresponding to category 'f'
    df_dummies = pd.get_dummies (df).iloc[:,:-1]
    assert (dfr==df_dummies).all().all()

    # test categories='auto'
    one_hot_encoder = OneHotEncoder(categories='auto')

    # 1.- if categories='auto', we must call fit  before calling transform:
    with pytest.raises(Exception):
        dfr = one_hot_encoder.transform (df)

    # test that 'auto' returns the same as pd.get_dummies
    dfr = one_hot_encoder.fit_transform (df)
    print ('result:'); display(dfr)
    assert (dfr==pd.get_dummies (df)).all().all()

# Comes from preprocessing.ipynb, cell
def generate_input_for_window_generator ():
    index = pd.period_range (start='2022-01-14', end='2022-01-18', freq='H')
    df = pd.DataFrame (index=index.to_timestamp (), columns=['a','b','label'])
    df.index.name = 'timestamp'
    df['a'] = 10
    df['b'] = 20
    df['label'] = 0
    df.loc[('2022-01-15' <= index) & (index < '2022-01-16'),'label'] = 1
    df.loc[('2022-01-15' <= index) & (index < '2022-01-16'),'a'] = 11
    df.loc[('2022-01-15' <= index) & (index < '2022-01-16'),'b'] = 21
    return df

# Comes from preprocessing.ipynb, cell
<EMAIL>
def test_window_generator ():
    df = generate_input_for_window_generator ()
    df.plot()

    window_generator = WindowGenerator(sequence_length=3, sequence_stride=1, sampling_interval=60)
    window_data = window_generator.transform(df)
    display(window_data.head())

    assert window_data.shape == (91, 7)

    assert list(window_data.columns)==[('a', 0), ('a', 1), ('a', 2), ('b', 0), ('b', 1), ('b', 2), ('label', '')]

    assert (window_data.loc['2022-01-14 01:00:00'].values==[10, 10, 10, 20, 20, 20,  0]).all()

    df_plot = window_data.loc['2022-01-15 00:00:00':'2022-01-15 02:00:00']
    window_generator.plot (df_plot,['a','b'])
    #display ('new plot')
    #window_generator.plot (df_plot, title='new window', lw=3)

    windows_of_interest = window_data.loc['2022-01-14 00:00:00':'2022-01-16 00:00:00', ['a','b']]
    df_d = window_generator.describe(windows_of_interest)

    display(df_d)

    assert 10.488 < df_d.loc['mean','a'] < 10.49
    assert 0.5017 < df_d.loc['std','b'] < 0.5018

# Comes from preprocessing.ipynb, cell
<EMAIL>
def test_window_aggregator ():
    df = generate_input_for_window_generator ()
    window_generator = WindowGenerator(sequence_length=3, sequence_stride=1, sampling_interval=60)
    window_data = window_generator.transform(df)
    window_aggregator = WindowAggregator (sequence_length=3, sequence_stride=1, sampling_interval=60)
    df2 = window_aggregator.transform (window_data)
    assert (df==df2).all().all() and (df.index==df2.index).all()
# AUTOGENERATED! DO NOT EDIT! File to edit: nbs/00_tests/blocks/tst.preprocessing.ipynb (unless otherwise specified).

__all__ = ['test_one_hot_encoder', 'generate_input_for_window_generator', 'test_window_generator',
           'test_window_aggregator']

# Cell
import pytest
import pandas as pd
import os
import joblib
from IPython.display import display

from dsblocks.blocks.preprocessing import *

# Cell
#<EMAIL>.fixture (name='example_people_data')
#def example_people_data_fixture():
#    return example_people_data()

# Comes from preprocessing.ipynb, cell
<EMAIL>
def test_one_hot_encoder ():
    df = pd.DataFrame({'x1':['b','b','a','b','a'],
                   'x2':['e','c','d','e','f'],
                  })
    one_hot_encoder = OneHotEncoder(categories=[['a','b'],['c','d','e']])
    dfr = one_hot_encoder.transform (df)

    # show result
    print ('result:')
    display (dfr)

    # compare against pd.get_dummies, removing last column corresponding to category 'f'
    df_dummies = pd.get_dummies (df).iloc[:,:-1]
    assert (dfr==df_dummies).all().all()

    # test categories='auto'
    one_hot_encoder = OneHotEncoder(categories='auto')

    # 1.- if categories='auto', we must call fit  before calling transform:
    with pytest.raises(Exception):
        dfr = one_hot_encoder.transform (df)

    # test that 'auto' returns the same as pd.get_dummies
    dfr = one_hot_encoder.fit_transform (df)
    print ('result:'); display(dfr)
    assert (dfr==pd.get_dummies (df)).all().all()

# Comes from preprocessing.ipynb, cell
def generate_input_for_window_generator ():
    index = pd.period_range (start='2022-01-14', end='2022-01-18', freq='H')
    df = pd.DataFrame (index=index.to_timestamp (), columns=['a','b','label'])
    df.index.name = 'timestamp'
    df['a'] = 10
    df['b'] = 20
    df['label'] = 0
    df.loc[('2022-01-15' <= index) & (index < '2022-01-16'),'label'] = 1
    df.loc[('2022-01-15' <= index) & (index < '2022-01-16'),'a'] = 11
    df.loc[('2022-01-15' <= index) & (index < '2022-01-16'),'b'] = 21
    return df

# Comes from preprocessing.ipynb, cell
<EMAIL>
def test_window_generator ():
    df = generate_input_for_window_generator ()
    df.plot()

    window_generator = WindowGenerator(sequence_length=3, sequence_stride=1, sampling_interval=60)
    window_data = window_generator.transform(df)
    display(window_data.head())

    assert window_data.shape == (91, 7)

    assert list(window_data.columns)==[('a', 0), ('a', 1), ('a', 2), ('b', 0), ('b', 1), ('b', 2), ('label', '')]

    assert (window_data.loc['2022-01-14 01:00:00'].values==[10, 10, 10, 20, 20, 20,  0]).all()

    df_plot = window_data.loc['2022-01-15 00:00:00':'2022-01-15 02:00:00']
    window_generator.plot (df_plot,['a','b'])
    #display ('new plot')
    #window_generator.plot (df_plot, title='new window', lw=3)

    windows_of_interest = window_data.loc['2022-01-14 00:00:00':'2022-01-16 00:00:00', ['a','b']]
    df_d = window_generator.describe(windows_of_interest)

    display(df_d)

    assert 10.488 < df_d.loc['mean','a'] < 10.49
    assert 0.5017 < df_d.loc['std','b'] < 0.5018

# Comes from preprocessing.ipynb, cell
<EMAIL>
def test_window_aggregator ():
    df = generate_input_for_window_generator ()
    window_generator = WindowGenerator(sequence_length=3, sequence_stride=1, sampling_interval=60)
    window_data = window_generator.transform(df)
    window_aggregator = WindowAggregator (sequence_length=3, sequence_stride=1, sampling_interval=60)
    df2 = window_aggregator.transform (window_data)
    assert (df==df2).all().all() and (df.index==df2.index).all()
# AUTOGENERATED! DO NOT EDIT! File to edit: nbs/00_tests/blocks/tst.preprocessing.ipynb (unless otherwise specified). # Cell # Cell #<EMAIL>.fixture (name='example_people_data') #def example_people_data_fixture(): #    return example_people_data() # Comes from preprocessing.ipynb, cell # show result # compare against pd.get_dummies, removing last column corresponding to category 'f' # test categories='auto' # 1.- if categories='auto', we must call fit  before calling transform: # test that 'auto' returns the same as pd.get_dummies # Comes from preprocessing.ipynb, cell # Comes from preprocessing.ipynb, cell #display ('new plot') #window_generator.plot (df_plot, title='new window', lw=3) # Comes from preprocessing.ipynb, cell