/
zolnikov
/
database_analyzer
Обзор
Документация
Войти
/
zolnikov
/
database_analyzer
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
logic/manager.py
232 строки
10 KB
Jaman
версия 01.06.26
01 июл 2026, 21:24
01 июл 2026, 21:24
93f1e01
Код
Авторство
О чём код?
import functools import os import pandas as pd from PySide6.QtCore import Slot from PySide6.QtWidgets import QFileDialog from common import TypesMessages, TimeObject from window.graph_window import WindowGraph from logic.logic_device import LogicDevice from window.main_window import MainWindow from window.select_point_table import SelectPointTable from window.types_window import TypeWindow from workers.thread_manager import NewThread class MainManager: def __init__(self, app): self.app = app self.main_window = MainWindow(self, app) self.graph_window = WindowGraph(self) self.points_data_table = SelectPointTable() self.type_window = TypeWindow() self.file_names = [] self.start_time = '' self.end_time = '' self.devices = {} self.dataframe = pd.DataFrame() self.new_thread = NewThread(self.show_error, self.show_loading) self.new_thread.result_read_files.connect(self.set_dataframe) self.new_thread.result_statistic.connect(self.continuation_of_the_analysis) def show_window(self): self.main_window.show() def select_files(self): self.clear_all_data() file_names = self.open_files_dialog() if file_names: self.main_window.add_files_in_tab(file_names) self.update_files() self.read_data_from_files(file_names) def read_data_from_files(self, file_names): self.new_thread.quit() self.new_thread.set_function(functools.partial(self.new_thread.read_files_data, file_names)) self.new_thread.start() @Slot(object) def set_dataframe(self, df): start_time = pd.to_datetime(df.iloc[0]["event_time"]).strftime("%Y-%m-%d %H:%M:%S") end_time = pd.to_datetime(df.iloc[-1]["event_time"]).strftime("%Y-%m-%d %H:%M:%S") self.dataframe = df self.main_window.set_range_time_files(start_time, end_time) def update_files(self): self.file_names = self.main_window.get_files_from_tab() def start_analysis(self): self.clear_analysis() if len(self.file_names) > 0: self.start_time = self.main_window.start_time_analysis self.end_time = self.main_window.end_time_analysis self.new_thread.quit() self.new_thread.set_function(functools.partial(self.new_thread.data_acquisition, self.dataframe, self.start_time, self.end_time)) self.new_thread.start() @Slot(object) def continuation_of_the_analysis(self, result): united_df = result[0] data = result[1] filter_df = result[2] dict_name_df = result[3] self.create_log_dev_from_dict(dict_name_df) self.set_data_in_main_window(data) self.set_data_in_main_table(filter_df) self.create_main_graph(united_df) def set_data_in_main_window(self, data): self.main_window.add_static_in_table(data) def set_data_in_main_table(self, filter_df): for i in range(0, len(filter_df)): name_device = filter_df.at[i, 'name_device'] info = filter_df.at[i, TypesMessages.info] alert = filter_df.at[i, TypesMessages.alert] warning = filter_df.at[i, TypesMessages.warning] error = filter_df.at[i, TypesMessages.error] id_type = filter_df.at[i, 'type_device'] self.main_window.add_device_in_table(name_device, info, alert, warning, error, id_type) def create_log_dev_from_dict(self, dict_name_df): for name, data in dict_name_df.items(): df = data[0] type_dev = data[1] device = LogicDevice.create(self, name, df, type_dev) if device is not None: self.devices[name] = device def open_files_dialog(self): list_file_names, _ = QFileDialog.getOpenFileNames(None, "Путь для сохранения файла", f"{os.getcwd()}", "runtime_apps files (*.csv)") if len(list_file_names) > 0: return list_file_names else: return None def selected_device(self, name_device, selected_column): device = self.devices.get(name_device) types_msg = [] if device: type_msg = TypesMessages.types_msg.get(selected_column) if type_msg is None: types_msg = list(TypesMessages.types_msg.values()) else: types_msg.append(type_msg) device.select(types_msg, self.start_time, self.end_time) def create_main_graph(self, united_df: pd.DataFrame): for type_msg, legend, color, symbol, in zip([TypesMessages.alert, TypesMessages.warning, TypesMessages.error], ['Аварийные', 'Предупреждения', 'Ошибки'], ['r', 'y', 'm'], ['o', 's', 't'], ): data_graph = self.prepare_graph_data(united_df, type_msg) self.graph_window.set_data(data_graph, type_msg, color, legend, symbol) def prepare_graph_data(self, united_df, type_msg): start_time = pd.to_datetime(self.start_time) end_time = pd.to_datetime(self.end_time) filtered_df = united_df[(united_df['type_message'] == type_msg) & (united_df['event_time'] >= start_time) & (united_df['event_time'] <= end_time)] graph_df = filtered_df.groupby(pd.Grouper(key='event_time', freq='min')).size() graph_df = graph_df[graph_df > 0] return graph_df def show_graphics(self): if not self.graph_window.isVisible(): self.graph_window.show() else: self.graph_window.close() def clear_all_data(self): self.clear_file_date() self.clear_analysis() self.clear_data() def clear_file_date(self): self.main_window.clear_select_files() self.dataframe = None self.file_names.clear() def clear_analysis(self): self.graph_window.clear_data() self.graph_window.close() self.points_data_table.clear_table() self.points_data_table.close() self.main_window.clear_analysis_data() self.clear_data() def clear_data(self): self.start_time = '' self.end_time = '' for device in self.devices.values(): device.clear_data() self.devices.clear() def show_data_select_points(self, groups_point): result_df = self.create_point_data(self.dataframe, groups_point) if not result_df.empty: for row in range(result_df.shape[0]): name_dev = result_df.at[row, 'name_device'] param = result_df.at[row, 'name_parameter'] msg = result_df.at[row, 'text_message'] type_msg = result_df.at[row, 'type_message'] event_time = result_df.at[row, 'event_time'] self.points_data_table.add_data_in_table(name_dev, param, msg, type_msg, event_time) first_row = result_df.head(1) last_row = result_df.tail(1) first_time = pd.to_datetime(first_row['event_time'].values[0]).strftime("%Y-%m-%d %H:%M:%S") last_time = pd.to_datetime(last_row['event_time'].values[0]).strftime("%Y-%m-%d %H:%M:%S") self.points_data_table.set_time(first_time, last_time) self.points_data_table.show() def create_point_data(self, dataframe, groups_point: list): list_df = [] result_df = pd.DataFrame() offset_sec = TimeObject.get_time_offset_seconds() for group in groups_point: for type_msg, timestamps in group.items(): start_dt = pd.to_datetime(timestamps.min(), unit='s') + pd.Timedelta(seconds=offset_sec) end_dt = (pd.to_datetime(timestamps.max(), unit='s') + pd.Timedelta(minutes=1)) + pd.Timedelta( seconds=offset_sec) df = dataframe[(dataframe['event_time'] >= start_dt) & (dataframe['event_time'] < end_dt) & (dataframe['type_message'] == type_msg)] list_df.append(df) if len(list_df) > 0: result_df = pd.concat(list_df, ignore_index=True).sort_values('event_time', ascending=True) return result_df def get_events_between(self, event_count: int, delta_time: pd.Timedelta, target_time: pd.Timestamp): idx = self.dataframe[(self.dataframe['event_time'] == target_time)].index[0] before = self.dataframe[(self.dataframe['event_time'] < target_time) & (self.dataframe['event_time'] >= target_time-delta_time)].tail(event_count) after = self.dataframe[(self.dataframe['event_time'] > target_time) & (self.dataframe['event_time'] <= target_time+delta_time)].tail(event_count) return before, self.dataframe.loc[[idx]], after @Slot() def show_error(self, error): print(error) @Slot() def show_loading(self, value: bool): self.main_window.play_loading_screen(value) def open_types(self): self.type_window.close() if self.type_window.isVisible() else self.type_window.show() def close_depend_window(self): self.graph_window.close() for device in self.devices.values(): device.close_depend_window() self.type_window.close()