Skip to content
Snippets Groups Projects
Commit a0147299 authored by Bert Palm's avatar Bert Palm 🎇
Browse files

fixed harmo tests

parent 88fab5d3
No related branches found
No related tags found
3 merge requests!271Static expansion of regular expressions,!260Follow-Up Translations,!237Flagger Translations
Hide whitespace changes
Inline Side-by-side
tests/funcs/test_harm_funcs.py
+ 103
− 170
View file @ a0147299
#! /usr/bin/env python #! /usr/bin/env python
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
# see test/functs/conftest.py for global fixtures "course_..."
import pytest import pytest
import numpy as np import numpy as np
import pandas as pd import pandas as pd
import dios import dios
from test.common import TESTFLAGGER
from saqc.flagger import Flagger, initFlagsLike from saqc.flagger import Flagger, initFlagsLike
from saqc.common import BAD from saqc.constants import BAD, UNFLAGGED
from saqc.funcs.resampling import ( from saqc.funcs.resampling import (
linear, linear,
interpolate, interpolate,
...@@ -20,10 +16,6 @@ from saqc.funcs.resampling import ( ...@@ -20,10 +16,6 @@ from saqc.funcs.resampling import (
mapToOriginal, mapToOriginal,
) )
RESHAPERS = ["nshift", "fshift", "bshift", "nagg", "bagg", "fagg", "interpolation"]
INTERPOLATIONS = ["time", "polynomial"]
@pytest.fixture @pytest.fixture
def data(): def data():
...@@ -41,184 +33,125 @@ def data(): ...@@ -41,184 +33,125 @@ def data():
return data return data
@pytest.mark.parametrize("flagger", TESTFLAGGER) @pytest.mark.parametrize("reshaper", ["nshift", "fshift", "bshift", "nagg", "bagg", "fagg", "interpolation"])
@pytest.mark.parametrize("reshaper", RESHAPERS) def test_harmSingleVarIntermediateFlagging(data, reshaper):
def test_harmSingleVarIntermediateFlagging(data, flagger, reshaper):
flagger = initFlagsLike(data) flagger = initFlagsLike(data)
# make pre harm copies: field = 'data'
pre_data = data.copy() pre_data = data.copy()
pre_flags = flagger['data'] pre_flagger = flagger.copy()
freq = "15min"
assert len(data.columns) == 1 data, flagger = linear(data, field, flagger, freq="15min")
field = data.columns[0]
data, flagger = linear(data, "data", flagger, freq)
# flag something bad # flag something bad
f_ser = pd.Series(data=[-np.inf] * len(data[field]), index=data[field].index) flagger[data[field].index[3:4], field] = BAD
f_ser[3:4] = BAD data, flagger = mapToOriginal(data, field, flagger, method="inverse_" + reshaper)
flagger[field] = f_ser
data, flagger = mapToOriginal(data, "data", flagger, method="inverse_" + reshaper) assert len(data[field]) == len(flagger[field])
d = data[field] assert data[field].equals(pre_data[field])
if reshaper == "nagg": assert flagger[field].index.equals(pre_flagger[field].index)
assert flagger.isFlagged(loc=d.index[3:7]).squeeze().all()
assert (~flagger.isFlagged(loc=d.index[0:3]).squeeze()).all() if 'agg' in reshaper:
assert (~flagger.isFlagged(loc=d.index[7:]).squeeze()).all() if reshaper == "nagg":
if reshaper == "nshift": start, end = 3, 7
assert (flagger.isFlagged().squeeze() == [False, False, False, False, True, False, False, False, False]).all() elif reshaper == "fagg":
if reshaper == "bagg": start, end = 3, 5
assert flagger.isFlagged(loc=d.index[5:7]).squeeze().all() elif reshaper == "bagg":
assert (~flagger.isFlagged(loc=d.index[0:5]).squeeze()).all() start, end = 5, 7
assert (~flagger.isFlagged(loc=d.index[7:]).squeeze()).all() else:
if reshaper == "bshift": raise NotImplementedError('untested test case')
assert (flagger.isFlagged().squeeze() == [False, False, False, False, False, True, False, False, False]).all()
if reshaper == "fagg": assert all(flagger[field].iloc[start:end])
assert flagger.isFlagged(loc=d.index[3:5]).squeeze().all() assert all(~flagger[field].iloc[:start])
assert (~flagger.isFlagged(loc=d.index[0:3]).squeeze()).all() assert all(~flagger[field].iloc[end:])
assert (~flagger.isFlagged(loc=d.index[5:]).squeeze()).all()
if reshaper == "fshift": elif 'shift' in reshaper:
assert (flagger.isFlagged().squeeze() == [False, False, False, False, True, False, False, False, False]).all() if reshaper == "nshift":
exp = [False, False, False, False, True, False, False, False, False]
flags = flagger.getFlags() elif reshaper == "fshift":
assert pre_data[field].equals(data[field]) exp = [False, False, False, False, True, False, False, False, False]
assert len(data[field]) == len(flags[field]) elif reshaper == "bshift":
assert (pre_flags[field].index == flags[field].index).all() exp = [False, False, False, False, False, True, False, False, False]
else:
raise NotImplementedError('untested test case')
@pytest.mark.parametrize("flagger", TESTFLAGGER)
def test_harmSingleVarInterpolations(data, flagger): flagged = flagger[field] > UNFLAGGED
flagger = flagger.initFlags(data) assert all(flagged == exp)
field = data.columns[0]
pre_data = data[field] else:
pre_flags = flagger.getFlags(field) raise NotImplementedError('untested test case')
tests = [
(
"nagg", @pytest.mark.parametrize(
"15Min", 'params, expected',
pd.Series( [
data=[-87.5, -25.0, 0.0, 37.5, 50.0], (("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", "2011-01-01 01:00:00", freq="15min"))),
index=pd.date_range("2011-01-01 00:00:00", "2011-01-01 01:00:00", freq="15min"), (("nagg", "30Min"), pd.Series(data=[-87.5, -25.0, 87.5], index=pd.date_range("2011-01-01 00:00:00", "2011-01-01 01:00:00", freq="30min"))),
), (("bagg", "15Min"), pd.Series(data=[-50.0, -37.5, -37.5, 12.5, 37.5, 50.0], index=pd.date_range("2010-12-31 23:45:00", "2011-01-01 01:00:00", freq="15min"))),
), (("bagg", "30Min"), pd.Series(data=[-50.0, -75.0, 50.0, 50.0], index=pd.date_range("2010-12-31 23:30:00", "2011-01-01 01:00:00", freq="30min"))),
( ])
"nagg", def test_harmSingleVarInterpolationAgg(data, params, expected):
"30Min", flagger = initFlagsLike(data)
pd.Series( field = 'data'
data=[-87.5, -25.0, 87.5], pre_data = data.copy()
index=pd.date_range("2011-01-01 00:00:00", "2011-01-01 01:00:00", freq="30min"), pre_flaggger = flagger.copy()
), method, freq = params
),
( data_harm, flagger_harm = aggregate(data, field, flagger, freq, value_func=np.sum, method=method)
"bagg", assert data_harm[field].equals(expected)
"15Min",
pd.Series( data_deharm, flagger_deharm = mapToOriginal(data_harm, "data", flagger_harm, method="inverse_" + method)
data=[-50.0, -37.5, -37.5, 12.5, 37.5, 50.0], assert data_deharm[field].equals(pre_data[field])
index=pd.date_range("2010-12-31 23:45:00", "2011-01-01 01:00:00", freq="15min"), assert flagger_deharm[field].equals(pre_flaggger[field])
),
),
( @pytest.mark.parametrize(
"bagg", 'params, expected',
"30Min", [
pd.Series( (("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"))),
data=[-50.0, -75.0, 50.0, 50.0], (("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"))),
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: def test_harmSingleVarInterpolationShift(data, params, expected):
data_harm, flagger_harm = aggregate( flagger = initFlagsLike(data)
data, field, flagger, freq, value_func=np.sum, method=interpolation field = 'data'
) pre_data = data.copy()
assert data_harm[field].equals(expected) pre_flagger = flagger.copy()
data_deharm, flagger_deharm = mapToOriginal( method, freq = params
data_harm, "data", flagger_harm, method="inverse_" + interpolation
) data_harm, flagger_harm = shift(data, field, flagger, freq, method=method)
assert data_deharm[field].equals(pre_data) assert data_harm[field].equals(expected)
assert flagger_deharm.getFlags([field]).squeeze().equals(pre_flags)
data_deharm, flagger_deharm = mapToOriginal(data_harm, "data", flagger_harm, method="inverse_" + method)
tests = [ assert data_deharm[field].equals(pre_data[field])
( assert flagger_deharm[field].equals(pre_flagger[field])
"fshift",
"15Min",
pd.Series( @pytest.mark.parametrize("method", ["time", "polynomial"])
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 = shift(data, field, flagger, freq, method=interpolation)
assert data_harm[field].equals(expected)
data_deharm, flagger_deharm = mapToOriginal(
data_harm, "data", flagger_harm, method="inverse_" + interpolation
)
assert data_deharm[field].equals(pre_data)
assert flagger_deharm.getFlags([field]).squeeze().equals(pre_flags)
@pytest.mark.parametrize("method", INTERPOLATIONS)
def test_gridInterpolation(data, method): def test_gridInterpolation(data, method):
freq = "15min" freq = "15min"
data = data.squeeze() field = 'data'
field = data.name data = data[field]
data = (data * np.sin(data)).append(data.shift(1, "2h")).shift(1, "3s") data = (data * np.sin(data)).append(data.shift(1, "2h")).shift(1, "3s")
data = dios.DictOfSeries(data) data = dios.DictOfSeries(data)
flagger = TESTFLAGGER[0].initFlags(data) flagger = initFlagsLike(data)
# we are just testing if the interpolation gets passed to the series without causing an error: # we are just testing if the interpolation gets passed to the series without causing an error:
interpolate(data, field, flagger, freq, method=method, downcast_interpolation=True) interpolate(data, field, flagger, freq, method=method, downcast_interpolation=True)
if method == "polynomial": if method == "polynomial":
interpolate(data, field, flagger, freq, order=2, method=method, downcast_interpolation=True) interpolate(data, field, flagger, freq, order=2, method=method, downcast_interpolation=True)
interpolate(data, field, flagger, freq, order=10, method=method, downcast_interpolation=True) interpolate(data, field, flagger, freq, order=10, method=method, downcast_interpolation=True)
@pytest.mark.parametrize("flagger", TESTFLAGGER) def test_wrapper(data):
def test_wrapper(data, flagger):
# we are only testing, whether the wrappers do pass processing: # we are only testing, whether the wrappers do pass processing:
field = data.columns[0] field = 'data'
freq = "15min" freq = "15min"
flagger = flagger.initFlags(data) flagger = initFlagsLike(data)
linear(data, field, flagger, freq, to_drop=None) linear(data, field, flagger, freq, to_drop=None)
aggregate(data, field, flagger, freq, value_func=np.nansum, method="nagg", to_drop=None) aggregate(data, field, flagger, freq, value_func=np.nansum, method="nagg", to_drop=None)
......
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