From 4dfd9a5a8fae875ccbc76abe918059f8be005399 Mon Sep 17 00:00:00 2001 From: Andrew Date: Thu, 23 Jan 2025 18:13:44 +0300 Subject: [PATCH] =?UTF-8?q?chore:=20=D1=81=D0=BE=D0=B7=D0=B4=D0=B0=D0=BD?= =?UTF-8?q?=20=D0=BA=D0=BB=D0=B0=D1=81=D1=81=20=D0=B4=D0=B0=D0=BD=D0=BD?= =?UTF-8?q?=D1=8B=D1=85=20Settings,=20=D1=81=D0=BA=D0=BE=D1=80=D1=80=D0=B5?= =?UTF-8?q?=D0=BA=D1=82=D0=B8=D1=80=D0=BE=D0=B2=D0=B0=D0=BD=20=D0=BE=D0=B1?= =?UTF-8?q?=D0=BC=D0=B5=D0=BD=20=D0=B4=D0=B0=D0=BD=D0=BD=D1=8B=D0=BC=D0=B8?= =?UTF-8?q?=20=D0=B2=20=D0=BC=D0=B5=D0=B4=D0=B8=D0=B0=D1=82=D0=BE=D1=80?= =?UTF-8?q?=D0=B5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/base/base.py | 25 ++- src/controller/controller.py | 6 +- src/controller/file_manager.py | 99 +++++---- src/controller/mediator.py | 16 +- src/controller/passport_former.py | 329 +++++++++++++++--------------- src/gui/main_gui.py | 59 +++--- src/gui/plotter.py | 89 ++++---- src/main.py | 1 - 8 files changed, 315 insertions(+), 309 deletions(-) diff --git a/src/base/base.py b/src/base/base.py index 7a11479..1e219eb 100644 --- a/src/base/base.py +++ b/src/base/base.py @@ -27,6 +27,12 @@ class GraphicPassport: useful_data: dict +@dataclass +class Settings: + operator: dict + system: dict + + class BaseMediator: def __init__(self, converter: BaseDataConverter, @@ -118,9 +124,11 @@ class BaseDataConverter: class BasePlotWidget: def __init__(self, - mediator: Optional[BaseMediator] = None): + mediator: Optional[BaseMediator] = None, + controller: BaseController = None): super().__init__() self._mediator = mediator + self._controller = controller self._stage_colors = { "Closing": [220, 20, 60, 100], # Crimson @@ -242,6 +250,10 @@ class BasePlotWidget: font-family: "Segoe UI", sans-serif; }""") + @property + def controller(self) -> BaseController: + return self._controller + @property def mediator(self) -> BaseMediator: return self._mediator @@ -348,11 +360,10 @@ class BaseFileManager: class BaseIdealDataBuilder(OptAlgorithm): - def __init__(self, params: list[dict]): - operator_params, system_params = params - self.mul = system_params['time_capture'] - self.welding_time = operator_params['time_wielding'] - super().__init__(operator_params, system_params) + def __init__(self, settings: Settings): + self.mul = settings.system['time_capture'] + self.welding_time = settings.operator['time_wielding'] + super().__init__(settings.system, settings.operator) def get_closingDF(self) -> pd.DataFrame: ... @@ -405,6 +416,7 @@ class BasePointPassportFormer: mediator: Optional[BaseMediator] = None): self._mediator = mediator self._clear_stage = "Welding" + self._settings = Settings self._stages = [ "Closing", "Squeeze", @@ -418,7 +430,6 @@ class BasePointPassportFormer: "Tesla welding", "Tesla oncomming_relief" ] - self._params = [] self._ideal_data_cashe = LRUCache(maxsize=1000) self._OptAlgorithm_operator_params = [ "dist_open_start_1", diff --git a/src/controller/controller.py b/src/controller/controller.py index fd8156a..c7ed62e 100644 --- a/src/controller/controller.py +++ b/src/controller/controller.py @@ -2,7 +2,7 @@ from PyQt5.QtWidgets import QWidget, QTabWidget from PyQt5.QtGui import QPixmap from PyQt5.QtCore import pyqtSignal -from base.base import BaseController +from base.base import BaseController, Settings class Controller(BaseController): @@ -20,8 +20,8 @@ class Controller(BaseController): def send_widgets(self, widgets: list[QWidget]) -> None: self.signal_widgets.emit(widgets) - def update_settings(self, settings: list[dict]) -> None: - self.mediator.notify(self, settings) + def update_settings(self, settings: Settings) -> None: + self._mediator.notify(self, settings) def update_status(self, msg:str) -> None: self.signal_status_text.emit(msg) diff --git a/src/controller/file_manager.py b/src/controller/file_manager.py index 04dcc3a..0b18f49 100644 --- a/src/controller/file_manager.py +++ b/src/controller/file_manager.py @@ -2,7 +2,53 @@ import os from loguru import logger -from base.base import BaseDirectoryMonitor, BaseFileManager +from base.base import BaseDirectoryMonitor, BaseFileManager, Settings + + +class FileManager(BaseFileManager): + + def replot_all(self) -> None: + if self._paths_library is not None: + self._mediator.notify(self, list(self._paths_library)) + + def open_custom_file(self, path:str) -> None: + self._paths_library.add(path) + self._mediator.notify(self, [path if path else '']) + + def set_mode(self, num:int) -> None: + match num: + case 1: # Режим создания отчета + self._monitor.stop() + self._paths_library.clear() + self._paths_library.add('') + print(self._paths_library) + self._mediator.notify(self, list(self._paths_library)) + + case 2: # Режим онлайн-мониторинга папки + self._monitor.init_state() + self._monitor.start() + + def update_monitor_settings(self, settings:Settings) -> None: + directory_path = settings.system['trace_storage_path'][0] + update_time = settings.system['monitor_update_period'][0] + + if not os.path.exists(directory_path): + logger.error(f"Путь {directory_path} не существует.") + #raise FileNotFoundError(f"Путь {directory_path} не существует.") + + if update_time <= 0.01: + logger.error(f"Путь {directory_path} не существует.") + + if self._monitor.isActive: self._monitor.pause() + self._monitor._directory_path = directory_path + self._monitor._update_time = update_time + if self._monitor.isActive: self._monitor.start() + + def add_new_paths(self, paths): + paths_set = set(paths) + new = self._paths_library.difference(paths_set) + self._paths_library.update(new) + self._mediator.notify(list(new)) class DirectoryMonitor(BaseDirectoryMonitor): @@ -33,53 +79,4 @@ class DirectoryMonitor(BaseDirectoryMonitor): self._file_manager.add_new_paths(new_files) self._files = current_files if not current_files: - self._files = [] - - -class FileManager(BaseFileManager): - - def replot_all(self) -> None: - if self._paths_library is not None: - self._mediator.notify(self, list(self._paths_library)) - - def open_custom_file(self, path:str) -> None: - self._paths_library.add(path) - self._mediator.notify(self, [path if path else '']) - - def set_mode(self, num:int) -> None: - match num: - case 1: # Режим создания отчета - self._monitor.stop() - self._paths_library.clear() - self._paths_library.add('') - print(self._paths_library) - self._mediator.notify(self, list(self._paths_library)) - - case 2: # Режим онлайн-мониторинга папки - self._monitor.init_state() - self._monitor.start() - - def update_monitor_settings(self, settings:list[dict]) -> None: - _, system_params = settings - directory_path = system_params['trace_storage_path'][0] - update_time = system_params['monitor_update_period'][0] - - if not os.path.exists(directory_path): - logger.error(f"Путь {directory_path} не существует.") - #raise FileNotFoundError(f"Путь {directory_path} не существует.") - - if update_time <= 0.01: - logger.error(f"Путь {directory_path} не существует.") - - if self._monitor.isActive: self._monitor.pause() - self._monitor._directory_path = directory_path - self._monitor._update_time = update_time - if self._monitor.isActive: self._monitor.start() - - def add_new_paths(self, paths): - paths_set = set(paths) - new = self._paths_library.difference(paths_set) - self._paths_library.update(new) - self._mediator.notify(list(new)) - - + self._files = [] \ No newline at end of file diff --git a/src/controller/mediator.py b/src/controller/mediator.py index 2a00e73..7bfb7cd 100644 --- a/src/controller/mediator.py +++ b/src/controller/mediator.py @@ -3,17 +3,23 @@ from typing import Union import pandas as pd from PyQt5.QtWidgets import QWidget -from base.base import (BaseMediator, BaseFileManager, - BaseDataConverter, BasePlotWidget, - BasePointPassportFormer, - BaseController) +from base.base import ( + BaseMediator, + BaseFileManager, + BaseDataConverter, + BasePlotWidget, + BasePointPassportFormer, + BaseController, + GraphicPassport, + Settings + ) class Mediator(BaseMediator): def notify(self, source: Union[BaseFileManager, BaseDataConverter, BasePointPassportFormer, BasePlotWidget, BaseController], - data: Union[list[str], list[pd.DataFrame], list[list], list[QWidget], list[dict]]): + data: Union[list[str], list[pd.DataFrame], list[GraphicPassport], list[QWidget], Settings]): if issubclass(source.__class__, BaseFileManager): self._controller.update_status("CSV found! Calculating...") diff --git a/src/controller/passport_former.py b/src/controller/passport_former.py index 68b2515..765e071 100644 --- a/src/controller/passport_former.py +++ b/src/controller/passport_former.py @@ -6,8 +6,170 @@ import numpy as np import pandas as pd from loguru import logger -from base.base import BasePointPassportFormer, BaseIdealDataBuilder, PointPassport, GraphicPassport +from base.base import BasePointPassportFormer, BaseIdealDataBuilder, PointPassport, GraphicPassport, Settings + +class PassportFormer(BasePointPassportFormer): + + def form_passports(self, data: list[pd.DataFrame]) -> list[GraphicPassport]: + try: + return_data = [self._build_graphic_passport(dataframe) for dataframe in data] + except: + # TODO: обработка исключений!!! + tb = sys.exc_info()[2] + # TODO: Нормальные сообщения в лог! + tbinfo = traceback.format_tb(tb)[0] + pymsg = "Traceback info:\n" + tbinfo + "\nError Info:\n" + str(sys.exc_info()[1]) + logger.error(pymsg) + return_data = [] + finally: + self._mediator.notify(self, return_data) + + def update_settings(self, settings: Settings): + self._settings = settings + + @staticmethod + def _find_indexes(signal: str, + dataframe: pd.DataFrame) -> tuple[np.ndarray, np.ndarray]: + stage_diff = np.diff(dataframe[signal]) + start_idx = np.where(stage_diff == 1) + finish_idx = np.where(stage_diff == -1) + return start_idx[0], finish_idx[0] + + def _find_events(self, + signal: str, + times:pd.Series, + dataframe: pd.DataFrame) -> tuple[list[float], list[float]]: + start_idx, finish_idx = self._find_indexes(signal, dataframe) + + if len(start_idx) > 0 and len(finish_idx) > 0 and start_idx[0] > finish_idx[0]: + start_idx = np.insert(start_idx, 0, 0) + start_list = times.iloc[start_idx].tolist() if len(start_idx) > 0 else [] + end_list = times.iloc[finish_idx].tolist() if len(finish_idx) > 0 else [] + if len(start_list) - len(end_list) == 1: + end_list.append(float(times.iloc[-1])) + + return start_list, end_list + + def _filter_events(self, + times: pd.Series, + dataframe: pd.DataFrame) -> tuple[dict[str, list[list[float]]], int]: + events = {} + point_quantity = 0 + if self._clear_stage in self._stages: + start_list, end_list = self._find_events(self._clear_stage, times, dataframe) + point_quantity = len(start_list) + if point_quantity == 0: + #TODO: добавить обработку исключения + return [] + + for stage in self._stages: + start_list, end_list = self._find_events(stage, times, dataframe) + temp = min([len(start_list), len(end_list)]) + if temp < point_quantity: + print ("cant find enough", stage) + start_list += [0]*(point_quantity - temp) + end_list += [1]*(point_quantity - temp) + + events[stage] = [start_list, end_list] + return events, point_quantity + + def _build_ideal_data(self, + idealDataBuilder: Optional[BaseIdealDataBuilder] = None, + point_settings: Settings = None) -> dict: + self.opt_algorithm = idealDataBuilder(point_settings) + stage_ideals = { + "Closing": self._opt_algorithm.get_closingDF(), + "Squeeze": self._opt_algorithm.get_compressionDF(), + "Welding": self._opt_algorithm.get_weldingDF(), + "Relief": self._opt_algorithm.get_openingDF(), + "Oncomming": self._opt_algorithm.get_oncomingDF(), + "Ideal cycle": self._opt_algorithm.get_cycle_time(), + "Ideal timings": self._opt_algorithm.get_ideal_timings() + } + return stage_ideals + + def _generate_cache_key(self, + point_settings:Settings) -> tuple[tuple[tuple[str, Any], ...], tuple[tuple[str, Any], ...]]: + """ + Преобразует point_settings в хешируемый ключ для кэша. + """ + # Преобразуем словари в отсортированные кортежи пар ключ-значение + operator_tuple = frozenset((key, value) + for key, value in point_settings.operator.items() + if str(key) in self._OptAlgorithm_operator_params) + + system_tuple = frozenset((key, value) + for key, value in point_settings.system.items() + if str(key) in self._OptAlgorithm_system_params) + return (operator_tuple, system_tuple) + + def _build_graphic_passport(self, dataframe: pd.DataFrame) -> GraphicPassport: + + if dataframe is not None: + events, point_quantity = self._filter_events(dataframe["time"], dataframe) + if point_quantity == 0: + return [] + else: + events = None + key = list(self._settings.operator.keys())[0] + point_quantity = len(self._settings.operator[key]) + + graphic_passport = GraphicPassport() + graphic_passport.dataframe = dataframe + graphic_passport.points_pocket = [] + system_settings = {key: value[0] for key, value in self._settings.system.items()} + graphic_passport.useful_data = self._form_graphic_useful_data(system_settings) + + for i in range(point_quantity): + point_settings = Settings() + point_settings.operator = self._get_operator_settings_part(i) + point_settings.system = system_settings + + point_passport = PointPassport() + point_passport.ideal_data = self._form_point_ideal_data(point_settings) + point_passport.useful_data = self._form_point_useful_data(point_settings.operator) + point_passport.timeframe, point_passport.events = self._form_point_events(events, i) + + graphic_passport.points_pocket.append(point_passport) + return graphic_passport + + def _form_graphic_useful_data(self, system_settings:dict) -> dict: + tesla_time = sum(self._settings.operator.get("Tesla summary time", [])) + useful_data = {"tesla_time": tesla_time, + "range_ME": system_settings["Range ME, mm"], + "k_hardness": system_settings["k_hardness_1"] + } + return useful_data + + def _form_point_useful_data(self, operator_settings:dict) -> dict: + useful_data = {"thickness": operator_settings["object_thickness"], + "L2": operator_settings["distance_l_2"], + "force": operator_settings["force_target"]} + return useful_data + + def _form_point_ideal_data(self, point_settings:Settings) -> dict: + cache_key = self._generate_cache_key(point_settings) + ideal_data = self._ideal_data_cashe.get(cache_key, + self._build_ideal_data(idealDataBuilder=IdealDataBuilder, params=point_settings)) + self._ideal_data_cashe[cache_key] = ideal_data + return ideal_data + + def _get_operator_settings_part(self, idx:int) -> dict: + operator_settings = { + key: (value[idx] if idx < len(value) else value[0]) + for key, value in self._settings.operator.items() + } + return operator_settings + + def _form_point_events(self, events:dict, idx) -> list: + timeframe, point_events = None, None + if events is not None: + idx_shift = idx+1 if events[self._stages[-1]][0][0] == 0 else idx + timeframe = [events[self._stages[0]][0][idx], events[self._stages[-1]][1][idx_shift]] + point_events = {key: [value[0][idx], value[1][idx]] for key, value in events.items()} + return timeframe, point_events + class IdealDataBuilder(BaseIdealDataBuilder): @@ -78,167 +240,4 @@ class IdealDataBuilder(BaseIdealDataBuilder): "Rotor Speed ME":V2*1000, "Force":F }) - return pd.DataFrame(data) - - -class PassportFormer(BasePointPassportFormer): - - def form_passports(self, data: list[pd.DataFrame]) -> list[GraphicPassport]: - try: - return_data = [self._build_graphic_passport(dataframe) for dataframe in data] - except: - # TODO: обработка исключений!!! - tb = sys.exc_info()[2] - # TODO: Нормальные сообщения в лог! - tbinfo = traceback.format_tb(tb)[0] - pymsg = "Traceback info:\n" + tbinfo + "\nError Info:\n" + str(sys.exc_info()[1]) - logger.error(pymsg) - return_data = [] - finally: - self._mediator.notify(self, return_data) - - def update_settings(self, params: list[dict, dict]): - self._params = params - - @staticmethod - def _find_indexes(signal: str, - dataframe: pd.DataFrame) -> tuple[np.ndarray, np.ndarray]: - stage_diff = np.diff(dataframe[signal]) - start_idx = np.where(stage_diff == 1) - finish_idx = np.where(stage_diff == -1) - return start_idx[0], finish_idx[0] - - def _find_events(self, - signal: str, - times:pd.Series, - dataframe: pd.DataFrame) -> tuple[list[float], list[float]]: - start_idx, finish_idx = self._find_indexes(signal, dataframe) - - if len(start_idx) > 0 and len(finish_idx) > 0 and start_idx[0] > finish_idx[0]: - start_idx = np.insert(start_idx, 0, 0) - start_list = times.iloc[start_idx].tolist() if len(start_idx) > 0 else [] - end_list = times.iloc[finish_idx].tolist() if len(finish_idx) > 0 else [] - if len(start_list) - len(end_list) == 1: - end_list.append(float(times.iloc[-1])) - - return start_list, end_list - - def _filter_events(self, - times: pd.Series, - dataframe: pd.DataFrame) -> tuple[dict[str, list[list[float]]], int]: - events = {} - point_quantity = 0 - if self._clear_stage in self._stages: - start_list, end_list = self._find_events(self._clear_stage, times, dataframe) - point_quantity = len(start_list) - if point_quantity == 0: - #TODO: добавить обработку исключения - return [] - - for stage in self._stages: - start_list, end_list = self._find_events(stage, times, dataframe) - temp = min([len(start_list), len(end_list)]) - if temp < point_quantity: - print ("cant find enough", stage) - start_list += [0]*(point_quantity - temp) - end_list += [1]*(point_quantity - temp) - - events[stage] = [start_list, end_list] - return events, point_quantity - - def _build_ideal_data(self, - idealDataBuilder: Optional[BaseIdealDataBuilder] = None, - params: list[dict] = None) -> dict: - self.opt_algorithm = idealDataBuilder(params) - stage_ideals = { - "Closing": self._opt_algorithm.get_closingDF(), - "Squeeze": self._opt_algorithm.get_compressionDF(), - "Welding": self._opt_algorithm.get_weldingDF(), - "Relief": self._opt_algorithm.get_openingDF(), - "Oncomming": self._opt_algorithm.get_oncomingDF(), - "Ideal cycle": self._opt_algorithm.get_cycle_time(), - "Ideal timings": self._opt_algorithm.get_ideal_timings() - } - return stage_ideals - - def _generate_cache_key(self, - params_list: list[dict[str, Any]]) -> tuple[tuple[tuple[str, Any], ...], tuple[tuple[str, Any], ...]]: - """ - Преобразует params_list в хешируемый ключ для кэша. - """ - operator_settings, system_settings = params_list - - # Преобразуем словари в отсортированные кортежи пар ключ-значение - operator_tuple = frozenset((key, value) - for key, value in operator_settings.items() - if str(key) in self._OptAlgorithm_operator_params) - - system_tuple = frozenset((key, value) - for key, value in system_settings.items() - if str(key) in self._OptAlgorithm_system_params) - return (operator_tuple, system_tuple) - - def _build_graphic_passport(self, dataframe: pd.DataFrame) -> GraphicPassport: - - if dataframe is not None: - events, point_quantity = self._filter_events(dataframe["time"], dataframe) - if point_quantity == 0: - return [] - else: - events = None - key = list(self._params[0].keys())[0] - point_quantity = len(self._params[0][key]) - - graphic_passport = GraphicPassport() - graphic_passport.dataframe = dataframe - graphic_passport.points_pocket = [] - system_settings = {key: value[0] for key, value in self._params[1].items()} - graphic_passport.useful_data = self._form_graphic_useful_data(system_settings) - - for i in range(point_quantity): - operator_settings = self._get_operator_settings_part(i) - params_list = [operator_settings, system_settings] - - point_passport = PointPassport() - point_passport.ideal_data = self._form_point_ideal_data(params_list) - point_passport.useful_data = self._form_point_useful_data(operator_settings) - point_passport.timeframe, point_passport.events = self._form_point_events(events, i) - - graphic_passport.points_pocket.append(point_passport) - return graphic_passport - - def _form_graphic_useful_data(self, system_settings:dict) -> dict: - tesla_time = sum(self._params[0].get("Tesla summary time", [])) - useful_data = {"tesla_time": tesla_time, - "range_ME": system_settings["Range ME, mm"], - "k_hardness": system_settings["k_hardness_1"] - } - return useful_data - - def _form_point_useful_data(self, operator_settings:dict) -> dict: - useful_data = {"thickness": operator_settings["object_thickness"], - "L2": operator_settings["distance_l_2"], - "force": operator_settings["force_target"]} - return useful_data - - def _form_point_ideal_data(self, params_list:list) -> dict: - cache_key = self._generate_cache_key(params_list) - ideal_data = self._ideal_data_cashe.get(cache_key, - self._build_ideal_data(idealDataBuilder=IdealDataBuilder, params=params_list)) - self._ideal_data_cashe[cache_key] = ideal_data - return ideal_data - - def _get_operator_settings_part(self, idx:int) -> dict: - operator_settings = { - key: (value[idx] if idx < len(value) else value[0]) - for key, value in self._params[0].items() - } - return operator_settings - - def _form_point_events(self, events:dict, idx) -> list: - timeframe, point_events = None, None - if events is not None: - idx_shift = idx+1 if events[self._stages[-1]][0][0] == 0 else idx - timeframe = [events[self._stages[0]][0][idx], events[self._stages[-1]][1][idx_shift]] - point_events = {key: [value[0][idx], value[1][idx]] for key, value in events.items()} - return timeframe, point_events \ No newline at end of file + return pd.DataFrame(data) \ No newline at end of file diff --git a/src/gui/main_gui.py b/src/gui/main_gui.py index deb2f45..a834d74 100644 --- a/src/gui/main_gui.py +++ b/src/gui/main_gui.py @@ -1,7 +1,7 @@ from PyQt5 import QtWidgets from PyQt5.QtCore import Qt, pyqtSignal -from base.base import BaseMainWindow +from base.base import BaseMainWindow, Settings from gui.start_widget import (CustomMenuBar, CustomStatusBar, StartWidget, CustomTabWidget, RaportWidget, SeekingWidget) @@ -13,7 +13,7 @@ from gui.report_gui import ReportSettings class MainWindow(BaseMainWindow): signal_mode = pyqtSignal(int) - signal_settings = pyqtSignal(list) + signal_settings = pyqtSignal(Settings) signal_replot_all = pyqtSignal() signal_open_file = pyqtSignal(str) signal_save_file = pyqtSignal(list) @@ -22,6 +22,30 @@ class MainWindow(BaseMainWindow): super().__init__() self._init_startUI() + def show_plot_tabs(self, plot_widgets: list[QtWidgets.QWidget]) -> None: + for plot_widget in plot_widgets: + self._tab_widget.create_tab(plot_widget) + + tab_count = self._tab_widget.count() + if tab_count > 10: + for i in range(0, tab_count-10): + self._tab_widget.close_tab(i) + + self.status_widget.set_note("Done!") + self.status_widget.set_progress(100) + + def keyPressEvent(self, a0) -> None: + if a0.key() == Qt.Key_F5: + tab_count = self._tab_widget.count() + for i in range(0, tab_count): + self._tab_widget.close_tab(i) + + def closeEvent(self, a0): + self.operSettings.close() + self.sysSettings.close() + self.repSettings.close() + super().closeEvent(a0) + def _init_startUI(self) -> None: self._init_settings() self._init_menu() @@ -91,34 +115,11 @@ class MainWindow(BaseMainWindow): self.status_widget.set_mode("online mode") self.signal_mode.emit(2) - def show_plot_tabs(self, plot_widgets: list[QtWidgets.QWidget]) -> None: - for plot_widget in plot_widgets: - self._tab_widget.create_tab(plot_widget) - - tab_count = self._tab_widget.count() - if tab_count > 10: - for i in range(0, tab_count-10): - self._tab_widget.close_tab(i) - - self.status_widget.set_note("Done!") - self.status_widget.set_progress(100) - - def keyPressEvent(self, a0) -> None: - if a0.key() == Qt.Key_F5: - tab_count = self._tab_widget.count() - for i in range(0, tab_count): - self._tab_widget.close_tab(i) - - def closeEvent(self, a0): - self.operSettings.close() - self.sysSettings.close() - self.repSettings.close() - super().closeEvent(a0) - def _transfer_settings(self) -> None: - operator_params = self.operSettings.getParams() - system_params = self.sysSettings.getParams() - self.signal_settings.emit([operator_params, system_params]) + settings = Settings() + settings.operator = self.operSettings.getParams() + settings.system = self.sysSettings.getParams() + self.signal_settings.emit(Settings) def _upd_settings(self) -> None: self._transfer_settings() diff --git a/src/gui/plotter.py b/src/gui/plotter.py index cacc0ac..85df5cd 100644 --- a/src/gui/plotter.py +++ b/src/gui/plotter.py @@ -3,20 +3,46 @@ import traceback import sys from typing import Optional, Any -from PyQt5.QtWidgets import (QWidget, QVBoxLayout, - QHBoxLayout, QLabel, - QGraphicsRectItem, QSpacerItem, - QSizePolicy) +from PyQt5.QtWidgets import ( + QWidget, + QVBoxLayout, + QHBoxLayout, + QLabel, + QGraphicsRectItem, + QSpacerItem, + QSizePolicy + ) from PyQt5.QtCore import Qt from loguru import logger import pyqtgraph as pg import pandas as pd -from base.base import BasePlotWidget +from base.base import BasePlotWidget, GraphicPassport class PlotWidget(BasePlotWidget): + def build(self, data: list[GraphicPassport]) -> None: + """ + Создает набор виджетов по предоставленному списку данных. + """ + try: + # TODO: Про инициализацию атрибутов класса где-то уже писал. На всякий случай: она делается в конструкторе. + self._datalen = len(data) + widgets_datapack = [self._build_widget(data_sample) for self._datastep, data_sample in enumerate(data)] + except: + # TODO: Добавить конкретные исключения. + tb = sys.exc_info()[2] + tbinfo = traceback.format_tb(tb)[0] + pymsg = "Traceback info:\n" + tbinfo + "\nError Info:\n" + str(sys.exc_info()[1]) + # TODO: Логи должны быть понятны обычному пользователю. Traceback что-то там - это не информативно и может + # быть непонятно, поэтому пишем осознанные сообщения. Если надо, то создаем собственные + # классы исключений. (Касается всего проекта - уже не первый раз натыкаюсь на это.) + logger.error(pymsg) + widgets_datapack = [QLabel(pymsg)] + finally: + self._mediator.notify(self, widgets_datapack) + def _downsample_data(self, x, y, max_points=5000): # TODO: Данный метод статичный. Для обозначения подобных методов используется декоратор @staticmethod. # TODO: Какой тип данных у 'x'? Какой тип данных у 'y'? Надо добавить аннотации типов. @@ -273,16 +299,13 @@ class PlotWidget(BasePlotWidget): qt_items["real to ideal performance"] = kdip_label - def _build_widget(self, data: list[Any]) -> QWidget: - # TODO: Либо передавать в метод 3 аргумента с аннотацией типа каждого, либо передавать - # NamedTuple / dataclass. + def _build_widget(self, graphic_passport: GraphicPassport) -> QWidget: # TODO: Исходя из сигнатуры метода, он создает и возвращает виджет с графиками. # Простыня "result_widget, reg_items, curve_items, qt_items" не похожа на виджет с графиком. # TODO: Данный метод должен содержать только ту логику, которая относится непосредственно к графикам. # Все остальное - вынести. Оставшийся метод декомпозировать - очень трудно вообще понять, что тут происходит. """ Собирает графический виджет для одного набора данных. - Параметр `data` предполагается списком: [dataframe, points_pocket, useful_data]. """ result_widget = QWidget() result_layout = QVBoxLayout(result_widget) @@ -290,19 +313,17 @@ class PlotWidget(BasePlotWidget): reg_items = {"real":{}, "ideal":{}} curve_items = {"real":{}, "ideal":{}} qt_items = {} - - dataframe, points_pocket, useful_data = data - tesla_time = useful_data["tesla_time"] - range_ME = useful_data["range_ME"] - k_hardness = useful_data["k_hardness"] + dataframe = graphic_passport.dataframe + tesla_time = graphic_passport.useful_data["tesla_time"] + range_ME = graphic_passport.useful_data["range_ME"] + k_hardness = graphic_passport.useful_data["k_hardness"] dat_is_none = dataframe is None - widget_steps = len(self._plt_channels) if not dat_is_none: dataframe_headers = dataframe.columns.tolist() - point_steps = len(points_pocket) + point_steps = len(graphic_passport.points_pocket) else: point_steps = 1 for widget_num, (channel, description) in enumerate(self._plt_channels.items()): @@ -323,7 +344,7 @@ class PlotWidget(BasePlotWidget): ) # Итерация по точкам - for cur_point, point_data in enumerate(points_pocket): + for cur_point, point_data in enumerate(graphic_passport.points_pocket): # point_data структура: [point_timeframe, ideal_data, point_events, useful_p_data] point_timeframe, ideal_dat, point_events, useful_p_data = point_data ideal_data = copy.deepcopy(ideal_dat) @@ -384,13 +405,13 @@ class PlotWidget(BasePlotWidget): # Добавляем идеальные стадии и идеальные сигналы if settings["ideals"]: - is_last_point = (cur_point == len(points_pocket) - 1) + is_last_point = (cur_point == len(graphic_passport.points_pocket) - 1) self._add_ideal_stage_regions(plot_item, ideal_data, point_events, reg_items, 100) self._add_ideal_signals(plot_item, legend, ideal_data, point_events, description["Ideal_signals"], curve_items, is_last_point) # Подсчёт производительности if settings["performance"]: - is_last_point = (cur_point == len(points_pocket) - 1) + is_last_point = (cur_point == len(graphic_passport.points_pocket) - 1) if is_last_point: if not dat_is_none: TWC_delta = sum([point_events[stage][1] - point_events[stage][0] for stage in ["Closing", "Squeeze", "Welding"]]) @@ -436,34 +457,6 @@ class PlotWidget(BasePlotWidget): result_layout.addWidget(plot_layout) return result_widget, reg_items, curve_items, qt_items - def build(self, data: list[list[Any]]) -> None: - """ - Создает набор виджетов по предоставленному списку данных. - Предполагается, что data — это список элементов вида: - [ - [dataframe, points_pocket, useful_data], - [dataframe, points_pocket, useful_data], - ... - ] - """ - # TODO: Про входные аргументы см. выше. - try: - # TODO: Про инициализацию атрибутов класса где-то уже писал. На всякий случай: она делается в конструкторе. - self._datalen = len(data) - widgets_datapack = [self._build_widget(data_sample) for self._datastep, data_sample in enumerate(data)] - except: - # TODO: Добавить конкретные исключения. - tb = sys.exc_info()[2] - tbinfo = traceback.format_tb(tb)[0] - pymsg = "Traceback info:\n" + tbinfo + "\nError Info:\n" + str(sys.exc_info()[1]) - # TODO: Логи должны быть понятны обычному пользователю. Traceback что-то там - это не информативно и может - # быть непонятно, поэтому пишем осознанные сообщения. Если надо, то создаем собственные - # классы исключений. (Касается всего проекта - уже не первый раз натыкаюсь на это.) - logger.error(pymsg) - widgets_datapack = [QLabel(pymsg)] - finally: - self._mediator.notify(self, widgets_datapack) - def _update_status(self, widgsteps:int, pointsteps:int, cur_widg:int, cur_point:int): if self._datalen: sycle_start = self._datastep/self._datalen*100 + 1 @@ -477,7 +470,7 @@ class PlotWidget(BasePlotWidget): progress = sycle_start + period2*cur_widg + period3*cur_point # TODO: см. модуль mediator.py - self._mediator.update_status(progress) + self._controller.update_progress(progress) \ No newline at end of file diff --git a/src/main.py b/src/main.py index 37188c4..636a8a3 100644 --- a/src/main.py +++ b/src/main.py @@ -11,7 +11,6 @@ from gui.plotter import PlotWidget from controller.controller import Controller from src.controller.passport_former import PassportFormer -# TODO: Именование модулей: lowercase / snake_case. def main(): pg.setConfigOptions(useOpenGL=False, antialias=False)