From 8adf6744e8a501ac6c0127f5af98d07153a15d5b Mon Sep 17 00:00:00 2001
From: Bert Palm <bert.palm@ufz.de>
Date: Sat, 20 Mar 2021 02:59:02 +0100
Subject: [PATCH] fixed other processing tests
---
saqc/flagger/flags.py | 4 ++--
saqc/funcs/drift.py | 3 ++-
tests/funcs/test_modelling.py | 29 +++++++++++--------------
tests/funcs/test_proc_functions.py | 34 ++++++++++++------------------
4 files changed, 30 insertions(+), 40 deletions(-)
diff --git a/saqc/flagger/flags.py b/saqc/flagger/flags.py
index e6ef0ad5b..e6d3076fa 100644
--- a/saqc/flagger/flags.py
+++ b/saqc/flagger/flags.py
@@ -401,8 +401,8 @@ def mergeHistoryByFunc(flags: Flags, field, source, merge_func, merge_func_kws,
target_history = flags.history[field]
source_history = flags.history[source]
new_target_history = History()
- import pdb
- pdb.set_trace()
+ # import pdb
+ # pdb.set_trace()
for k in target_history.hist.columns:
col_args_h = dict(source_col=source_history.hist[k])
col_args_m = dict(source_col=source_history.mask[k])
diff --git a/saqc/funcs/drift.py b/saqc/funcs/drift.py
index dc2265e8a..ea1a4d20c 100644
--- a/saqc/funcs/drift.py
+++ b/saqc/funcs/drift.py
@@ -454,7 +454,8 @@ def correctRegimeAnomaly(
cluster_field: ColumnName,
model: CurveFitter,
regime_transmission: Optional[FreqString]=None,
- x_date: bool=False
+ x_date: bool=False,
+ **kwargs
) -> Tuple[DictOfSeries, Flagger]:
"""
Function fits the passed model to the different regimes in data[field] and tries to correct
diff --git a/tests/funcs/test_modelling.py b/tests/funcs/test_modelling.py
index 6d99d5786..23cc82ab2 100644
--- a/tests/funcs/test_modelling.py
+++ b/tests/funcs/test_modelling.py
@@ -6,23 +6,20 @@
import dios
+from saqc.flagger import initFlagsLike
from saqc.funcs.tools import mask
from saqc.funcs.residues import calculatePolynomialResidues, calculateRollingResidues
from tests.fixtures import *
-from tests.common import TESTFLAGGER
-TF = TESTFLAGGER[:1]
-
-@pytest.mark.parametrize("flagger", TF)
@pytest.mark.parametrize("dat", [pytest.lazy_fixture("course_2")])
-def test_modelling_polyFit_forRegular(dat, flagger):
+def test_modelling_polyFit_forRegular(dat):
data, _ = dat(freq="10min", periods=30, initial_level=0, final_level=100, out_val=-100)
# add some nice sine distortion
data = data + 10 * np.sin(np.arange(0, len(data.indexes[0])))
data = dios.DictOfSeries(data)
- flagger = flagger.initFlags(data)
+ flagger = initFlagsLike(data)
result1, _ = calculatePolynomialResidues(data, "data", flagger, 11, 2, numba=False)
result2, _ = calculatePolynomialResidues(data, "data", flagger, 11, 2, numba=True)
assert (result1["data"] - result2["data"]).abs().max() < 10 ** -10
@@ -35,39 +32,37 @@ def test_modelling_polyFit_forRegular(dat, flagger):
assert result5["data"].iloc[10:19].isna().all()
-@pytest.mark.parametrize("flagger", TF)
@pytest.mark.parametrize("dat", [pytest.lazy_fixture("course_2")])
-def test_modelling_rollingMean_forRegular(dat, flagger):
+def test_modelling_rollingMean_forRegular(dat):
data, _ = dat(freq="10min", periods=30, initial_level=0, final_level=100, out_val=-100)
data = dios.DictOfSeries(data)
- flagger = flagger.initFlags(data)
+ flagger = initFlagsLike(data)
calculateRollingResidues(data, "data", flagger, 5, func=np.mean, eval_flags=True, min_periods=0, center=True)
calculateRollingResidues(data, "data", flagger, 5, func=np.mean, eval_flags=True, min_periods=0, center=False)
-@pytest.mark.parametrize("flagger", TF)
@pytest.mark.parametrize("dat", [pytest.lazy_fixture("course_1")])
-def test_modelling_mask(dat, flagger):
+def test_modelling_mask(dat):
data, _ = dat()
data = dios.DictOfSeries(data)
- flagger = flagger.initFlags(data)
+ flagger = initFlagsLike(data)
data_seasonal, flagger_seasonal = mask(data, "data", flagger, mode='periodic', period_start="20:00",
period_end="40:00", include_bounds=False)
- flaggs = flagger_seasonal._flags["data"]
+ flaggs = flagger_seasonal["data"]
assert flaggs[np.logical_and(20 <= flaggs.index.minute, 40 >= flaggs.index.minute)].isna().all()
data_seasonal, flagger_seasonal = mask(data, "data", flagger, mode='periodic', period_start="15:00:00",
period_end="02:00:00")
- flaggs = flagger_seasonal._flags["data"]
+ flaggs = flagger_seasonal["data"]
assert flaggs[np.logical_and(15 <= flaggs.index.hour, 2 >= flaggs.index.hour)].isna().all()
data_seasonal, flagger_seasonal = mask(data, "data", flagger, mode='periodic', period_start="03T00:00:00",
period_end="10T00:00:00")
- flaggs = flagger_seasonal._flags["data"]
+ flaggs = flagger_seasonal["data"]
assert flaggs[np.logical_and(3 <= flaggs.index.hour, 10 >= flaggs.index.hour)].isna().all()
mask_ser = pd.Series(False, index=data["data"].index)
mask_ser[::5] = True
data["mask_ser"] = mask_ser
- flagger = flagger.initFlags(data)
+ flagger = initFlagsLike(data)
data_masked, flagger_masked = mask(data, "data", flagger, mode='mask_var', mask_var="mask_ser")
- flaggs = flagger_masked._flags["data"]
+ flaggs = flagger_masked["data"]
assert flaggs[data_masked['mask_ser']].isna().all()
diff --git a/tests/funcs/test_proc_functions.py b/tests/funcs/test_proc_functions.py
index d7cada078..d9d137359 100644
--- a/tests/funcs/test_proc_functions.py
+++ b/tests/funcs/test_proc_functions.py
@@ -7,6 +7,7 @@
import dios
from saqc.constants import *
+from saqc.flagger import initFlagsLike
from saqc.funcs.transformation import transform
from saqc.funcs.drift import correctOffset
from saqc.funcs.interpolation import interpolateByRolling, interpolateInvalid, interpolateIndex
@@ -14,15 +15,13 @@ from saqc.funcs.resampling import resample
from saqc.lib.ts_operators import linearInterpolation, polynomialInterpolation
from tests.fixtures import *
-from tests.common import TESTFLAGGER
-@pytest.mark.parametrize("flagger", TESTFLAGGER)
-def test_rollingInterpolateMissing(course_5, flagger):
+def test_rollingInterpolateMissing(course_5):
data, characteristics = course_5(periods=10, nan_slice=[5, 6])
field = data.columns[0]
data = dios.DictOfSeries(data)
- flagger = flagger.initFlags(data)
+ flagger = initFlagsLike(data)
dataInt, *_ = interpolateByRolling(
data, field, flagger, 3, func=np.median, center=True, min_periods=0, interpol_flag=UNFLAGGED
)
@@ -35,12 +34,11 @@ def test_rollingInterpolateMissing(course_5, flagger):
assert dataInt[field][characteristics["missing"]].isna().all()
-@pytest.mark.parametrize("flagger", TESTFLAGGER)
-def test_interpolateMissing(course_5, flagger):
+def test_interpolateMissing(course_5):
data, characteristics = course_5(periods=10, nan_slice=[5])
field = data.columns[0]
data = dios.DictOfSeries(data)
- flagger = flagger.initFlags(data)
+ flagger = initFlagsLike(data)
dataLin, *_ = interpolateInvalid(data, field, flagger, method="linear")
dataPoly, *_ = interpolateInvalid(data, field, flagger, method="polynomial")
assert dataLin[field][characteristics["missing"]].notna().all()
@@ -54,12 +52,11 @@ def test_interpolateMissing(course_5, flagger):
assert dataLin3[field][characteristics["missing"]].notna().all()
-@pytest.mark.parametrize("flagger", TESTFLAGGER)
-def test_transform(course_5, flagger):
+def test_transform(course_5):
data, characteristics = course_5(periods=10, nan_slice=[5, 6])
field = data.columns[0]
data = dios.DictOfSeries(data)
- flagger = flagger.initFlags(data)
+ flagger = initFlagsLike(data)
data1, *_ = transform(data, field, flagger, func=linearInterpolation)
assert data1[field][characteristics["missing"]].isna().all()
data1, *_ = transform(data, field, flagger, func=lambda x: linearInterpolation(x, inter_limit=3))
@@ -70,35 +67,32 @@ def test_transform(course_5, flagger):
assert data1[field][characteristics["missing"]].notna().all()
-@pytest.mark.parametrize("flagger", TESTFLAGGER)
-def test_resample(course_5, flagger):
+def test_resample(course_5):
data, characteristics = course_5(freq="1min", periods=30, nan_slice=[1, 11, 12, 22, 24, 26])
field = data.columns[0]
data = dios.DictOfSeries(data)
- flagger = flagger.initFlags(data)
+ flagger = initFlagsLike(data)
data1, *_ = resample(data, field, flagger, "10min", np.mean, max_invalid_total_d=2, max_invalid_consec_d=1)
assert ~np.isnan(data1[field].iloc[0])
assert np.isnan(data1[field].iloc[1])
assert np.isnan(data1[field].iloc[2])
-@pytest.mark.parametrize("flagger", TESTFLAGGER)
-def test_interpolateGrid(course_5, course_3, flagger):
+def test_interpolateGrid(course_5, course_3):
data, _ = course_5()
data_grid, characteristics = course_3()
data['grid'] = data_grid.to_df()
# data = dios.DictOfSeries(data)
- flagger = flagger.initFlags(data)
+ flagger = initFlagsLike(data)
dataInt, *_ = interpolateIndex(data, 'data', flagger, '1h', 'time', grid_field='grid', inter_limit=10)
-@pytest.mark.parametrize("flagger", TESTFLAGGER)
-def test_offsetCorrecture(flagger):
+def test_offsetCorrecture():
data = pd.Series(0, index=pd.date_range('2000', freq='1d', periods=100), name='dat')
data.iloc[30:40] = -100
data.iloc[70:80] = 100
data = dios.DictOfSeries(data)
- flagger = flagger.initFlags(data)
- data, flagger = correctOffset(data, 'dat', flagger, 40, 20, '3d', 1)
+ flagger = initFlagsLike(data)
+ data, _ = correctOffset(data, 'dat', flagger, 40, 20, '3d', 1)
assert (data == 0).all()[0]
--
GitLab