1import os 2from collections import defaultdict 3 4from .. import importcompletion 5 6try: 7 from watchdog.observers import Observer 8 from watchdog.events import FileSystemEventHandler 9except ImportError: 10 11 def ModuleChangedEventHandler(*args): 12 return None 13 14 15else: 16 17 class ModuleChangedEventHandler(FileSystemEventHandler): # type: ignore [no-redef] 18 def __init__(self, paths, on_change): 19 self.dirs = defaultdict(set) 20 self.on_change = on_change 21 self.modules_to_add_later = [] 22 self.observer = Observer() 23 self.old_dirs = defaultdict(set) 24 self.started = False 25 self.activated = False 26 for path in paths: 27 self._add_module(path) 28 29 super().__init__() 30 31 def reset(self): 32 self.dirs = defaultdict(set) 33 del self.modules_to_add_later[:] 34 self.old_dirs = defaultdict(set) 35 self.observer.unschedule_all() 36 37 def _add_module(self, path): 38 """Add a python module to track changes""" 39 path = os.path.abspath(path) 40 for suff in importcompletion.SUFFIXES: 41 if path.endswith(suff): 42 path = path[: -len(suff)] 43 break 44 dirname = os.path.dirname(path) 45 if dirname not in self.dirs: 46 self.observer.schedule(self, dirname, recursive=False) 47 self.dirs[dirname].add(path) 48 49 def _add_module_later(self, path): 50 self.modules_to_add_later.append(path) 51 52 def track_module(self, path): 53 """ 54 Begins tracking this if activated, or remembers to track later. 55 """ 56 if self.activated: 57 self._add_module(path) 58 else: 59 self._add_module_later(path) 60 61 def activate(self): 62 if self.activated: 63 raise ValueError(f"{self!r} is already activated.") 64 if not self.started: 65 self.started = True 66 self.observer.start() 67 for dirname in self.dirs: 68 self.observer.schedule(self, dirname, recursive=False) 69 for module in self.modules_to_add_later: 70 self._add_module(module) 71 del self.modules_to_add_later[:] 72 self.activated = True 73 74 def deactivate(self): 75 if not self.activated: 76 raise ValueError(f"{self!r} is not activated.") 77 self.observer.unschedule_all() 78 self.activated = False 79 80 def on_any_event(self, event): 81 dirpath = os.path.dirname(event.src_path) 82 paths = [path + ".py" for path in self.dirs[dirpath]] 83 if event.src_path in paths: 84 self.on_change(files_modified=[event.src_path]) 85