42 lines
1.4 KiB
Python
42 lines
1.4 KiB
Python
import pandas as pd
|
|
import time
|
|
|
|
from typing import Union
|
|
from PyQt5.QtWidgets import QWidget
|
|
|
|
from utils.base.base import (BaseMediator, BaseDirectoryMonitor,
|
|
BaseDataConverter, BasePlotWidget,
|
|
BasePointPassportFormer)
|
|
|
|
|
|
class Mediator(BaseMediator):
|
|
|
|
def notify(self,
|
|
source: Union[BaseDirectoryMonitor, BaseDataConverter, BasePointPassportFormer, BasePlotWidget],
|
|
data: Union[list[str], list[pd.DataFrame], list[list], list[QWidget]]):
|
|
|
|
if issubclass(source.__class__, BaseDirectoryMonitor):
|
|
self.update_status("CSV found! Calculating...")
|
|
self._converter.convert_data(data)
|
|
|
|
if issubclass(source.__class__, BaseDataConverter):
|
|
self.update_status(0.5)
|
|
self._passportFormer.form_passports(data)
|
|
|
|
if issubclass(source.__class__, BasePointPassportFormer):
|
|
self.update_status(1)
|
|
self._plot.build(data)
|
|
|
|
if issubclass(source.__class__, BasePlotWidget):
|
|
self.update_status(100)
|
|
self._controller.send_widgets(data)
|
|
|
|
def update_settings(self, settings: list[dict]):
|
|
self._monitor.update_settings(settings)
|
|
self._passportFormer.update_settings(settings)
|
|
|
|
def update_status(self, msg: Union[str, float]) -> None:
|
|
self._controller.update_status(msg)
|
|
|
|
|