workers/file_reader.py - master - zolnikov/database_analyzer

workers/file_reader.py
84 строки4 KB
Jaman
версия 01.06.26
01 июл 2026, 21:24
01 июл 2026, 21:2493f1e01
import pandas as pd


class FileReader:
    def __init__(self):
        pass

    def read_file(self, file_path):
        df = pd.read_csv(file_path, delimiter=',', encoding='utf-8', low_memory=False)
        return df

    def optimize_df_events(self, df: pd.DataFrame):
        values = {"text_message": '', 'text_recommend': '',
                  'type_message': '',
                  }
        df.fillna(values, inplace=True)
        df[['name_device', 'name_parameter', 'text_message']] = df[
            ['name_device', 'name_parameter', 'text_message']].convert_dtypes(convert_string=True)
        df[['text_recommend', 'type_message', 'value']] = df[
            ['text_recommend', 'type_message', 'value']].convert_dtypes(convert_string=True)
        # df[['text_recommend', 'type_message', 'value']] = df[
        #     ['text_recommend', 'type_message', 'value']].convert_dtypes(convert_string=True)
        df["event_time"] = pd.to_datetime(df["event_time"])
        df['type_message'] = df['type_message'].astype('category')
        df['name_device'] = df['name_device'].astype('category')
        return df

    def time_condition(self, df_events: pd.DataFrame, start_time: str, end_time: str):
        if start_time and end_time:
            start_time = pd.to_datetime(start_time)
            end_time = pd.to_datetime(end_time)
            return df_events.loc[(df_events["event_time"] >= start_time) &
                                 (df_events["event_time"] <= end_time), :]
        else:
            return df_events

    def read_data_from_files(self, file_names):
        dataframes = []
        for file_name in file_names:
            df = self.read_file(file_name)
            df = self.optimize_df_events(df)
            dataframes.append(df)
        united_df = self.unite_df(dataframes)
        return united_df

    def unite_df(self, dataframes: list):
        return pd.concat(dataframes, ignore_index=True)

    def count_type_message(self, df: pd.DataFrame, column_name: str, type_msg: str):
        return df[column_name].value_counts().get(type_msg, 0)

    def get_all_device_with_count_type_msg(self, df: pd.DataFrame, device_column: str, msg_column: str, type_msg: str):
        result_df = df.loc[df[msg_column].eq(type_msg)].groupby(device_column, observed=True).size().reset_index(name='msg_count')
        return result_df.set_index(device_column)['msg_count'].to_dict()

    def count_device_with_type_msg(self, df: pd.DataFrame, device_column: str, msg_column: str, type_msg: str):
        devices_count = df.loc[df[msg_column].eq(type_msg), device_column].nunique()
        return devices_count

    def get_df_device_with_count_type_msg(self, df: pd.DataFrame, group_list: list):
        result = df.groupby(group_list, observed=True).size().unstack(fill_value=0).reset_index()
        result.columns.name = ''
        return result

    def reindex_column(self, df: pd.DataFrame, columns: list):
        return df.reindex(columns=columns, fill_value=0)

    def filter_type_msgs(self, df: pd.DataFrame, msg_column: str, type_msgs: list):
        result = df.loc[df[msg_column].isin(type_msgs)]
        return result

    def count_device(self, df: pd.DataFrame, column_name: str):
        return df[column_name].unique().shape[0]

    def get_list_unique(self, df: pd.DataFrame, column_name: str):
        return df[column_name].unique().tolist()

    def create_mapping_group(self, df: pd.DataFrame, column_name: str, column_value: str):
        mapping = df.groupby(column_name)[column_value].first()
        return mapping

    def add_column_to_df_map(self, df1, new_column, add_column, mapping):
        df1[new_column] = df1[add_column].map(mapping)
        return df1