chore: управление открываемыми файлами вынесено в отдельный класс
This commit is contained in:
Andrew
parent
6999f12354
commit
16910133ac
@ -19,14 +19,12 @@ from utils.qt_settings import dark_style
|
||||
|
||||
class BaseMediator:
|
||||
def __init__(self,
|
||||
monitor: BaseDirectoryMonitor,
|
||||
converter: BaseDataConverter,
|
||||
# TODO: passport_former
|
||||
passportFormer: BasePointPassportFormer,
|
||||
plot: BasePlotWidget,
|
||||
controller: BaseController):
|
||||
self._monitor = monitor
|
||||
self._monitor.mediator = self
|
||||
controller: BaseController,
|
||||
file_manager: BaseFileManager):
|
||||
self._converter = converter
|
||||
self._converter.mediator = self
|
||||
self._passportFormer = passportFormer
|
||||
@ -35,6 +33,8 @@ class BaseMediator:
|
||||
self._plot.mediator = self
|
||||
self._controller = controller
|
||||
self._controller.mediator = self
|
||||
self._file_manager = file_manager
|
||||
self._file_manager.mediator = self
|
||||
|
||||
def notify(self,
|
||||
source: Union[BaseDirectoryMonitor, BaseDataConverter, BasePointPassportFormer, BasePlotWidget],
|
||||
@ -52,14 +52,13 @@ class BaseDirectoryMonitor:
|
||||
update_timer = QTimer()
|
||||
|
||||
def __init__(self,
|
||||
mediator: Optional[BaseMediator] = None):
|
||||
file_manager: Optional[BaseFileManager] = None):
|
||||
super().__init__()
|
||||
self._directory_path = None
|
||||
self._update_time = None
|
||||
self.isActive = False
|
||||
self._files = []
|
||||
self._mediator = mediator
|
||||
|
||||
self._file_manager = file_manager
|
||||
|
||||
@property
|
||||
def directory_path(self) -> str:
|
||||
@ -74,14 +73,14 @@ class BaseDirectoryMonitor:
|
||||
return self._files
|
||||
|
||||
@property
|
||||
def mediator(self) -> BaseMediator:
|
||||
return self._mediator
|
||||
def file_manager(self) -> BaseFileManager:
|
||||
return self._file_manager
|
||||
|
||||
@mediator.setter
|
||||
def mediator(self, mediator: BaseMediator) -> None:
|
||||
self._mediator = mediator
|
||||
@file_manager.setter
|
||||
def file_manager(self, file_manager: BaseFileManager) -> None:
|
||||
self._file_manager = file_manager
|
||||
|
||||
def _init_state(self):
|
||||
def init_state(self):
|
||||
files = os.listdir(self._directory_path)
|
||||
self._files = files
|
||||
|
||||
@ -92,6 +91,10 @@ class BaseDirectoryMonitor:
|
||||
def stop(self):
|
||||
self.isActive = False
|
||||
self.update_timer.stop()
|
||||
|
||||
def pause(self):
|
||||
self.update_timer.stop()
|
||||
|
||||
# TODO: Почему монитор директорий обновляет какие-то настройки, графики и т.д.? Помним про
|
||||
# принцип единичной ответственности.
|
||||
def update_settings(self, data: list[dict]) -> None:
|
||||
@ -361,6 +364,10 @@ class BasePlotWidget:
|
||||
|
||||
class BaseController(QObject):
|
||||
|
||||
def __init__(self,
|
||||
mediator: Optional[BaseMediator] = None):
|
||||
self._mediator = mediator
|
||||
|
||||
def send_widgets(self, widgets: list[QWidget]) -> None:
|
||||
...
|
||||
|
||||
@ -383,6 +390,49 @@ class BaseController(QObject):
|
||||
|
||||
def update_status(self, msg: Union[str, float]) -> None:
|
||||
...
|
||||
|
||||
@property
|
||||
def mediator(self) -> BaseMediator:
|
||||
return self._mediator
|
||||
|
||||
@mediator.setter
|
||||
def mediator(self, mediator: BaseMediator) -> None:
|
||||
self._mediator = mediator
|
||||
|
||||
|
||||
class BaseFileManager:
|
||||
|
||||
def __init__(self,
|
||||
mediator: Optional[BaseMediator] = None,
|
||||
monitor: Optional[BaseDirectoryMonitor] = None):
|
||||
self._mediator = mediator
|
||||
self._monitor = monitor
|
||||
self._paths_library = set()
|
||||
|
||||
@property
|
||||
def paths_library(self) -> set:
|
||||
return self._paths_library
|
||||
|
||||
@property
|
||||
def mediator(self) -> BaseMediator:
|
||||
return self._mediator
|
||||
|
||||
@mediator.setter
|
||||
def mediator(self, mediator: BaseMediator) -> None:
|
||||
self._mediator = mediator
|
||||
|
||||
@property
|
||||
def monitor(self) -> BaseDirectoryMonitor:
|
||||
return self._monitor
|
||||
|
||||
@monitor.setter
|
||||
def monitor(self, monitor: BaseDirectoryMonitor) -> None:
|
||||
self._monitor = monitor
|
||||
|
||||
def add_new_paths(self, paths:list[str]):
|
||||
...
|
||||
|
||||
|
||||
|
||||
class BaseIdealDataBuilder(OptAlgorithm):
|
||||
|
||||
|
||||
@ -2,17 +2,19 @@ import os
|
||||
|
||||
from loguru import logger
|
||||
|
||||
from base.base import BaseDirectoryMonitor
|
||||
from base.base import BaseDirectoryMonitor, BaseFileManager
|
||||
|
||||
|
||||
class DirectoryMonitor(BaseDirectoryMonitor):
|
||||
|
||||
def _init_state(self):
|
||||
def init_state(self):
|
||||
if not os.path.exists(self._directory_path):
|
||||
logger.error(f"Путь {self._directory_path} не существует.")
|
||||
raise FileNotFoundError(f"Путь {self._directory_path} не существует.")
|
||||
|
||||
self._files = [
|
||||
os.path.join(self._directory_path, file)
|
||||
for file in os.listdir(self._directory_path)
|
||||
# TODO: FileNotFoundError: [Errno 2] No such file or directory: 'D:/downloads/a22'
|
||||
# при выборе real time mode.
|
||||
if file.lower().endswith('.csv')
|
||||
]
|
||||
|
||||
@ -28,50 +30,53 @@ class DirectoryMonitor(BaseDirectoryMonitor):
|
||||
new_files = sorted(list(filter(lambda x: x not in self._files, current_files)))
|
||||
if new_files:
|
||||
logger.info(f"New files detected: {new_files}")
|
||||
self._mediator.notify(self, new_files)
|
||||
self._file_manager.add_new_paths(new_files)
|
||||
self._files = current_files
|
||||
if not current_files:
|
||||
self._files = []
|
||||
|
||||
# TODO: Почему монитор директории обновляет какие-то настройки, графики, стартует что-то?
|
||||
# Это не должно быть в его зоне ответственности.
|
||||
# TODO: Может стоит сделать какой-то класс, который будет заниматься обновлениями?
|
||||
def update_settings(self, data: list[dict]) -> None:
|
||||
if self.isActive:
|
||||
self.stop()
|
||||
_, system_params = data
|
||||
self._directory_path = system_params['trace_storage_path'][0]
|
||||
self._update_time = system_params['monitor_update_period'][0]
|
||||
|
||||
if not os.path.exists(self._directory_path):
|
||||
logger.error(f"Путь {self._directory_path} не существует.")
|
||||
raise FileNotFoundError(f"Путь {self._directory_path} не существует.")
|
||||
else:
|
||||
self._init_state()
|
||||
self.start()
|
||||
else:
|
||||
_, system_params = data
|
||||
self._directory_path = system_params['trace_storage_path'][0]
|
||||
self._update_time = system_params['monitor_update_period'][0]
|
||||
|
||||
class FileManager(BaseFileManager):
|
||||
|
||||
def update_plots(self):
|
||||
if self._files is not None:
|
||||
self._mediator.notify(self, self._files)
|
||||
if self._paths_library is not None:
|
||||
self._mediator.notify(self, list(self._paths_library))
|
||||
|
||||
def open_custom_file(self, path:str):
|
||||
self._paths_library.add(path)
|
||||
self._mediator.notify(self, [path if path else ''])
|
||||
|
||||
def start_raport_mode(self) -> None:
|
||||
self._monitor.stop()
|
||||
self._paths_library.clear()
|
||||
self._paths_library.add('')
|
||||
self._mediator.notify(self, list(self._paths_library))
|
||||
|
||||
def start_seeking_mode(self) -> None:
|
||||
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))
|
||||
|
||||
def custom_csv_extract_only(self, path: str):
|
||||
self._files.append(path)
|
||||
if path is not None:
|
||||
self._mediator.notify(self, [path])
|
||||
else:
|
||||
self._mediator.notify(self, [''])
|
||||
|
||||
def start_raport(self) -> None:
|
||||
self.stop()
|
||||
self._files = ['']
|
||||
self._mediator.notify(self, [''])
|
||||
|
||||
|
||||
def start_seeking(self) -> None:
|
||||
self._init_state()
|
||||
self.start()
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user