cdplib/import_process_instances/CleanRs0.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on Mon Sep 30 10:14:46 2019

@author: tanya
"""

import pandas as pd
import os
import sys

sys.path.append(os.getcwd())

from libraries.import_process_instances.CleanProcessTable import CleanTable


class CleanRs0(CleanTable):
    '''
    '''
    def __init__(self):
        '''
        '''
        super().__init__(
                mapping_path=os.path.join(".", "migration_mappings",
                                          "rs0_mapping.json"),
                inconsist_report_table="inconsist_rs0",
                filter_index_columns=["radsatznummer"],
                sort_columns=["radsatznummer", "eingabe_datum"],
                index_columns=["radsatznummer", "eingabe_datum"],
                log_name="CleanRs0:")

    def restrict_to_process_data(self, data: pd.DataFrame) -> pd.DataFrame:
        '''
        '''
        process_columns = ["radsatznummer", "aufarbeitungstyp", "ihs",
                           "befundung_code_1", "befundung_code_2",
                           "befundung_code_3"]

        self.error_column_abscence(columns=process_columns,
                                   data=data)

        return data[process_columns]

    def add_ist_schrott(self, data: pd.DataFrame) -> pd.DataFrame:
        '''
        '''
        mongo_name = "final_state.ist_schrott"

        self.error_column_abscence(columns=["aufarbeitungstyp"],
                                   data=data)

        data[mongo_name] = (data["aufarbeitungstyp"] == 2)

        return data

    def restrict_to_meta_data(self, data: pd.DataFrame) -> pd.DataFrame:
        '''
        '''
        meta_columns = [c for c in data.columns if c not in
                        ["aufarbeitungstyp", "ihs",
                         "befundung_code_1", "befundung_code_2",
                         "befundung_code_3"]]

        self.error_column_abscence(columns=meta_columns,
                                   data=data)

        return data[meta_columns]

    def filter_invalid_metacolumns(self, data: pd.DataFrame,
                                   metacolumns: list = None) -> pd.DataFrame:
        '''
        '''
        if metacolumns is None:
            metacolumns = ["wellentype", "Lagerbauart", "tauschgruppe"]

        for column in metacolumns:

            invalid_mask = data[column].isnull()

            reason = "Missing {}".format(column)

            data = self._filter_invalid_data(invalid_mask=invalid_mask,
                                             reason=reason,
                                             data=data)

        return data