Compare commits
No commits in common. "34b8b4962453020e4fef77e0d502419662d92088" and "78fa10d897185a5412387f9dcf7dccab40895ad4" have entirely different histories.
34b8b49624
...
78fa10d897
@ -1,4 +1,3 @@
|
||||
loguru==0.7.2
|
||||
numpy==2.1.3
|
||||
pandas==2.2.3
|
||||
PyQt5==5.15.11
|
||||
|
||||
183
src/base/base.py
183
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:
|
||||
...
|
||||
@ -3,4 +3,4 @@ from typing import NamedTuple
|
||||
|
||||
class ConfigSchema(NamedTuple):
|
||||
trace_storage_path: str
|
||||
monitor_update_period: int | float
|
||||
monitor_update_period: int
|
||||
|
||||
@ -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)
|
||||
@ -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)
|
||||
@ -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)
|
||||
@ -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 = []
|
||||
|
||||
@ -1,7 +1,4 @@
|
||||
from datetime import datetime as dt
|
||||
|
||||
from PyQt5 import QtWidgets
|
||||
from PyQt5.QtCore import Qt
|
||||
|
||||
from gui.widgets import mainForm
|
||||
|
||||
@ -10,16 +7,3 @@ class MainWindow(QtWidgets.QWidget, mainForm.Ui_Form):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
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()
|
||||
|
||||
@ -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)
|
||||
16
src/main.py
16
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_())
|
||||
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user