From e02c8d846d21192b2fc3cf65c1d909d41b4f3e26 Mon Sep 17 00:00:00 2001
From: Bert Palm <bert.palm@ufz.de>
Date: Sat, 20 Mar 2021 15:13:04 +0100
Subject: [PATCH] cleanup, and to_mask-fix
---
saqc/core/register.py | 4 +--
saqc/funcs/interpolation.py | 53 ++++++++++++++++------------------
saqc/funcs/resampling.py | 7 ++---
saqc/lib/ts_operators.py | 2 +-
tests/funcs/test_harm_funcs.py | 9 ++++--
5 files changed, 37 insertions(+), 38 deletions(-)
diff --git a/saqc/core/register.py b/saqc/core/register.py
index 50df11b6b..ce88dc4bc 100644
--- a/saqc/core/register.py
+++ b/saqc/core/register.py
@@ -237,7 +237,7 @@ def _maskData(data, flagger, columns, thresh) -> Tuple[dios.DictOfSeries, dios.D
# we use numpy here because it is faster
for c in columns:
- col_mask = _getMask(flagger[c].to_numpy(), thresh)
+ col_mask = isflagged(flagger[c].to_numpy(), thresh)
if any(col_mask):
col_data = data[c].to_numpy(dtype=np.float64)
@@ -249,7 +249,7 @@ def _maskData(data, flagger, columns, thresh) -> Tuple[dios.DictOfSeries, dios.D
return data, mask
-def _getMask(flags: Union[np.array, pd.Series], thresh: float) -> Union[np.array, pd.Series]:
+def isflagged(flags: Union[np.array, pd.Series], thresh: float) -> Union[np.array, pd.Series]:
"""
Return a mask of flags accordingly to `thresh`. Return type is same as flags.
"""
diff --git a/saqc/funcs/interpolation.py b/saqc/funcs/interpolation.py
index dd5036d9c..c0b9b8ee0 100644
--- a/saqc/funcs/interpolation.py
+++ b/saqc/funcs/interpolation.py
@@ -10,21 +10,26 @@ import pandas as pd
from dios import DictOfSeries
from saqc.constants import *
-from saqc.core.register import register
+from saqc.core.register import register, isflagged
from saqc.flagger import Flagger
from saqc.flagger.flags import applyFunctionOnHistory
from saqc.lib.tools import toSequence, evalFreqStr, getDropMask
from saqc.lib.ts_operators import interpolateNANs
+_SUPPORTED_METHODS = Literal[
+ "linear", "time", "nearest", "zero", "slinear", "quadratic", "cubic", "spline", "barycentric",
+ "polynomial", "krogh", "piecewise_polynomial", "spline", "pchip", "akima"
+]
+
@register(masking='field', module="interpolation")
def interpolateByRolling(
data: DictOfSeries, field: str, flagger: Flagger,
winsz: Union[str, int],
- func: Callable[[pd.Series], float]=np.median,
- center: bool=True,
- min_periods: int=0,
+ func: Callable[[pd.Series], float] = np.median,
+ center: bool = True,
+ min_periods: int = 0,
flag: float = UNFLAGGED,
**kwargs
) -> Tuple[DictOfSeries, Flagger]:
@@ -93,10 +98,10 @@ def interpolateInvalid(
data: DictOfSeries,
field: str,
flagger: Flagger,
- method: Literal["linear", "time", "nearest", "zero", "slinear", "quadratic", "cubic", "spline", "barycentric", "polynomial", "krogh", "piecewise_polynomial", "spline", "pchip", "akima"],
- inter_order: int=2,
- inter_limit: int=2,
- downgrade_interpolation: bool=False,
+ method: _SUPPORTED_METHODS,
+ inter_order: int = 2,
+ inter_limit: int = 2,
+ downgrade_interpolation: bool = False,
not_interpol_flags=None,
flag: float = UNFLAGGED,
**kwargs
@@ -165,7 +170,7 @@ def interpolateInvalid(
return data, flagger
-def _overlap_rs(x, freq='1min', fill_value=-np.inf):
+def _overlap_rs(x, freq='1min', fill_value=UNFLAGGED):
end = x.index[-1].ceil(freq)
x = x.resample(freq).max()
x = x.combine(x.shift(1, fill_value=fill_value), max)
@@ -184,10 +189,7 @@ def interpolateIndex(
field: str,
flagger: Flagger,
freq: str,
- method: Literal[
- "linear", "time", "nearest", "zero", "slinear", "quadratic", "cubic", "spline", "barycentric",
- "polynomial", "krogh", "piecewise_polynomial", "spline", "pchip", "akima"
- ],
+ method: _SUPPORTED_METHODS,
inter_order: int = 2,
downgrade_interpolation: bool = False,
inter_limit: int = 2,
@@ -252,23 +254,19 @@ def interpolateIndex(
start, end = datcol.index[0].floor(freq), datcol.index[-1].ceil(freq)
grid_index = pd.date_range(start=start, end=end, freq=freq, name=datcol.index.name)
- # always injected by register
- to_mask = kwargs['to_mask']
+ flagged = isflagged(flagscol, kwargs['to_mask'])
- datcol.drop(flagscol[flagscol >= to_mask].index, inplace=True)
- datcol.dropna(inplace=True)
- dat_index = datcol.index
+ # drop all points that hold no relevant grid information
+ datcol = datcol[~flagged].dropna()
# account for annoying case of subsequent frequency aligned values,
# that differ exactly by the margin of 2*freq
- gaps = ((dat_index[1:] - dat_index[:-1]) == 2*pd.Timedelta(freq))
- gaps = dat_index[1:][gaps]
- aligned_gaps = gaps.join(grid_index, how='inner')
- if not aligned_gaps.empty:
- aligned_gaps = aligned_gaps.shift(-1, freq)
+ gaps = datcol.index[1:] - datcol.index[:-1] == 2 * pd.Timedelta(freq)
+ gaps = datcol.index[1:][gaps]
+ gaps = gaps.intersection(grid_index).shift(-1, freq)
# prepare grid interpolation:
- datcol = datcol.reindex(datcol.index.join(grid_index, how="outer",))
+ datcol = datcol.reindex(datcol.index.union(grid_index))
# do the grid interpolation
inter_data = interpolateNANs(
@@ -280,18 +278,17 @@ def interpolateIndex(
)
# override falsely interpolated values:
- inter_data[aligned_gaps] = np.nan
+ inter_data[gaps] = np.nan
# store interpolated grid
data[field] = inter_data[grid_index]
# flags reshaping
- flagscol.drop(flagscol[flagscol >= to_mask].index, inplace=True)
+ flagscol = flagscol[~flagged]
flagscol = _overlap_rs(flagscol, freq, UNFLAGGED)
flagger = applyFunctionOnHistory(
- flagger,
- field,
+ flagger, field,
hist_func=_overlap_rs, hist_kws=dict(freq=freq, fill_value=UNFLAGGED),
mask_func=_overlap_rs, mask_kws=dict(freq=freq, fill_value=False),
last_column=flagscol
diff --git a/saqc/funcs/resampling.py b/saqc/funcs/resampling.py
index 3e24cd505..848cf6ee9 100644
--- a/saqc/funcs/resampling.py
+++ b/saqc/funcs/resampling.py
@@ -12,7 +12,7 @@ import pandas as pd
from dios import DictOfSeries
from saqc.constants import *
-from saqc.core.register import register
+from saqc.core.register import register, isflagged
from saqc.flagger import Flagger, initFlagsLike, History
from saqc.funcs.tools import copy, drop, rename
from saqc.funcs.interpolation import interpolateIndex
@@ -329,7 +329,7 @@ def mapToOriginal(
"""
newfield = str(field) + '_original'
- data, flagger = reindexFlags(data, newfield, flagger, method, source=field, to_drop=to_drop, **kwargs)
+ data, flagger = reindexFlags(data, newfield, flagger, method, source=field, to_mask=False)
data, flagger = drop(data, field, flagger)
data, flagger = rename(data, newfield, flagger, field)
return data, flagger
@@ -756,8 +756,7 @@ def reindexFlags(
merge_dict = dict(freq=tolerance, method=projection_method)
if method[-5:] == "shift":
- to_mask = kwargs['to_mask']
- drop_mask = (target_datcol.isna() | target_flagscol >= to_mask)
+ drop_mask = (target_datcol.isna() | isflagged(target_flagscol, kwargs['to_mask']))
projection_method = METHOD2ARGS[method][0]
tolerance = METHOD2ARGS[method][1](freq)
merge_func = _inverseShift
diff --git a/saqc/lib/ts_operators.py b/saqc/lib/ts_operators.py
index 44f91cb64..de9de79d2 100644
--- a/saqc/lib/ts_operators.py
+++ b/saqc/lib/ts_operators.py
@@ -203,7 +203,7 @@ def interpolateNANs(data, method, order=2, inter_limit=2, downgrade_interpolatio
"""
inter_limit = int(inter_limit)
data = pd.Series(data).copy()
- gap_mask = (data.rolling(inter_limit, min_periods=0).apply(lambda x: np.sum(np.isnan(x)), raw=True)) != inter_limit
+ gap_mask = data.isna().rolling(inter_limit, min_periods=0).sum() != inter_limit
if inter_limit == 2:
gap_mask = gap_mask & gap_mask.shift(-1, fill_value=True)
diff --git a/tests/funcs/test_harm_funcs.py b/tests/funcs/test_harm_funcs.py
index f78f8e573..a83368090 100644
--- a/tests/funcs/test_harm_funcs.py
+++ b/tests/funcs/test_harm_funcs.py
@@ -61,9 +61,9 @@ def test_harmSingleVarIntermediateFlagging(data, reshaper):
else:
raise NotImplementedError('untested test case')
- assert all(flagger[field].iloc[start:end])
- assert all(~flagger[field].iloc[:start])
- assert all(~flagger[field].iloc[end:])
+ assert all(flagger[field].iloc[start:end] > UNFLAGGED)
+ assert all(~flagger[field].iloc[:start] == UNFLAGGED)
+ assert all(~flagger[field].iloc[end:] == UNFLAGGED)
elif 'shift' in reshaper:
if reshaper == "nshift":
@@ -78,6 +78,9 @@ def test_harmSingleVarIntermediateFlagging(data, reshaper):
flagged = flagger[field] > UNFLAGGED
assert all(flagged == exp)
+ elif reshaper == 'interpolation':
+ pytest.skip('no testcase for interpolation')
+
else:
raise NotImplementedError('untested test case')
--
GitLab