/usr/lib/python2.7/dist-packages/hupper/polling.py is in python-hupper 1.0-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 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 | import os
import threading
import time
from .interfaces import IFileMonitor
class PollingFileMonitor(threading.Thread, IFileMonitor):
"""
An :class:`hupper.interfaces.IFileMonitor` that stats the files
at periodic intervals.
``callback`` is a callable that accepts a path to a changed file.
``poll_interval`` is a value in seconds between scans of the files on
disk. Do not set this too low or it will eat your CPU and kill your drive.
"""
def __init__(self, callback, poll_interval=1):
super(PollingFileMonitor, self).__init__()
self.callback = callback
self.poll_interval = poll_interval
self.paths = set()
self.mtimes = {}
self.lock = threading.Lock()
def add_path(self, path):
with self.lock:
self.paths.add(path)
def run(self):
self.enabled = True
while self.enabled:
with self.lock:
paths = list(self.paths)
self.check_reload(paths)
time.sleep(self.poll_interval)
def stop(self):
self.enabled = False
def check_reload(self, paths):
changes = set()
for path in paths:
mtime = get_mtime(path)
if path not in self.mtimes:
self.mtimes[path] = mtime
elif self.mtimes[path] < mtime:
self.mtimes[path] = mtime
changes.add(path)
for path in sorted(changes):
self.callback(path)
def get_mtime(path):
try:
stat = os.stat(path)
if stat:
return stat.st_mtime
except (OSError, IOError): # pragma: no cover
pass
return 0
|