nirdizati_light.encoding.data_encoder

  1import numpy as np
  2import pandas as pd
  3from pandas import DataFrame
  4from pandas.core.dtypes.common import is_numeric_dtype
  5from sklearn.preprocessing import LabelEncoder,MinMaxScaler,OneHotEncoder
  6
  7PADDING_VALUE = 0
  8#onehot and minmaxscaler not fully done
  9
 10class Encoder:
 11    def __init__(self, df: DataFrame = None, attribute_encoding=None, prefix_length=None):
 12        self.attribute_encoding = attribute_encoding
 13        self.prefix_length = prefix_length
 14        self._label_encoder = {}
 15        self._numeric_encoder = {}
 16        self._label_dict = {}
 17        self._label_dict_decoder = {}
 18        self._scaled_values = {}
 19        self._unscaled_values = {}
 20        
 21        for column in df:
 22            if column != 'trace_id':
 23                if not is_numeric_dtype(df[column].dtype):#or (is_numeric_dtype(df[column].dtype) and np.any(df[column] < 0)):
 24                    if attribute_encoding == 'label':
 25                        if column == 'label':
 26                            self._label_encoder[column] = LabelEncoder().fit(
 27                                sorted(df[column].apply(lambda x: str(x))))
 28                            classes = self._label_encoder[column].classes_
 29                            transforms = self._label_encoder[column].transform(classes)
 30                            self._label_dict[column] = dict(zip(classes, transforms))
 31                            self._label_dict_decoder[column] = dict(zip(transforms, classes))
 32                        else:
 33                            self._label_encoder[column] = LabelEncoder().fit(
 34                                sorted(pd.concat([pd.Series([str(PADDING_VALUE)]), df[column].apply(lambda x: str(x))])))
 35                            classes = self._label_encoder[column].classes_
 36                            transforms = self._label_encoder[column].transform(classes)
 37                            self._label_dict[column] = dict(zip(classes, transforms))
 38                            self._label_dict_decoder[column] = dict(zip(transforms, classes))
 39                    elif attribute_encoding == "onehot":
 40                        if column == 'label':
 41                            self._label_encoder[column] = LabelEncoder().fit(
 42                                sorted(df[column].apply(lambda x: str(x))))
 43                            classes = self._label_encoder[column].classes_
 44                            transforms = self._label_encoder[column].transform(classes)
 45                            self._label_dict[column] = dict(zip(classes, transforms))
 46                            self._label_dict_decoder[column] = dict(zip(transforms, classes))
 47                        else:
 48                            self._label_encoder[column] = OneHotEncoder(drop='if_binary', sparse_output=False,
 49                                       handle_unknown='ignore').fit(df[column].astype(str).values.reshape(-1,1))
 50                            categories = self._label_encoder[column].categories_[0].reshape(-1, 1)
 51                            transforms = [tuple(enc) for enc in self._label_encoder[column].transform(categories)]
 52                            classes = list(categories.flatten())
 53                            self._label_dict[column] = dict(zip(classes, transforms))
 54                            self._label_dict_decoder[column] = dict(zip(transforms, classes))
 55
 56                else:
 57                    self._numeric_encoder[column] = MinMaxScaler().fit(
 58                        df[column].values.reshape(-1,1)
 59                    )
 60                    unscaled = df[column].values
 61                    scaled = self._numeric_encoder[column].transform(df[column].values.reshape(-1,1)).flatten()
 62                    self._scaled_values[column] = scaled
 63                    self._unscaled_values[column] = unscaled
 64                    print('column:', column, 'considered number, top 5 values are:', list(df[column][:5]))
 65
 66    def encode(self, df: DataFrame) -> None:
 67        for column in df:
 68            if column != 'trace_id':
 69                if column in self._label_encoder:
 70                    try:
 71                        df[column] = df[column].apply(lambda x: self._label_dict[column].get(str(x), PADDING_VALUE))
 72                    except:
 73                        print('Error')
 74                else:
 75                    try:
 76                        df[column] = self._numeric_encoder[column].transform(df[column].values.reshape(-1,1)).flatten()
 77                    except:
 78                        print('Error')
 79
 80
 81    def decode(self, df: DataFrame) -> None:
 82        for column in df:
 83                if column != 'trace_id':
 84                        if column in self._label_encoder:
 85                            df[column] = df[column].apply(lambda x: self._label_dict_decoder[column].get(x, PADDING_VALUE))
 86                        else:
 87                            df[column] = self._numeric_encoder[column].inverse_transform(df[column].values.reshape(-1,1)).flatten()
 88
 89    def decode_row(self, row) -> np.array:
 90        decoded_row = []
 91        for column, value in row.items():
 92            if column != 'trace_id':
 93                if column in self._label_encoder:
 94                     decoded_row += [self._label_dict_decoder[column].get(value, PADDING_VALUE)]
 95                elif column in self._numeric_encoder:
 96                    decoded_row += [self._numeric_encoder[column].inverse_transform(np.array(value).reshape(-1,1))[0][0]]
 97            else:
 98                decoded_row += [value]
 99        return np.array(decoded_row)
100
101    def decode_column(self, column, column_name) -> np.array:
102        decoded_column = []
103        if column != 'trace_id':
104            if column_name in self._encoder:
105                if not is_numeric_dtype(df[column].dtype):
106                    decoded_column += [self._label_dict_decoder[column_name].get(x, PADDING_VALUE) for x in column]
107                else:
108                    decoded_column += [self._unscaled_values[column_name].get(x) for x in column]
109        else:
110            decoded_column += list(column)
111        return np.array(decoded_column)
112
113    def get_values(self, column_name):
114        if not is_numeric_dtype(df[column].dtype):
115            return (self._label_dict[column_name].keys(), self._label_dict_decoder[column_name].keys())