html_compose.live

 35def live_server(
 36    daemon: ShellCommand,
 37    daemon_delay: float,
 38    conds: list[WatchCond],
 39    force_polling: bool = False,
 40    host: str = "localhost",
 41    port: int = 51353,
 42    print_paths=True,
 43    loop_delay=1,
 44    livereload_delay=0.2,
 45    daemon_host: str | None = None,
 46    daemon_port: int | None = None,
 47    daemon_timeout: float = 30.0,
 48    proxy_host: str | None = None,
 49    proxy_uri: str | None = None,
 50) -> None:
 51    """
 52    Run a live-reload server that also runs and reloads your Python server.
 53
 54    This is a development feature and not recommended for production use.
 55
 56    Delays are deduplicated after file changes by various delay properties
 57    to prevent chains of restarts.
 58
 59    :param daemon: Command to run in the background, typically a Python server
 60    :type daemon: ShellCommand
 61
 62    :param daemon_delay: Delay in seconds before restarting the daemon after a change.
 63    :type daemon_delay: float
 64
 65    :param conds: List of watch conditions, which are a path and action.
 66    :type conds: list[WatchCond]
 67
 68    :param force_polling: Force slow stat() polling backend - useful if your platform is unable to support OS based watching.
 69    :type force_polling: bool
 70
 71    :param host: Host for livereload server websocket to listen on
 72    :type host: str
 73
 74    :param port: Port for livereload server websocket to listen on
 75    :type port: int
 76
 77    :param print_paths: Enumerate paths being monitored
 78    :type print_paths: bool
 79
 80    :param loop_delay: Set delay between checks for changes. Usually unnecessary.
 81    :type loop_delay: float
 82
 83    :param livereload_delay: Delay livereload server update until x seconds after daemon update
 84    :type livereload_delay: float
 85
 86    :param daemon_host: Host the HTTP server daemon listens on. Used to determine when the server is back up.
 87    :type daemon_host: str | None
 88
 89    :param daemon_port: Port the HTTP server daemon listens on. Used to determine when the server is back up.
 90    :type daemon_port: int | None
 91
 92    :param daemon_timeout: Timeout in seconds to wait for daemon port to come online after restart.
 93    :type daemon_timeout: int | None
 94
 95    :param proxy_uri: If websocket is behind a reverse proxy, this is the URI to reach it by.
 96                      This is useful if you are developing behind SSL.
 97    :type proxy_uri: str
 98
 99    :param proxy_host: If websocket is behind a reverse proxy, this is the host to reach it by.
100                       This is useful if you are developing behind SSL.
101    :type proxy_host: str
102    """
103    w = Watcher(conds, force_polling=force_polling)
104    oh = w.overhead()
105    if print_paths:
106        for path in oh["paths"]:
107            print(f"Monitoring for changes: {path}")
108
109    if not w.force_polling:
110        print(
111            f"Monitoring {oh['path_count']} path(s) via RustNotify. "
112            f"{oh['recursive_count']} path(s) are monitored recursively."
113        )
114    else:
115        print(f"Monitoring {oh['path_count']} path(s) for changes via polling")
116
117    # Set livereload environment variables
118    daemon.env.update(
119        generate_livereload_env(host, port, proxy_host, proxy_uri)
120    )
121
122    daemon_task = ProcessTask(daemon, delay=0, sync=False)
123    daemon_stop_task = Task(action=lambda: daemon_task.cancel(), sync=True)
124    # Run livereload server
125    server = run_server(host, port)
126    tr = TaskRunner()
127    tr.add_task(daemon_task)
128    tr.run()  # Start task runner thread
129    pending_reload: set[str] = set()
130
131    def reload():
132        changed = list(pending_reload)
133        pending_reload.clear()
134        if daemon_port is not None:
135            # If specified, we want to wait for the listening daemon port
136            # to show up before we tell the browser to reload.
137            _wait_for_server(
138                daemon_host or host,
139                daemon_port,
140                daemon_timeout,
141                daemon_task=daemon_task,
142            )
143        reload_because(changed)
144
145    browser_update_task = Task(reload, delay=0, sync=False)
146    try:
147        while True:
148            if tr.cancelled:
149                print("Task runner has closed. Exiting...")
150                break
151
152            if daemon_task.has_ended_early():
153                status = daemon_task.status_code()
154                print(
155                    f"Daemon process has exited with code {status}. Exiting..."
156                )
157                break
158            hits = w.changed()
159            if hits:
160                paths_hit = set()
161                conds_hit: set[WatchCond] = set()
162                for hit in hits:
163                    paths_hit.add(hit.path)
164
165                    for cond in hit.conds:
166                        if cond.reload:
167                            pending_reload.add(hit.path)
168
169                        conds_hit.add(cond)
170
171                for path in paths_hit:
172                    print(f"Changed: {path}")
173
174                delay = 0.0
175                reload_tripped = False
176                for cond in conds_hit:
177                    if cond.task:
178                        tr.add_task(cond.task)
179
180                    if not cond.reload:
181                        continue
182                    delay = max(delay, cond.task.delay)
183                    reload_tripped = True
184
185                if reload_tripped:
186                    daemon_task.delay = delay + daemon_delay
187
188                    # this should make them fire on the same tick, in order
189                    daemon_stop_task.delay = daemon_task.delay
190
191                    # This constant should mean the server port is up
192                    browser_update_task.delay = (
193                        daemon_task.delay + livereload_delay
194                    )
195                    print(
196                        f"Reloading daemon after {daemon_task.delay} seconds..."
197                    )
198                    if any([c.server_reload for c in conds_hit]):
199                        tr.add_task(daemon_stop_task)
200                        tr.add_task(daemon_task)
201                    tr.add_task(browser_update_task)
202            sleep(loop_delay)
203    except KeyboardInterrupt:
204        print("Exiting...")
205    finally:
206        server.shutdown()
207
208        for watch in w.rust_watches:
209            watch.close()
210
211        daemon_task.cancel()
212        for cond in conds:
213            if cond.task:
214                cond.task.cancel()

