test_generic.py

#! /usr/bin/env python
# -*- coding: utf-8 -*-

import numpy as np
import pandas as pd
import pytest

from ..common import initData

from saqc.dsl.evaluator import evalExpression
from saqc.flagger.simpleflagger import SimpleFlagger
from saqc.funcs.functions import flagGeneric
from saqc.core.config import Params


def test_ismissing():

    nodata = -9999

    data = initData()
    data.iloc[:len(data)//2, 0] = np.nan
    data.iloc[(len(data)//2)+1:, 0] = nodata

    flagger = SimpleFlagger()
    flags = flagger.initFlags(data)

    var1, var2, *_ = data.columns

    idx = evalExpression("ismissing({:})".format(var1),
                         flagger,
                         data, flags,
                         var2,
                         nodata=nodata)

    fdata = data.loc[idx, var1]
    assert (pd.isnull(fdata) | (fdata == nodata)).all()


def test_isflagged():

    flagger = SimpleFlagger()
    data = initData()
    flags = flagger.initFlags(data)
    var1, var2, *_ = data.columns

    flags = flagger.setFlags(flags, var1, iloc=slice(None, None, 2))
    flags = flagger.setFlags(flags, var2, iloc=slice(None, None, 2))

    idx = evalExpression(f"isflagged({var1})", flagger, data, flags, var2)

    flagged = flagger.isFlagged(flags[var1])
    assert (flagged == idx).all


def test_isflaggedArgument():

    flagger = SimpleFlagger()
    data = initData()
    flags = flagger.initFlags(data)
    var1, var2, *_ = data.columns

    flags = flagger.setFlags(flags, var1, iloc=slice(None, None, 2), flag=1)

    idx = evalExpression(f"isflagged({var1}, 1)", flagger, data, flags, var2)

    flagged = flagger.isFlagged(flags[var1], 1)
    assert (flagged == idx).all


def test_flagFailure():
    flagger = SimpleFlagger()
    data = initData()
    flags = flagger.initFlags(data)
    var1, var2, *_ = data.columns

    # expression does not return a result of identical shape
    with pytest.raises(TypeError):
        flagGeneric(data, flags, var2, flagger,
                    **{Params.FUNC: f"sum({var1})"})

    # need a test for missing variables
    with pytest.raises(NameError):
        flagGeneric(data, flags, var2, flagger,
                    **{Params.FUNC: f"sum({var1 + 'x'})"})


if __name__ == "__main__":
    test_ismissing()
    test_isflagged()
    test_isflaggedArgument()
    test_flagFailure()