/usr/lib/python3/dist-packages/livereload/watcher.py is in python3-livereload 2.4.0-1.
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 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 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 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 | # -*- coding: utf-8 -*-
"""
livereload.watcher
~~~~~~~~~~~~~~~~~~
A file watch management for LiveReload Server.
:copyright: (c) 2013 - 2015 by Hsiaoming Yang
"""
import os
import glob
import time
try:
import pyinotify
except ImportError:
pyinotify = None
class Watcher(object):
"""A file watcher registery."""
def __init__(self):
self._tasks = {}
self._mtimes = {}
# setting changes
self._changes = []
# filepath that is changed
self.filepath = None
self._start = time.time()
def ignore(self, filename):
"""Ignore a given filename or not."""
_, ext = os.path.splitext(filename)
return ext in ['.pyc', '.pyo', '.o', '.swp']
def watch(self, path, func=None, delay=0, ignore=None):
"""Add a task to watcher.
:param path: a filepath or directory path or glob pattern
:param func: the function to be executed when file changed
:param delay: Delay sending the reload message. Use 'forever' to
not send it. This is useful to compile sass files to
css, but reload on changed css files then only.
:param ignore: A function return True to ignore a certain pattern of
filepath.
"""
self._tasks[path] = {
'func': func,
'delay': delay,
'ignore': ignore,
}
def start(self, callback):
"""Start the watcher running, calling callback when changes are
observed. If this returns False, regular polling will be used."""
return False
def examine(self):
"""Check if there are changes, if true, run the given task."""
if self._changes:
return self._changes.pop()
# clean filepath
self.filepath = None
delays = set()
for path in self._tasks:
item = self._tasks[path]
if self.is_changed(path, item['ignore']):
func = item['func']
func and func()
delay = item['delay']
if delay and isinstance(delay, int):
delays.add(delay)
if delays:
delay = max(delays)
else:
delay = None
return self.filepath, delay
def is_changed(self, path, ignore=None):
if os.path.isfile(path):
return self.is_file_changed(path, ignore)
elif os.path.isdir(path):
return self.is_folder_changed(path, ignore)
return self.is_glob_changed(path, ignore)
def is_file_changed(self, path, ignore=None):
if not os.path.isfile(path):
return False
if self.ignore(path):
return False
if ignore and ignore(path):
return False
mtime = os.path.getmtime(path)
if path not in self._mtimes:
self._mtimes[path] = mtime
self.filepath = path
return mtime > self._start
if self._mtimes[path] != mtime:
self._mtimes[path] = mtime
self.filepath = path
return True
self._mtimes[path] = mtime
return False
def is_folder_changed(self, path, ignore=None):
for root, dirs, files in os.walk(path, followlinks=True):
if '.git' in dirs:
dirs.remove('.git')
if '.hg' in dirs:
dirs.remove('.hg')
if '.svn' in dirs:
dirs.remove('.svn')
if '.cvs' in dirs:
dirs.remove('.cvs')
for f in files:
if self.is_file_changed(os.path.join(root, f), ignore):
return True
return False
def is_glob_changed(self, path, ignore=None):
for f in glob.glob(path):
if self.is_file_changed(f, ignore):
return True
return False
class INotifyWatcher(Watcher):
def __init__(self):
Watcher.__init__(self)
self.wm = pyinotify.WatchManager()
self.notifier = None
self.callback = None
def watch(self, path, func=None, delay=None, ignore=None):
flag = pyinotify.IN_CREATE | pyinotify.IN_DELETE | pyinotify.IN_MODIFY
self.wm.add_watch(path, flag, rec=True, do_glob=True, auto_add=True)
Watcher.watch(self, path, func, delay, ignore)
def inotify_event(self, event):
self.callback()
def start(self, callback):
if not self.notifier:
self.callback = callback
from tornado import ioloop
self.notifier = pyinotify.TornadoAsyncNotifier(
self.wm, ioloop.IOLoop.instance(),
default_proc_fun=self.inotify_event
)
callback()
return True
def get_watcher_class():
if pyinotify is None:
return Watcher
return INotifyWatcher
|