18class ShellCommand:
19    def __init__(
20        self, command: str | list[str], env: dict[str, str] | None = None
21    ):
22        self.command = command
23        self.env = os.environ.copy()
24        if env:
25            self.env.update(env)

198class WatchCond:
199    """
200    A condition for watching file(s) and trigger action.
201    """
202
203    def __init__(
204        self,
205        path_glob: str | list[str],
206        action: ShellCommand | Callable | None,
207        ignore_glob: str | list[str] | None = None,
208        delay: float = 0,
209        server_reload: bool = True,
210        reload: bool = True,
211    ):
212        """
213        Initializes a WatchCond.
214
215        Args:
216            path_glob:
217                Glob pattern(s) to watch for changes.
218
219            action:
220                Action to run when a change is detected. Shell command or function.
221                When action is None, no action is run but reloads may still occur.
222
223            ignore_glob:
224                Glob patterns to ignore.
225
226            delay:
227                Delay in seconds before running the action after a change.
228                The timer resets after each change to de-duplicate file change events.
229
230            reload:
231                If False, neither the browser nor the server will be reloaded.
232                This is useful for triggering js/css builds, which you might
233                pair with a separate WatchCond on the build output directory
234                that does the reloading.
235
236            server_reload:
237                If False, do not reload the daemon process after a change; just the browser.
238        """
239        if isinstance(path_glob, str):
240            self.path_glob = [path_glob]
241        else:
242            self.path_glob = path_glob
243
244        if not ignore_glob:
245            ignore_glob = []
246
247        if isinstance(ignore_glob, str):
248            self.ignore_glob = [ignore_glob]
249        else:
250            self.ignore_glob = ignore_glob
251
252        # We assume you want all paths to be relative to CWD
253
254        for i, p in enumerate(self.path_glob):
255            p = os.path.relpath(p, CWD) + ("/" if p.endswith("/") else "")
256            self.path_glob[i] = p
257
258        for i, p in enumerate(self.ignore_glob):
259            p = os.path.relpath(p, CWD) + ("/" if p.endswith("/") else "")
260            self.ignore_glob[i] = p
261
262        self.server_reload = server_reload
263        self.reload = reload
264        if action is None:
265            self.task = Task(None, delay)
266        elif isinstance(action, ShellCommand):
267            self.task = ProcessTask(action, delay)
268        else:
269            if not callable(action):
270                raise ValueError("Action must be a ShellCommand or a callable")
271
272            self.task = Task(action, delay)
273
274    def try_path_hit(self, path: str) -> bool:
275        """Check if a given path matches any of the glob patterns."""
276
277        for pattern in self.ignore_glob:
278            if glob_matcher(pattern, path):
279                return False
280
281        for pattern in self.path_glob:
282            if glob_matcher(pattern, path):
283                return True
284
285        return False

