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