test_translator.py

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

from collections import namedtuple
from dataclasses import dataclass
from typing import Dict, Union, Sequence

import numpy as np
import pandas as pd

import pytest

from dios import DictOfSeries

from saqc.constants import UNFLAGGED, BAD, DOUBTFUL
from saqc.core.translator import PositionalTranslator, Translator, DmpTranslator
from saqc.core.flags import Flags
from saqc.core.core import SaQC

from tests.common import initData

def _genTranslators():
    for dtype in (str, float, int):
        flags = {dtype(-2): UNFLAGGED, dtype(-1): BAD, **{dtype(f*10): float(f) for f in range(10)}}
        translator = Translator(flags)
        yield flags, translator


def _genFlags(data: Dict[str, Union[Sequence, pd.Series]]) -> Flags:

    flags = DictOfSeries()
    for k, v in data.items():
        if not isinstance(v, pd.Series):
            v = pd.Series(v, index=pd.date_range("2012-01-01", freq="1D", periods=len(v)))
        flags[k] = v

    return Flags(flags)


def test_forwardTranslation():
    for flags, translator in _genTranslators():
        for k, expected in flags.items():
            got = translator.forward(k)
            assert expected == got or np.isnan([got, expected]).all()

        for k in ["bad", 3.14, max]:
            with pytest.raises(ValueError):
                translator.forward(k)


def test_backwardTranslation():
    field = "var1"
    for _, translator in _genTranslators():
        keys = tuple(translator._backward.keys())
        flags = _genFlags({field: np.array(keys)})
        translated = translator.backward(flags, None)
        expected = set(translator._backward.values())
        assert not (set(translated[field]) - expected)


def test_backwardTranslationFail():
    field = "var1"
    for _, translator in _genTranslators():
        keys = tuple(translator._backward.keys())
        # add an scheme invalid value to the flags
        flags = _genFlags({field: np.array(keys + (max(keys) + 1,))})
        with pytest.raises(ValueError):
            translator.backward(flags, None)



def test_dmpTranslator():

    Selector = namedtuple("Selector", ["field"])
    Function = namedtuple("Function", ["name"])

    translator = DmpTranslator()
    keys = np.array(tuple(translator._backward.keys()) * 50)
    flags = _genFlags({"var1": keys, "var2": keys, "var3": keys})
    flags[:, "var1"] = BAD
    flags[:, "var1"] = DOUBTFUL
    flags[:, "var2"] = BAD
    to_call = [
        # # NOTE: work around the default first column history columns (see GL#182)
        # (Selector("var1"), None, Function("-")),
        (Selector("var1"), None, Function("flagFoo")),
        (Selector("var1"), None, Function("flagBar")),
        # # NOTE: work around the default first column history columns (see GL#182)
        # (Selector("var2"), None, Function("-")),
        (Selector("var2"), None, Function("flagFoo")),
    ]
    tflags = translator.backward(flags, to_call)
    assert set(tflags.columns.get_level_values(1)) == {"quality_flag", "quality_comment", "quality_cause"}
    assert (tflags.loc[:, ("var1", "quality_flag")] == "DOUBTFUL").all(axis=None)
    assert (tflags.loc[:, ("var1", "quality_comment")] == '{"test": "flagBar"}').all(axis=None)

    assert (tflags.loc[:, ("var2", "quality_flag")] == "BAD").all(axis=None)
    assert (tflags.loc[:, ("var2", "quality_comment")] == '{"test": "flagFoo"}').all(axis=None)

    assert (tflags.loc[:, ("var3", "quality_comment")] == '{"test": "-"}').all(axis=None)


def test_dmpTranslatorIntegration():

    data = initData(3)
    col = data.columns[0]

    saqc = SaQC(data=data, translator=DmpTranslator())
    saqc = (saqc
            .breaks.flagMissing(col)
            .outliers.flagRange(col, min=3, max=10))
    data, flags = saqc.getResult()


def test_positionalTranslator():
    translator = PositionalTranslator()
    flags = _genFlags({"var1": np.zeros(100), "var2": np.zeros(50)})
    flags[1::3, "var1"] = BAD
    flags[1::3, "var1"] = DOUBTFUL
    flags[2::3, "var1"] = BAD

    tflags = translator.backward(flags, None)  # type: ignore
    assert (tflags["var2"].dropna() == "9").all(axis=None)
    assert (tflags["var1"].iloc[1::3] == "9210").all(axis=None)
    assert (tflags["var1"].iloc[2::3] == "9002").all(axis=None)


def test_positionalTranslatorIntegration():

    data = initData(3)
    col: str = data.columns[0]

    saqc = SaQC(data=data, translator=PositionalTranslator())
    saqc = (saqc
            .breaks.flagMissing(col)
            .outliers.flagRange(col, min=3, max=10))
    data, flags = saqc.getResult()

    for field in flags.columns:
        assert flags[field].str.match("^9[012]*$").all()