fix: исправил падение приложения в случае отсутствия меток начала и конца этапа

requirements.txt

@@ -1,3 +1,4 @@
+loguru==0.7.2
 numpy==2.1.3
 pandas==2.2.3
 PyQt5==5.15.11

src/base/base.py

@@ -0,0 +1,183 @@
+from __future__ import annotations
+
+import os
+from typing import Optional, Union
+
+import pandas as pd
+from PyQt5.QtCore import QThread, QObject, QTimer
+from PyQt5.QtWidgets import QWidget
+
+
+class BaseMediator:
+    def __init__(self,
+                 monitor: BaseDirectoryMonitor,
+                 converter: BaseDataConverter,
+                 plot: BasePlotWidget,
+                 controller: BaseController):
+        self._monitor = monitor
+        self._monitor.mediator = self
+        self._converter = converter
+        self._converter.mediator = self
+        self._plot = plot
+        self._plot.mediator = self
+        self._controller = controller
+
+    def notify(self,
+               source: Union[BaseDirectoryMonitor, BaseDataConverter, BasePlotWidget],
+               data: Union[list[str], list[pd.DataFrame], list[QWidget]]):
+        ...
+
+
+class BaseDirectoryMonitor:
+
+    update_timer = QTimer()
+
+    def __init__(self,
+                 directory_path: str,
+                 update_time: int,
+                 mediator: Optional[BaseMediator] = None):
+        super().__init__()
+
+        self._directory_path = directory_path
+        self._update_time = update_time
+        self._mediator = mediator
+
+        self._files: list[str] = []
+
+        self._init_state()
+
+    @property
+    def directory_path(self) -> str:
+        return self._directory_path
+
+    @property
+    def update_time(self) -> int:
+        return self._update_time
+
+    @property
+    def files(self) -> list[str]:
+        return self._files
+
+    @property
+    def mediator(self) -> BaseMediator:
+        return self._mediator
+
+    @mediator.setter
+    def mediator(self, mediator: BaseMediator) -> None:
+        self._mediator = mediator
+
+    def _init_state(self):
+        files = os.listdir(self._directory_path)
+        self._files = files
+
+    def start(self):
+        self.update_timer.start(self._update_time)
+
+class BaseDataConverter:
+    def __init__(self, mediator: Optional[BaseMediator] = None):
+        self._mediator = mediator
+
+    @property
+    def mediator(self) -> BaseMediator:
+        return self._mediator
+
+    @mediator.setter
+    def mediator(self, mediator: BaseMediator) -> None:
+        self._mediator = mediator
+
+    def convert_data(self, files: list[str]) -> None:
+        ...
+
+
+class BasePlotWidget:
+    def __init__(self, mediator: Optional[BaseMediator] = None):
+        super().__init__()
+        self._mediator = mediator
+
+        self._stages = [
+            "Relief",
+            "Closing",
+            "Squeeze",
+            "Welding"
+        ]
+
+        self._stage_colors = {
+            "Closing": [208, 28, 31, 100],
+            "Squeeze": [45, 51, 89, 150],
+            "Welding": [247, 183, 24, 100],
+            "Relief": [0, 134, 88, 100]
+        }
+
+        self._plt_channels = {
+            "Electrode Force, N & Welding Current, kA": {
+                "Settings": {
+                    "zoom": False,
+                    "stages": True
+                },
+                "Signals": [
+                    {
+                        "name": "Electrode Force, N ME",
+                        "pen": 'r',
+                    },
+                    {
+                        "name": "Electrode Force, N FE",
+                        "pen": 'w',
+                    },
+                    {
+                        "name": "Welding Current ME",
+                        "pen": "y",
+                    }
+                ]
+            },
+            "Electrode Force, N": {
+                "Settings": {
+                    "zoom": True,
+                    "stages": False
+                },
+                "Signals": [
+                    {
+                        "name": "Electrode Force, N ME",
+                        "pen": 'r',
+                    },
+                    {
+                        "name": "Electrode Force, N FE",
+                        "pen": 'w',
+                    }
+                ]
+            },
+            "Electrode Speed, mm/s": {
+                "Settings": {
+                    "zoom": False,
+                    "stages": True
+                },
+                "Signals": [
+                    {
+                        "name": "Rotor Speed, mm/s ME",
+                        "pen": 'r',
+                        "zoom": False
+                    },
+                    {
+                        "name": "Rotor Speed, mm/s FE",
+                        "pen": 'w',
+                        "zoom": False
+                    }
+                ]
+            },
+        }
+
+    @property
+    def mediator(self) -> BaseMediator:
+        return self._mediator
+
+    @mediator.setter
+    def mediator(self, mediator: BaseMediator) -> None:
+        self._mediator = mediator
+
+    def build(self, data: list[pd.DataFrame]) -> list[QWidget]:
+        ...
+
+
+class BaseController(QObject):
+
+    def send_widgets(self, widgets: list[QWidget]) -> None:
+        ...

