WeldingSpotPerformance/src/controller/monitor.py

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:
            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, [''])
    
    def start_raport(self) -> None:
        self.stop()
        self._files = ['']
        self._mediator.notify(self, [''])
            
    
    def start_seeking(self) -> None:
        self._init_state()
        self.start()