Skip to content

file_monitor.py

This module contains file monitoring class that can be used to check files change periodically and call callback function to handle properly when detecting files change.

__all__ = ['FileChangesChecker', 'FileMonitor'] module-attribute

FileChangesChecker

Files change checker.

Source code in solnlib/file_monitor.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
class FileChangesChecker:
    """Files change checker."""

    def __init__(self, callback: Callable[[List[str]], Any], files: List):
        """Initializes FileChangesChecker.

        Arguments:
            callback: Callback function for files change.
            files: Files to be monitored with full path.
        """
        self._callback = callback
        self._files = files

        self.file_mtimes = {file_name: None for file_name in self._files}
        for k in self.file_mtimes:
            try:
                self.file_mtimes[k] = op.getmtime(k)
            except OSError:
                logging.debug(f"Getmtime for {k}, failed: {traceback.format_exc()}")

    def check_changes(self) -> bool:
        """Check files change.

        If some files are changed and callback function is not None, call
        callback function to handle files change.

        Returns:
            True if files changed else False
        """
        logging.debug(f"Checking files={self._files}")
        file_mtimes = self.file_mtimes
        changed_files = []
        for f, last_mtime in list(file_mtimes.items()):
            try:
                current_mtime = op.getmtime(f)
                if current_mtime != last_mtime:
                    file_mtimes[f] = current_mtime
                    changed_files.append(f)
                    logging.info(f"Detect {f} has changed", f)
            except OSError:
                pass
        if changed_files:
            if self._callback:
                self._callback(changed_files)
            return True
        return False

file_mtimes = {file_name: None for file_name in self._files} instance-attribute

__init__(callback, files)

Initializes FileChangesChecker.

Parameters:

Name Type Description Default
callback Callable[[List[str]], Any]

Callback function for files change.

 required
files List

Files to be monitored with full path.

 required
Source code in solnlib/file_monitor.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def __init__(self, callback: Callable[[List[str]], Any], files: List):
    """Initializes FileChangesChecker.

    Arguments:
        callback: Callback function for files change.
        files: Files to be monitored with full path.
    """
    self._callback = callback
    self._files = files

    self.file_mtimes = {file_name: None for file_name in self._files}
    for k in self.file_mtimes:
        try:
            self.file_mtimes[k] = op.getmtime(k)
        except OSError:
            logging.debug(f"Getmtime for {k}, failed: {traceback.format_exc()}")

check_changes()

Check files change.

If some files are changed and callback function is not None, call callback function to handle files change.

Returns:

Type Description
bool

True if files changed else False
Source code in solnlib/file_monitor.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def check_changes(self) -> bool:
    """Check files change.

    If some files are changed and callback function is not None, call
    callback function to handle files change.

    Returns:
        True if files changed else False
    """
    logging.debug(f"Checking files={self._files}")
    file_mtimes = self.file_mtimes
    changed_files = []
    for f, last_mtime in list(file_mtimes.items()):
        try:
            current_mtime = op.getmtime(f)
            if current_mtime != last_mtime:
                file_mtimes[f] = current_mtime
                changed_files.append(f)
                logging.info(f"Detect {f} has changed", f)
        except OSError:
            pass
    if changed_files:
        if self._callback:
            self._callback(changed_files)
        return True
    return False

FileMonitor

Files change monitor.

Monitor files change in a separated thread and call callback when there is files change.

Examples:

>>> import solnlib.file_monitor as fm
>>> fm = fm.FileMonitor(fm_callback, files_list, 5)
>>> fm.start()
Source code in solnlib/file_monitor.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
class FileMonitor:
    """Files change monitor.

    Monitor files change in a separated thread and call callback
    when there is files change.

    Examples:
      >>> import solnlib.file_monitor as fm
      >>> fm = fm.FileMonitor(fm_callback, files_list, 5)
      >>> fm.start()
    """

    def __init__(
        self, callback: Callable[[List[str]], Any], files: List, interval: int = 1
    ):
        """Initializes FileMonitor.

        Arguments:
            callback: Callback for handling files change.
            files: Files to monitor.
            interval: Interval to check files change.
        """
        self._checker = FileChangesChecker(callback, files)
        self._thr = threading.Thread(target=self._do_monitor)
        self._thr.daemon = True
        self._interval = interval
        self._started = False

    def start(self):
        """Start file monitor.

        Start a background thread to monitor files change.
        """

        if self._started:
            return
        self._started = True

        self._thr.start()

    def stop(self):
        """Stop file monitor.

        Stop the background thread to monitor files change.
        """

        self._started = False

    def _do_monitor(self):
        while self._started:
            self._checker.check_changes()

            for _ in range(self._interval):
                if not self._started:
                    break
                time.sleep(1)

__init__(callback, files, interval=1)

Initializes FileMonitor.

Parameters:

Name Type Description Default
callback Callable[[List[str]], Any]

Callback for handling files change.

 required
files List

Files to monitor.

 required
interval int

Interval to check files change.

 1
Source code in solnlib/file_monitor.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def __init__(
    self, callback: Callable[[List[str]], Any], files: List, interval: int = 1
):
    """Initializes FileMonitor.

    Arguments:
        callback: Callback for handling files change.
        files: Files to monitor.
        interval: Interval to check files change.
    """
    self._checker = FileChangesChecker(callback, files)
    self._thr = threading.Thread(target=self._do_monitor)
    self._thr.daemon = True
    self._interval = interval
    self._started = False

start()

Start file monitor.

Start a background thread to monitor files change.

Source code in solnlib/file_monitor.py 
106
107
108
109
110
111
112
113
114
115
116
def start(self):
    """Start file monitor.

    Start a background thread to monitor files change.
    """

    if self._started:
        return
    self._started = True

    self._thr.start()

stop()

Stop file monitor.

Stop the background thread to monitor files change.

Source code in solnlib/file_monitor.py 
118
119
120
121
122
123
124
def stop(self):
    """Stop file monitor.

    Stop the background thread to monitor files change.
    """

    self._started = False