src/cfg/schema.py

@@ -3,4 +3,4 @@ from typing import NamedTuple
 
 class ConfigSchema(NamedTuple):
     trace_storage_path: str
-    monitor_update_period: int
+    monitor_update_period: int | float

src/controller/controller.py

@@ -0,0 +1,12 @@
+from PyQt5.QtWidgets import QWidget
+from PyQt5.QtCore import pyqtSignal
+
+from base.base import BaseController
+
+
+class Controller(BaseController):
+
+    signal_widgets = pyqtSignal(list)
+
+    def send_widgets(self, widgets: list[QWidget]) -> None:
+        self.signal_widgets.emit(widgets)

src/controller/converter.py

@@ -0,0 +1,20 @@
+import pandas as pd
+
+#FIXME: костыль для выключения предупреждения "replace deprecated". Потом надо поправить.
+pd.set_option('future.no_silent_downcasting', True)
+
+from base.base import BaseDataConverter
+
+
+class DataConverter(BaseDataConverter):
+
+    @staticmethod
+    def _replace_bool(dataframe: pd.DataFrame) -> pd.DataFrame:
+        bool_columns = dataframe.columns[dataframe.isin([True, False]).any()]
+        dataframe[bool_columns] = dataframe[bool_columns].replace({True: 1, False: 0})
+        return dataframe
+
+    def convert_data(self, files: list[str]) -> None:
+        dataframes = [pd.read_csv(file) for file in files]
+        converted_dataframes = list(map(self._replace_bool, dataframes))
+        self._mediator.notify(self, converted_dataframes)

src/controller/mediator.py

@@ -0,0 +1,21 @@
+import pandas as pd
+
+from typing import Union
+from PyQt5.QtWidgets import QWidget
+
+from base.base import BaseMediator, BaseDirectoryMonitor, BaseDataConverter, BasePlotWidget
+
+
+class Mediator(BaseMediator):
+
+    def notify(self,
+               source: Union[BaseDirectoryMonitor, BaseDataConverter, BasePlotWidget],
+               data: Union[list[str], list[pd.DataFrame], list[QWidget]]):
+        if issubclass(source.__class__, BaseDirectoryMonitor):
+            self._converter.convert_data(data)
+
+        if issubclass(source.__class__, BaseDataConverter):
+            self._plot.build(data)
+
+        if issubclass(source.__class__, BasePlotWidget):
+            self._controller.send_widgets(data)

src/controller/monitor.py

@@ -0,0 +1,28 @@
+from time import sleep
+import os
+
+from loguru import logger
+
+from base.base import BaseDirectoryMonitor
+
+
+class DirectoryMonitor(BaseDirectoryMonitor):
+
+    def _init_state(self):
+        files = os.listdir(self._directory_path)
+        self._files = files
+
+        self.update_timer.timeout.connect(self._monitor)
+        logger.info("Monitor initiated!")
+
+    def _monitor(self):
+        files = os.listdir(self._directory_path)
+        new_files = sorted(list(map(lambda x: os.path.join(self._directory_path, x),
+                                    filter(lambda x: x not in self._files, files))))
+        if new_files:
+            logger.info(f"New files detected: {new_files}")
+            self._mediator.notify(self, new_files)
+            self._files = files
+        if not files:
+            self._files = []
+

src/gui/mainGui.py

@@ -1,4 +1,7 @@
+from datetime import datetime as dt
+
 from PyQt5 import QtWidgets
+from PyQt5.QtCore import Qt
 
 from gui.widgets import mainForm
 
@@ -6,4 +9,17 @@ from gui.widgets import mainForm
 class MainWindow(QtWidgets.QWidget, mainForm.Ui_Form):
     def __init__(self):
         super().__init__()
