Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
#! /usr/bin/env python
# -*- coding: utf-8 -*-
import pytest
import numpy as np
import pandas as pd
from saqc.flagger.baseflagger import BaseFlagger
from saqc.flagger.dmpflagger import DmpFlagger
from saqc.flagger.simpleflagger import SimpleFlagger
from saqc.funcs.functions import flagRange, flagSesonalRange, forceFlags, clearFlags
TESTFLAGGERS = [
BaseFlagger(['NIL', 'GOOD', 'BAD']),
DmpFlagger(),
SimpleFlagger()]
@pytest.mark.parametrize('flagger', TESTFLAGGERS)
def test_range(flagger):
# prepare
field = 'testdata'
index = pd.date_range(start='2011-01-01', end='2011-01-02', periods=100)
data = pd.DataFrame(data={field: np.linspace(0, index.size - 1, index.size)}, index=index)
flags = flagger.initFlags(data)
# test
data, flags = flagRange(data, flags, field, flagger, min=10, max=90)
flagged = flagger.isFlagged(flags[field])
assert len(flags[flagged]) == 10 + 10
@pytest.mark.parametrize('flagger', TESTFLAGGERS)
def test_flagSesonalRange(flagger):
# prepare
field = 'testdata'
index = pd.date_range(start='2011-01-01', end='2014-12-31', freq='1d')
d = [(x % 2) * 50 for x in range(index.size)]
data = pd.DataFrame(data={field: d}, index=index)
kwargs = dict(min=1, max=100, startmonth=7, startday=1, endmonth=8, endday=31)
data, flags = flagSesonalRange(data, flags, field, flagger, **kwargs)
kwargs = dict(min=1, max=100, startmonth=12, startday=16, endmonth=1, endday=15)
_, flags = flagSesonalRange(data, flags, field, flagger, **kwargs)
@pytest.mark.parametrize('flagger', TESTFLAGGERS)
def test_clearFlags(flagger):
# prepare
field = 'testdata'
index = pd.date_range(start='2011-01-01', end='2011-01-10', freq='1d')
data = pd.DataFrame(data={field: np.linspace(0, index.size - 1, index.size)}, index=index)
orig = flagger.initFlags(data)
flags = orig.copy()
# test
flags = flagger.setFlags(flags, field)
assert (orig != flags).all
_, cleared = clearFlags(data, flags, field, flagger)
assert (orig == cleared).all
@pytest.mark.parametrize('flagger', TESTFLAGGERS)
def test_forceFlags(flagger):
# prepare
field = 'testdata'
index = pd.date_range(start='2011-01-01', end='2011-01-10', freq='1d')
data = pd.DataFrame(data={field: np.linspace(0, index.size - 1, index.size)}, index=index)
flags = flagger.initFlags(data)
flags = flagger.setFlags(flags, field)
orig = flags.copy()
# test
_, foreced = forceFlags(data, flags, field, flagger, flag=flagger.GOOD)
assert (orig != foreced).all
if __name__ == '__main__':
for f in TESTFLAGGERS:
test_range(f)
test_flagSesonalRange(f)
test_clearFlags(f)
test_forceFlags(f)