Mercurial > repos > shellac > guppy_basecaller

comparison env/lib/python3.7/site-packages/galaxy/util/watcher.py @ 0:26e78fe6e8c4 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit c699937486c35866861690329de38ec1a5d9f783"
author shellac
date Sat, 02 May 2020 07:14:21 -0400
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:26e78fe6e8c4
1 # TODO: this is largely copied from galaxy.tools.toolbox.galaxy and generalized, the tool-oriented watchers in that
2 # module should probably be updated to use this where possible
3
4 from __future__ import absolute_import
5
6 import logging
7 import os.path
8 import time
9
10 from six.moves import filter
11
12 try:
13 from watchdog.events import FileSystemEventHandler
14 from watchdog.observers import Observer
15 from watchdog.observers.polling import PollingObserver
16 can_watch = True
17 except ImportError:
18 Observer = None
19 FileSystemEventHandler = object
20 PollingObserver = None
21 can_watch = False
22
23 from galaxy.util.hash_util import md5_hash_file
24
25 log = logging.getLogger(__name__)
26
27
28 def get_observer_class(config_name, config_value, default, monitor_what_str):
29 """
30 """
31 config_value = config_value or default
32 config_value = str(config_value).lower()
33 if config_value in ("true", "yes", "on", "auto"):
34 expect_observer = True
35 observer_class = Observer
36 elif config_value == "polling":
37 expect_observer = True
38 observer_class = PollingObserver
39 elif config_value in ('false', 'no', 'off'):
40 expect_observer = False
41 observer_class = None
42 else:
43 message = "Unrecognized value for %s config option: %s" % (config_name, config_value)
44 raise Exception(message)
45
46 if expect_observer and observer_class is None:
47 message = "Watchdog library unavailable, cannot monitor %s." % monitor_what_str
48 if config_value == "auto":
49 log.info(message)
50 else:
51 raise Exception(message)
52
53 return observer_class
54
55
56 def get_watcher(config, config_name, default="False", monitor_what_str=None, watcher_class=None,
57 event_handler_class=None, **kwargs):
58 config_value = getattr(config, config_name, None)
59 observer_class = get_observer_class(config_name, config_value, default=default, monitor_what_str=monitor_what_str)
60 if observer_class is not None:
61 watcher_class = watcher_class or Watcher
62 event_handler_class = event_handler_class or EventHandler
63 return watcher_class(observer_class, event_handler_class, **kwargs)
64 else:
65 return NullWatcher()
66
67
68 class BaseWatcher(object):
69
70 def __init__(self, observer_class, even_handler_class, **kwargs):
71 self.observer = None
72 self.observer_class = observer_class
73 self.event_handler = even_handler_class(self)
74 self.monitored_dirs = {}
75
76 def start(self):
77 if self.observer is None:
78 self.observer = self.observer_class()
79 self.observer.start()
80 self.resume_watching()
81
82 def monitor(self, dir_path, recursive=False):
83 self.monitored_dirs[dir_path] = recursive
84 if self.observer is not None:
85 self.observer.schedule(self.event_handler, dir_path, recursive=recursive)
86
87 def resume_watching(self):
88 for dir_path, recursive in self.monitored_dirs.items():
89 self.monitor(dir_path, recursive)
90
91 def shutdown(self):
92 if self.observer is not None:
93 self.observer.stop()
94 self.observer.join()
95 self.observer = None
96
97
98 class Watcher(BaseWatcher):
99
100 def __init__(self, observer_class, event_handler_class, **kwargs):
101 super(Watcher, self).__init__(observer_class, event_handler_class, **kwargs)
102 self.path_hash = {}
103 self.file_callbacks = {}
104 self.dir_callbacks = {}
105 self.ignore_extensions = {}
106 self.require_extensions = {}
107 self.event_handler = event_handler_class(self)
108
109 def watch_file(self, file_path, callback=None):
110 file_path = os.path.abspath(file_path)
111 dir_path = os.path.dirname(file_path)
112 if dir_path not in self.monitored_dirs:
113 if callback is not None:
114 self.file_callbacks[file_path] = callback
115 self.monitor(dir_path)
116 log.debug("Watching for changes to file: %s", file_path)
117
118 def watch_directory(self, dir_path, callback=None, recursive=False, ignore_extensions=None, require_extensions=None):
119 dir_path = os.path.abspath(dir_path)
120 if dir_path not in self.monitored_dirs:
121 if callback is not None:
122 self.dir_callbacks[dir_path] = callback
123 if ignore_extensions:
124 self.ignore_extensions[dir_path] = ignore_extensions
125 if require_extensions:
126 self.require_extensions[dir_path] = require_extensions
127 self.monitor(dir_path, recursive=recursive)
128 log.debug("Watching for changes in directory%s: %s", ' (recursively)' if recursive else '', dir_path)
129
130
131 class EventHandler(FileSystemEventHandler):
132
133 def __init__(self, watcher):
134 self.watcher = watcher
135
136 def on_any_event(self, event):
137 self._handle(event)
138
139 def _extension_check(self, key, path):
140 required_extensions = self.watcher.require_extensions.get(key)
141 if required_extensions:
142 return any(filter(path.endswith, required_extensions))
143 return not any(filter(path.endswith, self.watcher.ignore_extensions.get(key, [])))
144
145 def _handle(self, event):
146 # modified events will only have src path, move events will
147 # have dest_path and src_path but we only care about dest. So
148 # look at dest if it exists else use src.
149 path = getattr(event, 'dest_path', None) or event.src_path
150 path = os.path.abspath(path)
151 callback = self.watcher.file_callbacks.get(path)
152 if os.path.basename(path).startswith('.'):
153 return
154 if callback:
155 ext_ok = self._extension_check(path, path)
156 else:
157 # reversed sort for getting the most specific dir first
158 for key in reversed(sorted(self.watcher.dir_callbacks.keys())):
159 if os.path.commonprefix([path, key]) == key:
160 callback = self.watcher.dir_callbacks[key]
161 ext_ok = self._extension_check(key, path)
162 break
163 if not callback or not ext_ok:
164 return
165 cur_hash = md5_hash_file(path)
166 if cur_hash:
167 if self.watcher.path_hash.get(path) == cur_hash:
168 return
169 else:
170 time.sleep(0.5)
171 if cur_hash != md5_hash_file(path):
172 # We're still modifying the file, it'll be picked up later
173 return
174 self.watcher.path_hash[path] = cur_hash
175 callback(path=path)
176
177
178 class NullWatcher(object):
179
180 def start(self):
181 pass
182
183 def shutdown(self):
184 pass
185
186 def watch_file(self, *args, **kwargs):
187 pass
188
189 def watch_directory(self, *args, **kwargs):
190 pass