77 lines
2.5 KiB
Python
77 lines
2.5 KiB
Python
import os
|
|
from loguru import logger
|
|
|
|
from utils.base.base import BaseDirectoryMonitor
|
|
|
|
|
|
class DirectoryMonitor(BaseDirectoryMonitor):
|
|
|
|
def _init_state(self):
|
|
|
|
self._files = [
|
|
os.path.join(self._directory_path, file)
|
|
for file in os.listdir(self._directory_path)
|
|
if file.lower().endswith('.csv')
|
|
]
|
|
|
|
self.update_timer.timeout.connect(self._monitor)
|
|
logger.info("Monitor initiated!")
|
|
|
|
def _monitor(self):
|
|
current_files = [
|
|
os.path.join(self._directory_path, file)
|
|
for file in os.listdir(self._directory_path)
|
|
if file.lower().endswith('.csv')
|
|
]
|
|
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._files = current_files
|
|
if not current_files:
|
|
self._files = []
|
|
|
|
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]
|
|
|
|
def update_plots(self):
|
|
if self._files is not None:
|
|
print("updPlot")
|
|
self._mediator.notify(self, self._files)
|
|
|
|
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, [None])
|
|
|
|
def start_raport(self) -> None:
|
|
print("startRaport")
|
|
self.stop()
|
|
self._files = []
|
|
print(self._files)
|
|
self._mediator.notify(self, [None])
|
|
|
|
|
|
def start_seeking(self) -> None:
|
|
print("startSeeking")
|
|
self._init_state()
|
|
self.start()
|
|
|