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)]

        self.update_timer.timeout.connect(self._monitor)
        logger.info("Monitor initiated!")

    def _monitor(self):
        files = [os.path.join(self._directory_path, file) for file in os.listdir(self._directory_path)]
        new_files = sorted(list(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 = []

    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]
            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(self._files)
            self._mediator.notify(self, self._files)

    def force_all_dir(self):
        files = os.listdir(self._directory_path)
        self._files = files
        all_files = []
        for x in self._files:
            path = os.path.join(self._directory_path, x)
            all_files.append(path)
        if all_files:
            logger.info(f"Plotting all files: {all_files}")
            self._mediator.notify(self, all_files)
        else:
            logger.info(f"Failed pushing {str(all_files)}")

    def custom_csv_extract_only(self, path: str):
        self.stop()
        self._files = [path]
        if path is not None:
            self._mediator.notify(self, [path])
    
    def start_seeking(self) -> None:
        self._init_state()
        self.start()