我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用watchdog.events.FileSystemEventHandler()。
def on_created(self, event): """Overrides `on_created` in `FileSystemEventHandler` to filter on file creation events for `CompletedJobInfo.xml`.""" if isinstance(event, FileCreatedEvent) and event.src_path.endswith('CompletedJobInfo.xml'): logging.info("Observed new run in {}, telling the UI to start uploading it.".format(event.src_path)) directory = os.path.dirname(event.src_path) # tell the UI to clean itself up before observing new runs send_message(DirectoryMonitorTopics.new_run_observed) # this will send a bunch of events that the UI is listening for, but # unlike the UI (which runs this in a separate thread), we're going to do this # in our own thread and block on it so we can tell the UI to start # uploading once we've finished discovering the run find_runs_in_directory(directory) # now tell the UI to start send_message(DirectoryMonitorTopics.finished_discovering_run) else: logging.debug("Ignoring file event [{}] with path [{}]".format(str(event), event.src_path))
def __init__(self, *args, **kwargs): ReloaderLoop.__init__(self, *args, **kwargs) from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler self.observable_paths = set() def _check_modification(filename): if filename in self.extra_files: self.trigger_reload(filename) dirname = os.path.dirname(filename) if dirname.startswith(tuple(self.observable_paths)): if filename.endswith(('.pyc', '.pyo')): self.trigger_reload(filename[:-1]) elif filename.endswith('.py'): self.trigger_reload(filename) class _CustomHandler(FileSystemEventHandler): def on_created(self, event): _check_modification(event.src_path) def on_modified(self, event): _check_modification(event.src_path) def on_moved(self, event): _check_modification(event.src_path) _check_modification(event.dest_path) def on_deleted(self, event): _check_modification(event.src_path) reloader_name = Observer.__name__.lower() if reloader_name.endswith('observer'): reloader_name = reloader_name[:-8] reloader_name += ' reloader' self.name = reloader_name self.observer_class = Observer self.event_handler = _CustomHandler() self.should_reload = False
def __init__(self, *args, **kwargs): ReloaderLoop.__init__(self, *args, **kwargs) from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler self.observable_paths = set() def _check_modification(filename): if filename in self.extra_files: self.trigger_reload(filename) dirname = os.path.dirname(filename) if dirname.startswith(tuple(self.observable_paths)): if filename.endswith(('.pyc', '.pyo')): self.trigger_reload(filename[:-1]) elif filename.endswith('.py'): self.trigger_reload(filename) class _CustomHandler(FileSystemEventHandler): def on_created(self, event): _check_modification(event.src_path) def on_modified(self, event): _check_modification(event.src_path) reloader_name = Observer.__name__.lower() if reloader_name.endswith('observer'): reloader_name = reloader_name[:-8] reloader_name += ' reloader' self.name = reloader_name self.observer_class = Observer self.event_handler = _CustomHandler() self.should_reload = False
def log(s): print('[Monitor] %s' % s) # ???????????????watchdog.events.FileSystemEventHandler???
def watch_note(note, handle_func): """watch a single note for changes, call `handle_func` on change""" ob = Observer() handler = FileSystemEventHandler() note_filename = path.basename(note) server = MarkdownServer() server.start() def handle_event(event): _, filename = path.split(event.src_path) if note_filename == filename or \ path.normpath(event.src_path) == path.normpath(util.assets_dir(note)): print('compiling...') handle_func(note) server.update_clients() print('done') handler.on_any_event = handle_event print('watching "{0}"...'.format(note_filename)) ob.schedule(handler, path.dirname(note), recursive=True) ob.start() try: while True: time.sleep(1) except KeyboardInterrupt: print('stopping...') ob.stop() server.shutdown() ob.join()
def __init__(self, **kwargs): super(FilesContent, self).__init__(**kwargs) Clock.schedule_interval(self.collect_meta_data, 0.1) #if octoprint has changed files then update them session_saver.save_variable('file_callback', self.call_to_update) session_saver.save_variable('usb_mounted', False) #schedule directory observer http://pythonhosted.org/watchdog/api.html#watchdog.events.FileSystemEventHandler self.dir_observe = Observer() self.callback = FileSystemEventHandler() self.callback.on_any_event = self.on_any_event if sys.platform != "win32": self.dir_observe.schedule(self.callback, "/dev", recursive=True) self.dir_observe.start() self.update_files() # This seems like a roundabout way to call 'update_files' However if it is not called in this way # Graphical issues become present.It is called in this way to make the request come from the # thread kivy is on
