Skip to content

hmr-daemon API Reference

hmr_daemon.windows

excludes module-attribute 

excludes = (
    (venv,) if (venv := (getenv("VIRTUAL_ENV"))) else ()
)

shutdown_event module-attribute 

shutdown_event = Event()

patch_first module-attribute 

patch_first = 'hmr' in name

get_code 

get_code(_: ReactiveModuleLoader, fullname: str)
Source code in hmr_daemon/windows.py 
11
12
13
14
15
16
 
def get_code(_: ReactiveModuleLoader, fullname: str):
    from ast import parse
    from importlib.util import find_spec

    if (spec := find_spec(fullname)) is not None and (file := spec.origin) is not None:
        return compile(parse(Path(file).read_text(), str(file)), str(file), "exec", dont_inherit=True)

patch 

patch()
Source code in hmr_daemon/windows.py 
22
23
24
25
26
27
28
29
30
31
32
 
def patch():
    global original_init

    @wraps(original_init := BaseReloader.__init__)
    def wrapper(*args, **kwargs):
        if not state.disabled:
            shutdown_event.set()
            BaseReloader.__init__ = original_init
        original_init(*args, **kwargs)

    BaseReloader.__init__ = wrapper

main 

main()
Source code in hmr_daemon/windows.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
 
def main():
    state.disabled = True

    class Reloader(SyncReloader):
        def __init__(self):
            self.includes = (".",)
            self.excludes = excludes
            self.error_filter = ErrorFilter(*map(str, Path(hmr_file, "..").resolve().glob("**/*.py")), __file__)

        def start_watching(self):
            if shutdown_event.is_set():
                return

            from watchfiles import PythonFilter, watch

            if shutdown_event.is_set():
                return

            for events in watch(".", watch_filter=PythonFilter(), stop_event=shutdown_event):
                self.on_events(events)

    if not shutdown_event.is_set():
        Reloader().start_watching()

hmr_daemon.posix

_ 

_()
Source code in hmr_daemon/posix/__init__.py 
10
11
12
13
14
15
16
 
@register
def _():
    worker.terminate()
    try:
        worker.wait(timeout=0.1)
    except TimeoutExpired:
        worker.kill()

main

shutdown_event module-attribute 

shutdown_event = Event()

excludes module-attribute 

excludes = (
    (venv,) if (venv := (getenv("VIRTUAL_ENV"))) else ()
)

patch_first module-attribute 

patch_first = 'hmr' in name

PipeReloader

Bases: SyncReloader

Source code in hmr_daemon/posix/main.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
 
class PipeReloader(SyncReloader):
    def __init__(self, process: Popen):
        self._process = process
        self.includes = (".",)
        self.excludes = excludes
        self.error_filter = ErrorFilter(*map(str, Path(hmr_file, "..").resolve().glob("**/*.py")), __file__)

    def iterate_pipe_events(self) -> Iterable[set[tuple[Change, str]]]:
        from json import loads

        from watchfiles import Change

        while not shutdown_event.is_set():
            # Check if worker process is still alive
            if self._process.poll() is not None:
                # Worker process terminated
                return

            # Read data from pipe
            try:
                assert self._process.stdout is not None
                line = self._process.stdout.readline()
                if not line:  # EOF
                    return
            except OSError:
                # Pipe error
                return

            # Process events - each line is a complete events list
            if events_data := loads(line.decode()):
                yield {(Change(event_int), path) for event_int, path in events_data}

    def start_watching(self):
        for events in self.iterate_pipe_events():
            if shutdown_event.is_set():
                return
            self.on_events(events)

    def cleanup(self):
        if self._process.poll() is None:
            self._process.terminate()
            try:
                self._process.wait(timeout=0.1)
            except TimeoutExpired:
                self._process.kill()
_process instance-attribute 
_process = process
includes instance-attribute 
includes = ('.',)
excludes instance-attribute 
excludes = excludes
error_filter instance-attribute 
error_filter = ErrorFilter(
    *(map(str, glob("**/*.py"))), __file__
)
__init__ 
__init__(process: Popen)
Source code in hmr_daemon/posix/main.py 
54
55
56
57
58
 
def __init__(self, process: Popen):
    self._process = process
    self.includes = (".",)
    self.excludes = excludes
    self.error_filter = ErrorFilter(*map(str, Path(hmr_file, "..").resolve().glob("**/*.py")), __file__)
iterate_pipe_events 
iterate_pipe_events() -> Iterable[set[tuple[Change, str]]]
Source code in hmr_daemon/posix/main.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
 
def iterate_pipe_events(self) -> Iterable[set[tuple[Change, str]]]:
    from json import loads

    from watchfiles import Change

    while not shutdown_event.is_set():
        # Check if worker process is still alive
        if self._process.poll() is not None:
            # Worker process terminated
            return

        # Read data from pipe
        try:
            assert self._process.stdout is not None
            line = self._process.stdout.readline()
            if not line:  # EOF
                return
        except OSError:
            # Pipe error
            return

        # Process events - each line is a complete events list
        if events_data := loads(line.decode()):
            yield {(Change(event_int), path) for event_int, path in events_data}
start_watching 
start_watching()
Source code in hmr_daemon/posix/main.py 
85
86
87
88
89
 
def start_watching(self):
    for events in self.iterate_pipe_events():
        if shutdown_event.is_set():
            return
        self.on_events(events)
cleanup 
cleanup()
Source code in hmr_daemon/posix/main.py 
91
92
93
94
95
96
97
 
def cleanup(self):
    if self._process.poll() is None:
        self._process.terminate()
        try:
            self._process.wait(timeout=0.1)
        except TimeoutExpired:
            self._process.kill()

get_code 

get_code(_: ReactiveModuleLoader, fullname: str)
Source code in hmr_daemon/posix/main.py 
14
15
16
17
18
19
20
 
def get_code(_: ReactiveModuleLoader, fullname: str):
    from ast import parse
    from importlib.util import find_spec

    if (spec := find_spec(fullname)) is not None and (file := spec.origin) is not None:
        path = Path(file)
        return compile(parse(path.read_text(), str(path)), str(path), "exec", dont_inherit=True)

patch 

patch()
Source code in hmr_daemon/posix/main.py 
40
41
42
43
44
45
46
47
48
49
50
 
def patch():
    global original_init

    @wraps(original_init := BaseReloader.__init__)
    def wrapper(*args, **kwargs):
        if not state.disabled:
            shutdown_event.set()
            BaseReloader.__init__ = original_init
        original_init(*args, **kwargs)

    BaseReloader.__init__ = wrapper

_watch 

_watch(process: Popen)
Source code in hmr_daemon/posix/main.py 
100
101
102
103
104
105
106
107
 
def _watch(process: Popen):
    if shutdown_event.is_set():
        return
    reloader = PipeReloader(process)
    try:
        reloader.start_watching()
    finally:
        reloader.cleanup()

run_reloader 

run_reloader(process: Popen)
Source code in hmr_daemon/posix/main.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
 
def run_reloader(process: Popen):
    state.disabled = True  # disable self-shutdown wrapper until first reloader init

    def watch():
        try:
            _watch(process)
        finally:
            shutdown_event.set()

    if patch_first:
        patch()
        Thread(target=watch, daemon=True, name="hmr-daemon").start()
    else:
        Thread(target=lambda: [patch(), watch()], daemon=True, name="hmr-daemon").start()

worker

shutdown_event module-attribute 

shutdown_event = Event()

events_data module-attribute 

events_data = [(int(event), path) for event, path in events]