WeldingSpotPerformance/src/controller/monitor.py

import os
from loguru import logger

from utils.base.base import BaseDirectoryMonitor


class DirectoryMonitor(BaseDirectoryMonitor):

    def _init_state(self):
        files = os.listdir(self._directory_path)
        self._files = files

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

    def _monitor(self):
        files = os.listdir(self._directory_path)
        new_files = sorted(list(map(lambda x: os.path.join(self._directory_path, x),
                                    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:
        self.stop()
        operator_params, 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()

    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_dir_extract(self, dict_path:str ):
        files = os.listdir(dict_path)
        if files is not None:
            all_files = [os.path.join(dict_path, file) for file in files]
            self._mediator.notify(self, all_files)