294class Watcher:
295    """Simple file watcher with support for both stat and inotify."""
296
297    def __init__(self, conds: list[WatchCond], force_polling: bool):
298        self.conds = conds
299        self.watch_globs = []
300        self.mtimes: dict[str, int] = {}
301        self.force_polling = force_polling
302
303        for cond in conds:
304            for g in cond.path_glob:
305                self.watch_globs.append(g)
306
307        self.rust_watches = []
308        if not self.force_polling:
309            recursive = []
310            non_recursive = []
311            for path, is_recursive in self._get_fswatch_dirs():
312                if is_recursive:
313                    recursive.append(path)
314                else:
315                    non_recursive.append(path)
316            if recursive:
317                rn = RustNotify(
318                    recursive,
319                    recursive=True,
320                    debug=False,
321                    force_polling=False,
322                    poll_delay_ms=0,
323                    ignore_permission_denied=True,
324                )
325                self.rust_watches.append(rn)
326            if non_recursive:
327                nr = RustNotify(
328                    non_recursive,
329                    recursive=False,
330                    debug=False,
331                    force_polling=False,
332                    poll_delay_ms=0,
333                    ignore_permission_denied=True,
334                )
335                self.rust_watches.append(nr)
336
337    def overhead(self):
338        if not self.force_polling:
339            recursive_count = 0
340            dirs = set()
341            fswatch_dirs = self._get_fswatch_dirs()
342            for path, recursive in fswatch_dirs:
343                if recursive:
344                    recursive_count += 1
345                dirs.add(path)
346            return {
347                "path_count": len(dirs),
348                "recursive_count": recursive_count,
349                "paths": dirs,
350            }
351        else:
352            paths = list(self._resolve_paths())
353            return {"path_count": len(list(paths)), "paths": paths}
354
355    def _get_fswatch_dirs(self):
356        # We handle a few kinds of watch expressions:
357        # dir/: Watch dir/ recursively
358        # ./file.py: Watch just a single file
359        # ./src/*.py: Watch ./src/ non-recursively for .py files
360        # dir/**/*.py: Watch dir/ recursively
361        # **/*.py: watch ./ recursively
362        dirs = set()
363        for cond in self.conds:
364            for pattern in cond.path_glob:
365                recursive = False
366                if pattern.endswith("/"):
367                    recursive = True
368
369                path_component = Path(pattern)
370                if "**" in path_component.parts:
371                    recursive = True
372                    for i, part in reversed(
373                        list(enumerate(path_component.parts))
374                    ):
375                        if part == "**":
376                            pattern = str(Path(*path_component.parts[0:i]))
377                            if not pattern:
378                                pattern = "."
379
380                for p in glob.glob(pattern):
381                    path = Path(p)
382                    if not recursive and path.is_file():
383                        # If this is a file glob, get the containing dir
384                        p = path.parent
385
386                    as_t = (str(p), recursive)
387                    if as_t in dirs:
388                        continue
389
390                    dirs.add(as_t)
391
392        return dirs
393
394    def _get_matching_rules(self, path) -> list[WatchCond]:
395        """Find the first glob pattern that matches the path."""
396        rules = []
397        for cond in self.conds:
398            # Check if the path matches the glob pattern
399            if cond.try_path_hit(path):
400                rules.append(cond)
401
402        return rules
403
404    def _resolve_paths(self):
405        """Resolve all glob patterns to file paths."""
406
407        # Add all files matching watch patterns
408        for pattern in self.watch_globs:
409            for match in glob.iglob(pattern, recursive=True):
410                for cond in self.conds:
411                    if cond.try_path_hit(match):
412                        yield match
413
414    def stat_watcher(self):
415        paths = self._resolve_paths()
416        changes = []
417        current_mtimes = {}
418        changed = set()
419        if not self.mtimes:
420            current_mtimes = self.mtimes
421
422        for path in paths:
423            try:
424                if path in changed:
425                    # Already processed this file
426                    continue
427
428                st = os.stat(path)
429                if not stat.S_ISREG(st.st_mode):
430                    # Not a regular file, skip it
431                    continue
432                mtime = int(st.st_mtime)
433            except OSError:
434                # Might be deleted, we'll catch it later
435                continue
436
437            current_mtimes[path] = int(mtime)
438            old_mtime = self.mtimes.get(path, None)
439
440            if old_mtime != mtime:
441                matching_rules = self._get_matching_rules(path)
442                changes.append(Hit(path, matching_rules))
443                changed.add(path)
444
445        # Check for deleted files
446        for path in self.mtimes.keys():
447            if path not in current_mtimes:
448                matching_rules = self._get_matching_rules(path)
449                if matching_rules:
450                    changes.append(Hit(path, matching_rules))
451
452        self.mtimes = current_mtimes
453        return changes
454
455    def changed(self) -> list[Hit]:
456        """Check if any watched files have changed.
457
458        Returns (path, matching_glob) if a change was detected, None otherwise.
459        """
460        if self.force_polling:
461            return self.stat_watcher()
462
463        changes = []
464        for watch in self.rust_watches:
465            # Check and immediately return
466            result = watch.watch(
467                debounce_ms=100, step_ms=1, timeout_ms=1, stop_event=None
468            )
469            if result == "signal":
470                # Probably a ctrl + C
471                raise KeyboardInterrupt()
472            if result == "timeout":
473                continue
474            # We don't have a stop event and that's the only other type.
475            # Alas, ignore it.
476            if isinstance(result, str):
477                print(
478                    "watcher: Not sure what to do with",
479                    repr(result),
480                    " submit a git issue to html-compose",
481                )
482                continue
483
484            assert isinstance(result, set), (
485                f"Unexpected result type: {type(result)}"
486            )
487
488            result_tuples: set[tuple[int, str]] = result
489            for watch_id, path in result_tuples:
490                # RustWatch returns full paths, convert to relative
491                path = os.path.relpath(path, CWD)
492                matching_rules = self._get_matching_rules(path)
493                if matching_rules:
494                    changes.append(Hit(path, matching_rules))
495
496        return changes