workers/thread_manager.py - master - zolnikov/database_analyzer

workers/thread_manager.py
145 строк7 KB
Jaman
версия 01.06.26
01 июл 2026, 21:24
01 июл 2026, 21:2493f1e01
from PySide6.QtCore import QObject
from PySide6.QtCore import QThread, Signal, Slot

from common import TypesMessages
from logic.logic_device import LogicDevice
from workers.file_reader import FileReader


class Worker(QObject):
    result_ready = Signal(object)
    error_occurred = Signal(str)
    status_in_work = Signal(bool)
    finished = Signal()

    def __init__(self, func, *args, **kwargs):
        super(Worker, self).__init__()
        self.func = func
        self.args = args
        self.kwargs = kwargs

    @Slot()
    def run(self):
        try:
            self.status_in_work.emit(True)
            result = self.func(*self.args, **self.kwargs)
            self.result_ready.emit(result)
        except Exception as e:
            error_msg = f"Ошибка выполнения: {type(e).__name__}: {str(e)}"
            print(error_msg)
            self.error_occurred.emit(error_msg)
        finally:
            self.status_in_work.emit(False)
            self.finished.emit()


class NewThread(QThread):
    result_statistic = Signal(object)
    error_occurred = Signal(str)
    status_update = Signal(bool)
    result_read_files = Signal(object)
    finished = Signal()

    def __init__(self, on_error=None, on_status=None):
        super().__init__()
        self.func = None
        if on_error:
            self.error_occurred.connect(on_error)
        if on_status:
            self.status_update.connect(on_status)

        self.file_names = ''
        self.start_time = ''
        self.end_time = ''

    def run(self):
        try:
            self.status_update.emit(True)
            result = self.func()
        except Exception as e:
            error_msg = f"Ошибка выполнения: {type(e).__name__}: {str(e)}"
            print(error_msg)
            self.error_occurred.emit(error_msg)
        finally:
            self.finished.emit()
            self.status_update.emit(False)

    def set_function(self, func):
        self.func = func

    def read_files_data(self, file_names):
        f_reader = FileReader()
        df = f_reader.read_data_from_files(file_names)
        self.result_read_files.emit(df)

    def data_acquisition(self, df, start_time, end_time):
        f_reader = FileReader()
        united_df = f_reader.time_condition(df, start_time, end_time)
        data = self.create_main_statistic(f_reader, united_df)
        filter_df = self.create_device_statistic(f_reader, united_df)
        list_name_dev = f_reader.get_list_unique(united_df, 'name_device')
        dict_name_df = self.get_df_logic_device(list_name_dev, united_df)
        self.result_statistic.emit((united_df, data, filter_df, dict_name_df))

    def create_main_statistic(self, f_reader, df):
        count_alert_msg = f_reader.count_type_message(df, 'type_message', TypesMessages.alert)
        count_warning_msg = f_reader.count_type_message(df, 'type_message', TypesMessages.warning)
        count_error_msg = f_reader.count_type_message(df, 'type_message', TypesMessages.error)
        count_dev_alert = f_reader.count_device_with_type_msg(df,
                                                              'name_device',
                                                              'type_message',
                                                              TypesMessages.alert)
        count_dev_warning = f_reader.count_device_with_type_msg(df,
                                                                'name_device',
                                                                'type_message',
                                                                TypesMessages.warning)
        count_dev_error = f_reader.count_device_with_type_msg(df,
                                                              'name_device',
                                                              'type_message',
                                                              TypesMessages.error
                                                              )
        count_dev = f_reader.count_device(df, 'name_device')
        data = {'Всего устройств:': count_dev,
                'Аварийных сообщений': count_alert_msg,
                'Предупреждений': count_warning_msg,
                'Ошибок': count_error_msg,
                'Устройств с аварийными сообщениями': count_dev_alert,
                'Устройств с предупреждениями': count_dev_warning,
                'Устройств с ошибками': count_dev_error,
                }
        return data

    def create_device_statistic(self, f_reader, united_df):
        df = f_reader.get_df_device_with_count_type_msg(united_df,
                                                        ['name_device', 'type_message',
                                                         ])
        filter_df = f_reader.reindex_column(df, ['name_device', TypesMessages.info,
                                                 TypesMessages.alert, TypesMessages.warning,
                                                 TypesMessages.error])
        mapping = f_reader.create_mapping_group(united_df, 'name_device', 'type_device')
        filter_df = f_reader.add_column_to_df_map(filter_df, 'type_device', 'name_device', mapping)
        return filter_df

    def create_logic_device(self, list_name_dev, united_df):
        devices = {}
        for name in list_name_dev:
            device = LogicDevice.create(self, name, united_df)
            if device is not None:
                devices[name] = device
        return devices

    def get_df_logic_device(self, list_name, united_df):
        dict_name_df = {}
        for name in list_name:
            df_device = united_df[united_df['name_device'] == name]
            type_device = int(df_device['type_device'].values[0])
            df_device = (df_device[['name_parameter', 'text_message', 'type_message', 'event_time']].
                         sort_values("event_time", ascending=True).reset_index(drop=True))
            df_device = df_device.sort_values(['name_parameter', 'event_time'])
            df_device['prev_text_msg'] = df_device.groupby('name_parameter', observed=True)['text_message'].shift(1)
            df_device['prev_event_time'] = df_device.groupby('name_parameter', observed=True)['event_time'].shift(1)
            df_device['next_text_msg'] = df_device.groupby('name_parameter', observed=True)['text_message'].shift(-1)
            df_device['next_event_time'] = df_device.groupby('name_parameter', observed=True)['event_time'].shift(-1)
            df_device.sort_values("event_time", ascending=True, inplace=True)
            dict_name_df[name] = (df_device, type_device)
        return dict_name_df