我们从Python开源项目中，提取了以下21个代码示例，用于说明如何使用pyinotify.IN_MODIFY。
def start(self): for f in os.listdir(self.path): name = file2mod(f) if not name: continue self._load(name) wm = pyinotify.WatchManager() wm.add_watch(self.path, pyinotify.IN_DELETE | pyinotify.IN_CREATE | pyinotify.IN_MODIFY , rec = False) handle = EventHandler(self) self._notifier = pyinotify.TornadoAsyncNotifier( wm, self._ioloop, default_proc_fun = handle)
def __init__(self, rt): ManagerPlugin.__init__(self, rt) self.git_repo = GitRepo(directory=self.rt.paths.skills, url='https://github.com/MatthewScholefield/mycroft-light.git', branch='skills', update_freq=1) self.blacklist = self.config['blacklist'] sys.path.append(self.rt.paths.skills) GroupPlugin.__init__(self, SkillPlugin, 'mycroft.skills', '_skill') for cls in self._classes.values(): cls.rt = rt self.init_plugins() # The watch manager stores the watches and provides operations on watches wm = pyinotify.WatchManager() mask = pyinotify.IN_MODIFY | pyinotify.IN_CREATE | pyinotify.IN_DELETE | pyinotify.IN_MOVED_TO skills_dir = self.rt.paths.skills handler = EventHandler(self) notifier = pyinotify.ThreadedNotifier(wm, handler) notifier.daemon = True wdd = wm.add_watch(skills_dir, mask, rec=True) notifier.start()
def start_config_watch(self): wm = pyinotify.WatchManager() wm.add_watch('./config/mitmf.conf', pyinotify.IN_MODIFY) notifier = pyinotify.Notifier(wm, self) t = threading.Thread(name='ConfigWatcher', target=notifier.loop) t.setDaemon(True) t.start()
def addMonitor( self, filename, callback, args = None ): try: mask = pyinotify.IN_DELETE | pyinotify.IN_CREATE | pyinotify.IN_MODIFY mId = self.wm.add_watch( filename, mask, rec = True)[filename] if mId >= 0: self.callbacks[mId] = ( callback, args ) except Exception as detail: mId = 0 return mId
def watch_shakemaps_push( working_dir, timeout=None, handler=None, daemon=False): wm = pyinotify.WatchManager() if daemon: notifier = pyinotify.ThreadedNotifier(wm, handler, timeout=timeout) else: notifier = pyinotify.Notifier(wm, handler, timeout=timeout) flags = pyinotify.IN_CREATE | pyinotify.IN_MODIFY | pyinotify.IN_MOVED_TO wm.add_watch(working_dir, flags, rec=True, auto_add=True) return notifier
def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self._watch_manager = pyinotify.WatchManager() self._mask = pyinotify.IN_MODIFY # TODO: does this include rolling?
def addMonitor( self, filename, callback, args = None ): try: mask = pyinotify.IN_DELETE | pyinotify.IN_CREATE | pyinotify.IN_MODIFY mId = self.wm.add_watch( filename, mask, rec = True)[filename] if mId >= 0: self.callbacks[mId] = ( callback, args ) except Exception, detail: mId = 0 return mId
def inotify_code_changed(): """ Checks for changed code using inotify. After being called it blocks until a change event has been fired. """ class EventHandler(pyinotify.ProcessEvent): modified_code = None def process_default(self, event): if event.path.endswith('.mo'): EventHandler.modified_code = I18N_MODIFIED else: EventHandler.modified_code = FILE_MODIFIED wm = pyinotify.WatchManager() notifier = pyinotify.Notifier(wm, EventHandler()) def update_watch(sender=None, **kwargs): if sender and getattr(sender, 'handles_files', False): # No need to update watches when request serves files. # (sender is supposed to be a django.core.handlers.BaseHandler subclass) return mask = ( pyinotify.IN_MODIFY | pyinotify.IN_DELETE | pyinotify.IN_ATTRIB | pyinotify.IN_MOVED_FROM | pyinotify.IN_MOVED_TO | pyinotify.IN_CREATE | pyinotify.IN_DELETE_SELF | pyinotify.IN_MOVE_SELF ) for path in gen_filenames(only_new=True): wm.add_watch(path, mask) # New modules may get imported when a request is processed. request_finished.connect(update_watch) # Block until an event happens. update_watch() notifier.check_events(timeout=None) notifier.read_events() notifier.process_events() notifier.stop() # If we are here the code must have changed. return EventHandler.modified_code
def _reloader_inotify(extra_files=None, interval=None): # Mutated by inotify loop when changes occur. changed = [False] # Setup inotify watches from pyinotify import WatchManager, Notifier # this API changed at one point, support both try: from pyinotify import EventsCodes as ec ec.IN_ATTRIB except (ImportError, AttributeError): import pyinotify as ec wm = WatchManager() mask = ec.IN_DELETE_SELF | ec.IN_MOVE_SELF | ec.IN_MODIFY | ec.IN_ATTRIB def signal_changed(event): if changed[0]: return _log('info', ' * Detected change in %r, reloading' % event.path) changed[:] = [True] for fname in extra_files or (): wm.add_watch(fname, mask, signal_changed) # ... And now we wait... notif = Notifier(wm) try: while not changed[0]: # always reiterate through sys.modules, adding them for fname in _iter_module_files(): wm.add_watch(fname, mask, signal_changed) notif.process_events() if notif.check_events(timeout=interval): notif.read_events() # TODO Set timeout to something small and check parent liveliness finally: notif.stop() sys.exit(3) # currently we always use the stat loop reloader for the simple reason # that the inotify one does not respond to added files properly. Also # it's quite buggy and the API is a mess.
def inotify_code_changed(): """ Checks for changed code using inotify. After being called it blocks until a change event has been fired. """ class EventHandler(pyinotify.ProcessEvent): modified_code = None def process_default(self, event): if event.path.endswith('.mo'): EventHandler.modified_code = I18N_MODIFIED else: EventHandler.modified_code = FILE_MODIFIED wm = pyinotify.WatchManager() notifier = pyinotify.Notifier(wm, EventHandler()) def update_watch(sender=None, **kwargs): if sender and getattr(sender, 'handles_files', False): # No need to update watches when request serves files. # (sender is supposed to be a django.core.handlers.BaseHandler subclass) return mask = ( pyinotify.IN_MODIFY | pyinotify.IN_DELETE | pyinotify.IN_ATTRIB | pyinotify.IN_MOVED_FROM | pyinotify.IN_MOVED_TO | pyinotify.IN_CREATE ) for path in gen_filenames(only_new=True): wm.add_watch(path, mask) # New modules may get imported when a request is processed. request_finished.connect(update_watch) # Block until an event happens. update_watch() notifier.check_events(timeout=None) notifier.read_events() notifier.process_events() notifier.stop() # If we are here the code must have changed. return EventHandler.modified_code