-        self.setupUi(self)
+        self.setupUi(self)
+
+    def show_plot_tabs(self, plot_widgets: list[QtWidgets.QWidget]) -> None:
+        for plot_widget in plot_widgets:
+            tab = QtWidgets.QWidget()
+            grid = QtWidgets.QGridLayout()
+            grid.addWidget(plot_widget)
+            tab.setLayout(grid)
+            self.tabWidget.addTab(tab, "SF_trace_" + dt.now().strftime('%Y_%m_%d-%H_%M_%S'))
+            self.tabWidget.setCurrentWidget(tab)
+
+    def keyPressEvent(self, a0):
+        if a0.key() == Qt.Key_F5:
+            self.tabWidget.clear()

src/gui/widgets/plot.py

@@ -0,0 +1,64 @@
+from typing import Optional
+
+import pandas as pd
+from PyQt5.QtWidgets import QWidget, QVBoxLayout
+import pyqtgraph as pg
+import numpy as np
+
+from base.base import BasePlotWidget
+
+
+class PlotWidget(BasePlotWidget):
+
+    def _create_stage_region(self,
+                             stage: str,
+                             times: pd.Series,
+                             dataframe: pd.DataFrame) -> Optional[pg.LinearRegionItem]:
+        stage_diff = np.diff(dataframe[stage])
+        start_index = np.where(stage_diff == 1)[0]
+        finish_index = np.where(stage_diff == -1)[0]
+        if start_index:
+            start_timestamp = times[start_index[0]]
+            finish_timestamp = times[finish_index[0]] if finish_index else times[len(times) - 1]
+            region = pg.LinearRegionItem([start_timestamp, finish_timestamp], movable=False)
+            region.setBrush(pg.mkBrush(self._stage_colors[stage]))
+            return region
+        return None
+
+    def _build_widget(self, dataframe: pd.DataFrame) -> QWidget:
+        widget = QWidget()
+        layout = QVBoxLayout()
+
+        time_axis = dataframe["time"]
+        dataframe_headers = dataframe.columns.tolist()
+
+        for channel, description in self._plt_channels.items():
+            plot_widget = pg.PlotWidget(title=channel)
+            plot_widget.showGrid(x=True, y=True)
+            legend = pg.LegendItem((80, 60), offset=(70, 20))
+            legend.setParentItem(plot_widget.graphicsItem())
+            settings = description["Settings"]
+
+            if settings["stages"] and all([stage in dataframe_headers for stage in self._stages]):
+                for stage in self._stages:
+                    region = self._create_stage_region(stage, time_axis, dataframe)
+                    if region:
+                        plot_widget.addItem(region)
+
+            for signal in description["Signals"]:
+                if signal["name"] in dataframe_headers:
+                    plot = plot_widget.plot(time_axis, dataframe[signal["name"]], pen=signal["pen"])
+                    legend.addItem(plot, signal["name"])
+                    if settings["zoom"] and max(time_axis) < 5.0:
+                        max_value = max(dataframe[signal["name"]])
+                        plot_widget.setYRange(max_value - 200, max_value)
+                        plot_widget.setInteractive(False)
+
+            layout.addWidget(plot_widget)
+
+        widget.setLayout(layout)
+        return widget
+
+    def build(self, data: list[pd.DataFrame]) -> None:
+        widgets = [self._build_widget(data_sample) for data_sample in data]
+        self._mediator.notify(self, widgets)

src/main.py

@@ -5,6 +5,11 @@ from os import path
 
 from gui.mainGui import MainWindow
 from cfg.schema import ConfigSchema
+from controller.monitor import DirectoryMonitor
+from controller.mediator import Mediator
+from controller.converter import DataConverter
+from gui.widgets import plot
+from controller.controller import Controller
 
 
 def read_json(filepath: str) -> dict:
@@ -19,10 +24,19 @@ def read_json(filepath: str) -> dict:
 def main():
     app = QtWidgets.QApplication(sys.argv)
     config = ConfigSchema(**read_json("config/config.json"))
-
+    monitor = DirectoryMonitor(config.trace_storage_path, config.monitor_update_period)
+    data_converter = DataConverter()
+    plot_widget_builder = plot.PlotWidget()
+    controller = Controller()
+    mediator = Mediator(monitor, data_converter, plot_widget_builder, controller)
+    monitor.start()
     window = MainWindow()
+
     window.show()
 
+    controller.signal_widgets.connect(window.show_plot_tabs)
+
+
     sys.exit(app.exec_())