requirements.txt

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

src/base/base.py

@@ -1,183 +0,0 @@
-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 | float
+    monitor_update_period: int

src/controller/controller.py

@@ -1,12 +0,0 @@
-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

@@ -1,20 +0,0 @@
-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

@@ -1,21 +0,0 @@
-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

@@ -1,28 +0,0 @@
-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,7 +1,4 @@
-from datetime import datetime as dt
-
 from PyQt5 import QtWidgets
-from PyQt5.QtCore import Qt
 
 from gui.widgets import mainForm
 
@@ -9,17 +6,4 @@ 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

@@ -1,64 +0,0 @@
-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,11 +5,6 @@ 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:
@@ -24,19 +19,10 @@ 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_())