import pandas as pd
from pandas import DataFrame
import pandas as pd
import json
from loguru import logger
[docs]class Treatment:
def __init__(self) -> None:
self.colunas_vazias = []
pass
[docs] def add_column_year(self, data: DataFrame, column: str) -> DataFrame:
"""
Método add_column_year: Adiciona uma coluna de ano baseado na coluna de datas de um dataframe temporal.
Parâmetros:
data: Dataframe.
column: str.
Descrição dos parâmetros:
data -> Argumento onde o usuário fornece um conjunto de dados em formato de dataframe.
column -> Coluna de referência (DATA) para que a coluna de ano seja criada baseada nela.
Exemplo de uso:
tratamento = Treatment()
df = 'seu_dataframe'
df_com_coluna_de_ano = tratamento.add_column_year(data=df, column='DATA')
"""
logger.info('Adcionando coluna de ano...')
self.data = data.copy()
self.data.loc[:, 'Ano'] = self.data[column].apply(lambda x: x.year)
logger.info('Coluna de ano adiciona com suceso!')
return self.data
[docs] def add_column_month(self, data: DataFrame, column: str) -> DataFrame:
"""
Método add_column_month: Adiciona uma coluna de mês baseado na coluna de datas de um dataframe temporal.
Parâmetros:
data: Dataframe.
column: str.
Descrição dos parâmetros:
data -> Argumento onde o usuário fornece um conjunto de dados em formato de dataframe.
column -> Coluna de referência (DATA) para que a coluna de ano seja criada baseada nela.
Exemplo de uso:
tratamento = Treatment()
df = 'seu_dataframe'
df_com_coluna_de_mes = tratamento.add_column_month(data=df, column='DATA')
"""
logger.info('adicionando coluna de mês...')
self.data = data.copy()
self.data.loc[:, 'Mês'] = self.data[column].apply(lambda x: x.month)
logger.info('Coluna de mês adicionada com sucesso!')
return self.data
[docs] def add_column_day(self, data: DataFrame, column: str) -> DataFrame:
"""
Método add_column_day: Adiciona uma coluna de dia baseado na coluna de datas de um dataframe temporal.
Parâmetros:
data: Dataframe.
column: str.
Descrição dos parâmetros:
data -> Argumento onde o usuário fornece um conjunto de dados em formato de dataframe.
column -> Coluna de referência (DATA) para que a coluna de ano seja criada baseada nela.
Exemplo de uso:
tratamento = Treatment()
df = 'seu_dataframe'
df_com_coluna_de_ano = tratamento.add_column_day(data=df, column='DATA')
"""
logger.info('adicionando coluna de dia...')
self.data = data.copy()
self.data.loc[:, 'Dia'] = self.data[column].apply(lambda x: x.day_name())
logger.info('Coluna de dia adicionada com sucesso!')
return self.data
[docs] def treatment_of_missing_data(self, data: DataFrame, treatment: str) -> DataFrame:
"""
Método treatment_of_missing_data: Método de tratamento de dados faltantes.
Parâmetros:
data: Dataframe.
treatment: str.
Descrição dos parâmetros:
data -> Argumento onde o usuário fornece um conjunto de dados em formato de dataframe.
treatment -> Argumento onde o usuário fornece o tipo de tratamento que ele deseja (remove, average, zero)
Exemplo de uso:
tratamento = Treatment()
df = 'seu_dataframe'
CASO QUEIRA REMOVER AS LINHAS QUE POSSUEM VALORES NULOS
df_com_dados_nulos_removidos = tratamento.treatment_of_missing_data(data=df, treatment='remove')
CASO QUEIRA PREENCHER OS VALORES NULOS COM A MÉDIA DOS VALORES DA COLUNA
df_com_a_media_preenchendo_os_valores_nulos = tratamento.treatment_of_missing_data(data=df, treatment='average')
CASO QUEIRA PREENCHER OS VALORES NULOS COM ZERO
df_com_valores_nulos_preenchidos_com_zero = tratamento.treatment_of_missing_data(data=df, treatment='zero')
"""
match treatment:
case "remove":
logger.info('removendo valores nulos...')
self.data = data.copy()
self.data = data.dropna()
logger.info('Valores nulos removidos com sucesso!')
return self.data
case "average":
logger.info("preenchendo os valores nulos com a média dos dados...")
self.data = data.copy()
self.data = data.fillna(data.mean(), inplace=True)
logger.info('Valores nulos preenchidos pela média com sucesso!')
return self.data
case "zero":
logger.info("prenchendo os valores nulos com zero...")
self.data = data.copy()
self.data = data.fillna(0, inplace=True)
logger.info('Valores nulos preenchidos por zero com sucesso!')
return self.data
case _:
raise KeyError('Por favor, insira remover para remover valores nulos, media para preencher com a média os valores nulos ou zero para preencher com zero os valores nulos!')
[docs] def date_range(self, data: DataFrame, start_date: str, end_date: str, data_column: str) -> DataFrame:
"""
Método date_range: Filtra seu dataframe de acordo com um intervalo de datas.
Parâmetros:
data: Dataframe.
start_date: str.
end_date = str.
data_column: str.
Descrição dos parâmetros:
data -> Argumento onde o usuário fornece um conjunto de dados em formato de dataframe.
start_date -> Argumento onde o usuário fornece a data inicial para a filtragem.
end_date -> Argumento onde o usuário fornece a data final para a filtragem.
data_column -> Coluna de referência das datas.
Exemplo de uso:
tratamento = Treatment()
df = 'seu_dataframe'
df_filtrado_por_intervalo_de_datas = tratamento.date_range(data=df, start_date='01/01/2020', end_date='07/03/2020', data_column='DATA')
"""
logger.info('filtrando por intervalo de datas')
self.data = data.copy()
self.data = data.loc[(data[data_column] >= start_date) & (data[data_column] <= end_date)]
logger.info(f'Dataframe filtrado no intervalo de {start_date} até {end_date} com sucesso!')
return self.data
[docs] def year_range(self, data: DataFrame, start_year: int, end_year: int, year_column: str) -> DataFrame:
"""
Método year_range: Filtra seu dataframe de acordo com um intervalo de anos.
Parâmetros:
data: Dataframe.
start_year: str.
end_year = str.
year_column: str.
Descrição dos parâmetros:
data -> Argumento onde o usuário fornece um conjunto de dados em formato de dataframe.
start_year -> Argumento onde o usuário fornece o ano inicial para a filtragem.
end_year -> Argumento onde o usuário fornece o ano final para a filtragem.
year_column -> Coluna de referência dos anos.
Exemplo de uso:
tratamento = Treatment()
df = 'seu_dataframe'
df_filtrado_por_intervalo_de_anos = tratamento.year_range(data=df, start_year=1998, end_year=2015, year_column='ANO')
"""
logger.info('filtrando por intervalo de ano...')
self.data = data.copy()
self.data = data.loc[(data[year_column] >= start_year) & (data[year_column] <= end_year)]
logger.info(f'Dataframe filtrado no intervalo de {start_year} até {end_year} com sucesso!')
return self.data
[docs] def column_range(self, data: DataFrame, start_column: str, end_column: str) -> DataFrame:
"""
Método column_range: Filtra seu dataframe de acordo com um intervalo de colunas.
Parâmetros:
data: Dataframe.
start_column: str.
end_column = str.
Descrição dos parâmetros:
data -> Argumento onde o usuário fornece um conjunto de dados em formato de dataframe.
start_column -> Argumento onde o usuário fornece a coluna inicial da filtragem.
end_column -> argumento onde o usuário fornece a coluna final da filtragem.
Exemplo de uso:
tratamento = Treatment()
df = 'seu_dataframe'
df_filtrado_por_intervalo_de_colunas = tratamento.column_range(data=df, start_column='Coluna_5', end_column='coluna_10')
"""
logger.info('filtrando por intervalo de colunas...')
self.data = data.copy()
self.data = data.loc[:, start_column:end_column]
logger.info(f'Dataframe filtrado no intervalo de colunas da coluna {start_column} até a coluna {end_column} com sucesso!')
return self.data
[docs] def convert_json_to_dataframe(self, json_file: json) -> DataFrame:
"""
Método convert_json_to_dataframe: Método responsável por converter um arquivo json para o formato dataframe.
Parâmetros:
json_file: str
Descrição dos parâmetros:
json_file -> Argumento onde o usuário fornece o caminho do arquivo .json.
Exemplo de uso:
tratamento = Treatment()
json = 'meu_json.json'
df_que_era_json = tratamento.convert_json_to_dataframe(json_file=json)
"""
logger.info('Transformando json em dataframe...')
# Lendo o arquivo JSON em um dicionário
with open(json_file, 'r') as f:
data = json.load(f)
# Convertendo o dicionário em um DataFrame
df = pd.DataFrame.from_dict(data)
logger.info('Json transformado para Dataframe com sucesso!')
return df
[docs] def check_column_with_null_values(self, data: DataFrame) -> list[str]:
"""
Método check_column_with_null_values: Método responsável por informar uma lista com todas as colunas que possuem valores vazios.
Parâmetros:
data: Dataframe.
Descrição dos parâmetros:
data -> Argumento onde o usuário fornece o dataframe.
Exemplo de uso:
tratamento = Treatment()
df = 'meu_dataframe'
lista_das_colunas_com_valores_nulos = tratamento.check_column_with_null_values(data=df)
"""
logger.info('Checando colunas com valores nulos...')
columns_with_null_values = data.columns[data.isnull().any()].tolist()
logger.info('Colunas com valores nulos checadas com sucesso!')
return columns_with_null_values
[docs] def remove_column(self, data: DataFrame, column: str = None, index: int = None, last: str = None) -> DataFrame:
"""
Método remove_column: Método utilizado para remover uma coluna do dataframe.
Parâmetros:
data: Dataframe.
column: str.
index: int.
last: str.
Descrição dos parâmetros:
data -> Argumento onde o usuário fornece um conjunto de dados em formato de dataframe.
column -> Argumento onde o usuário fornece a coluna a ser removida. (OPCIONAL)
index -> Argumento onde o usuário fornece o índice da coluna a ser removido. (OPCIONAL)
last -> Argumento onde o usuário define se é a última coluna do cabeçalho ou não. (OPCIONAL)
Exemplo de uso:
tratamento = Treatment()
df = 'seu_dataframe'
CASO QUEIRA REMOVER A COLUNA PELO NOME DELA
df_coluna_removida_pelo_nome = tratamento.remove_column(data=df, column='Coluna_3')
CASO QUEIRA REMOVER A COLUNA PELO ÍNDICE DELA
df_coluna_removida_pelo_indice = tratamento.remove_column(data=df, index=2)
CASO QUEIRA REMOVER A ÚLTIMA COLUNA
df_ultima_coluna_removida = tratamento.remove_column(data=df, last='yes')
"""
if index == None and column != None:
logger.info('Removendo coluna...')
data = data.drop(column, axis=1)
logger.info('Coluna removida com sucesso!')
return data
elif column == None and index != None:
logger.info('Removendo coluna...')
data = data.pop(data.columns[index])
logger.info('Coluna removida com sucesso!')
return data
elif last == 'yes':
logger.info('Removendo a última coluna...')
data = data.iloc[:,:-1]
logger.info('Última coluna removida com sucesso!')
return data
else:
raise Exception('Error')
[docs] def remove_row(self, data: DataFrame, index: int) -> DataFrame:
"""
Método remove_row: Método responsável por remover uma linha do dataframe.
Parâmetros:
data: Dataframe.
index: int.
Descrição dos parâmetros:
data -> Argumento onde o usuário fornece o dataframe.
index -> Argumento onde o usuário fornece o índice da linha a ser removida do dataframe.
Exemplo de uso:
tratamento = Treatment()
df = 'meu_dataframe'
df_sem_a_linha_10 = tratamento.remove_row(data=df, index=10)
"""
logger.info('Removendo linha...')
data = data.drop(index, axis=0)
data = data.reset_index(drop=True)
logger.info('Linha removida com sucesso!')
return data
[docs] def convert_dataframe_to_json(self, data: DataFrame, json_path: str) -> json:
"""
Método convert_dataframe_to_json: Método responsável por converter um dataframe em um arquivo json.
Parâmetros:
data: Dataframe.
json_path: str.
Descrição dos parâmetros:
data -> Argumento onde o usuário fornece o dataframe.
json_path -> Argumento onde o usuário fornece o caminho de qual arquivo json ele deve ser salvo.
Exemplo de uso:
tratamento = Treatment()
df = 'meu_data_frame'
json = 'meu_json.json'
df_que_era_json = tratamento.convert_dataframe_to_json(data=df, json_path: json)
NOTA:
Crie um arquivo json que esteja vazio, e passe para a função para que seu dataframe seja salvo nesse arquivo .json criado.
"""
logger.info('Transformando dataframe em json...')
dados_dict = data.to_dict(orient='records')
dados_json = json.dumps(dados_dict)
try:
with open(json_path, 'w') as json_file:
json_file.write(dados_json)
logger.info('Dataframe transformado para json com sucesso!')
return 0
except Exception as e:
print(f'Erro no processo da transformação! {str(e)}')
return 1
[docs] def dataframe_for_a_row(self, data: DataFrame, index: int) -> DataFrame:
"""
Método dataframe_for_a_row: Método responsável por gerar um dataframe a partir de uma linha desse dataframe.
Parâmetros:
data: Dataframe.
index: int.
Descrição dos parâmetros:
data -> Argumento onde o usuário fornece o dataframe.
index -> Argumento onde o usuário fornece o índice da linha a ser usada como primeira linha do dataframe.
Exemplo de uso:
tratamento = Treatment()
df = 'meu_dataframe'
df_a_partir_da_linha_5 = tratamento.dataframe_for_a_row(data=df, index=5)
"""
logger.info('Gerando um dataframe a partir de uma linha...')
df = data.iloc[index:]
df = df.reset_index(drop=True)
logger.info('Dataframe gerado com sucesso!')
return df
[docs] def change_data_type(self, data: DataFrame, column: str, type: str) -> DataFrame:
"""
Método change_data_type: Método para alterar o tipo de dados de uma coluna.
Parâmetros:
data: Dataframe.
column: str.
type: str.
Descrição dos parâmetros:
data -> Argumento onde o usuário fornece um conjunto de dados em formato de dataframe.
column -> Argumento onde o usuário fornece a coluna a ser alterada o tipo de dados.
type -> Argumento onde o usuário fornece para qual tipo de dados deseja alterar.
Exemplo de uso:
tratamento = Treatment()
df = 'seu_dataframe'
CASO QUEIRA ALTERAR PARA DATETIME
df_com_coluna_alterada_para_datetime = tratamento.change_data_type(data=df, column='DATA', type='datetime')
NOTA:
Nessa versão só está disponível a mudança para datetime
"""
logger.info('Alterando tipo de dados...')
match type:
case 'datetime':
data[column] = pd.to_datetime(data[column])
logger.info(f'Coluna {column} alterada para datetime com sucesso!')
return data
case _:
return 'error'
[docs] def replace_value_in_a_position(self, data: DataFrame, column: int, row: int, name: str) -> DataFrame:
"""
Método replace_value_in_a_position: Método responsável por trocar o valor de uma posição do dataframe.
Parâmetros:
data: Dataframe.
column: int
row: int
name: str
Descrição dos parâmetros:
data -> Argumento onde o usuário fornece o dataframe.
column -> Argumento onde o usuário fornece o número da coluna.
row -> Argumento onde o usuário fornece o número da linha.
name -> Valor a ser atribuido a essa coordenada.
Exemplo de uso:
tratamento = Treatment()
df = 'meu_dataframe'
df_com_valor_substituido = tratamento.replace_value_in_a_position(data=Dataframe, column=3, row=2, name='Agora eu sou um novo valor')
"""
logger.info('Substituindo um valor em uma posição...')
data.iloc[column, row] = name
logger.info('Valor substituido com sucesso!')
return data