Skip to content
Snippets Groups Projects
Commit b7751ef1 authored by Peter Lünenschloß's avatar Peter Lünenschloß
Browse files

harm (wrapper) testmodule finished

parent 56f1368e
No related branches found
No related tags found
3 merge requests!193Release 1.4,!188Release 1.4,!49Dataprocessing features
Hide whitespace changes
Inline Side-by-side
test/funcs/test_harm_funcs.py
+ 38
− 57
View file @ b7751ef1
......@@ -49,33 +49,6 @@ def data():
return data
@pytest.fixture
def multi_data():
index = pd.date_range(start="1.1.2011 00:00:00", end="1.1.2011 01:00:00", freq="15min")
index = index.insert(2, pd.Timestamp(2011, 1, 1, 0, 29, 0))
index = index.insert(2, pd.Timestamp(2011, 1, 1, 0, 28, 0))
index = index.insert(5, pd.Timestamp(2011, 1, 1, 0, 32, 0))
index = index.insert(5, pd.Timestamp(2011, 1, 1, 0, 31, 0))
index = index.insert(0, pd.Timestamp(2010, 12, 31, 23, 57, 0))
index = index.drop(pd.Timestamp("2011-01-01 00:30:00"))
dat = pd.Series(np.linspace(-50, 50, index.size), index=index, name="data")
# good to have some nan
dat[-3] = np.nan
data = dat.to_frame()
data.index = data.index.shift(1, "2min")
dat2 = data.copy()
dat2.index = dat2.index.shift(1, "17min")
dat2.rename(columns={"data": "data2"}, inplace=True)
dat3 = data.copy()
dat3.index = dat3.index.shift(1, "1h")
dat3.rename(columns={"data": "data3"}, inplace=True)
dat3.drop(dat3.index[2:-2], inplace=True)
# merge
data = pd.merge(data, dat2, how="outer", left_index=True, right_index=True)
data = pd.merge(data, dat3, how="outer", left_index=True, right_index=True)
return dios.DictOfSeries(data)
@pytest.mark.parametrize("flagger", TESTFLAGGER)
@pytest.mark.parametrize("reshaper", RESHAPERS)
@pytest.mark.parametrize("co_flagging", COFLAGGING)
......@@ -124,9 +97,9 @@ def test_harmSingleVarIntermediateFlagging(data, flagger, reshaper, co_flagging)
@pytest.mark.parametrize("flagger", TESTFLAGGER)
def test_harmSingleVarInterpolations(data, flagger):
flagger = flagger.initFlags(data)
field = data.columns[0]
pre_data = data[field]
pre_flags = flagger.getFlags(field)
tests = [
("nagg", "15Min", pd.Series(data=[-87.5, -25.0, 0.0, 37.5, 50.0],
index=pd.date_range('2011-01-01 00:00:00',
......@@ -147,36 +120,44 @@ def test_harmSingleVarInterpolations(data, flagger):
]
for interpolation, freq, expected in tests:
data_harm, _ = harm_aggregate2Grid(data, field, flagger, freq, value_func=np.sum, method=interpolation)
data_harm, flagger_harm = harm_aggregate2Grid(data, field, flagger, freq, value_func=np.sum, method=interpolation)
assert data_harm[field].equals(expected)
data_deharm, flagger_deharm = harm_deharm(data_harm, "data", flagger_harm, method="inverse_" + interpolation)
assert data_deharm[field + ORIGINAL_SUFFIX].equals(pre_data)
assert flagger_deharm.getFlags([field + ORIGINAL_SUFFIX]).equals(pre_flags)
import pdb
pdb.set_trace()
#tests = [
# ("fshift", "15Min", [np.nan, -37.5, -25.0, 0.0, 37.5, 50.0]),
# ("fshift", "30Min", [np.nan, -37.5, 0.0, 50.0]),
# ("bshift", "15Min", [-50.0, -37.5, -25.0, 12.5, 37.5, 50.0]),
# ("bshift", "30Min", [-50.0, -37.5, 12.5, 50.0]),
# ("nshift", "15min", [np.nan, -37.5, -25.0, 12.5, 37.5, 50.0]),
# ("nshift", "30min", [np.nan, -37.5, 12.5, 50.0])]
#for interpolation, freq, expected in tests:
# data_harm, _ = harm_shift2Grid(data, field, flagger, freq, method=interpolation)
# harm_start = data[field].index[0].floor(freq=freq)
# harm_end = data[field].index[-1].ceil(freq=freq)
# test_index = pd.date_range(start=harm_start, end=harm_end, freq=freq)
# expected = pd.Series(expected, index=test_index)
# assert data_harm[field].equals(expected)
#data_deharm, flagger_deharm = harm_deharmonize(data, "data", flagger, co_flagging=True)
#flags = flagger.getFlags()
#flags_deharm = flagger_deharm.getFlags()
#assert data[field].equals(data[field])
#assert len(data_deharm[field]) == len(flags[field])
#assert (flags[field].index == flags_deharm[field].index).all()
tests = [
("fshift", "15Min", pd.Series(data=[np.nan, -37.5, -25.0, 0.0, 37.5, 50.0],
index=pd.date_range('2010-12-31 23:45:00',
'2011-01-01 01:00:00',
freq='15Min'))),
("fshift", "30Min", pd.Series(data=[np.nan, -37.5, 0.0, 50.0],
index=pd.date_range('2010-12-31 23:30:00',
'2011-01-01 01:00:00',
freq='30Min'))),
("bshift", "15Min", pd.Series(data=[-50.0, -37.5, -25.0, 12.5, 37.5, 50.0],
index=pd.date_range('2010-12-31 23:45:00',
'2011-01-01 01:00:00',
freq='15Min'))),
("bshift", "30Min", pd.Series(data=[-50.0, -37.5, 12.5, 50.0],
index=pd.date_range('2010-12-31 23:30:00',
'2011-01-01 01:00:00',
freq='30Min'))),
("nshift", "15min", pd.Series(data=[np.nan, -37.5, -25.0, 12.5, 37.5, 50.0],
index=pd.date_range('2010-12-31 23:45:00',
'2011-01-01 01:00:00',
freq='15Min'))),
("nshift", "30min", pd.Series(data=[np.nan, -37.5, 12.5, 50.0],
index=pd.date_range('2010-12-31 23:30:00',
'2011-01-01 01:00:00',
freq='30Min')))]
for interpolation, freq, expected in tests:
data_harm, flagger_harm = harm_shift2Grid(data, field, flagger, freq, method=interpolation)
assert data_harm[field].equals(expected)
data_deharm, flagger_deharm = harm_deharm(data_harm, "data", flagger_harm, method="inverse_" + interpolation)
assert data_deharm[field + ORIGINAL_SUFFIX].equals(pre_data)
assert flagger_deharm.getFlags([field + ORIGINAL_SUFFIX]).equals(pre_flags)
@pytest.mark.parametrize("method", INTERPOLATIONS2)
......
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