#! /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()