pyrtan / src / pyrtan / watch.py
12,685 bytes · last changed 2026-10-26 in Watches and growing from a gap: trees that run when something speaks, and the branch a gap asks for drafted by rules; release 0.9.0
| 1 | """Watches: a tree that runs when something speaks, without being asked. |
| 2 | |
| 3 | A tree may carry ``[[watches]]``: a clock (``every = "5m"``), a file |
| 4 | (``file = "/var/log/syslog"``: each time lines are appended), a queue |
| 5 | (``queue = { url, topic }``: each message) or a serial line (``serial = |
| 6 | { device, baud }``: each line). Each watch may give ``inputs``, templates |
| 7 | over ``event`` (what spoke), ``env`` and ``now``. A :class:`Watcher` holds |
| 8 | one thread per watch over a service's trees, runs the tree through the |
| 9 | service's own runner (so delegations reach the hub and gaps are kept), |
| 10 | and remembers for each watch when it last fired, what the outcome was, |
| 11 | and the *path* the run took: the sequence of nodes. When the path of a |
| 12 | scheduled run differs from the last one, the watch records that too. |
| 13 | That is how a person learns a site has changed before anyone asks. |
| 14 | |
| 15 | The trace of a watched run begins with a ``watch`` step, as an asked |
| 16 | run's begins with ``intent``. Watches are static: a path, a topic, a |
| 17 | device is literal in the file and checked against the tree's lists by |
| 18 | ``pyrtan check``, and a service started ``--egress public`` watches no |
| 19 | file and no serial line, as it reads none. |
| 20 | |
| 21 | pyrtan serve trees --watch # a service that also watches |
| 22 | pyrtan hub --trees trees --watch # likewise, in a hub |
| 23 | pyrtan watch trees # nothing served: just the watches, printing each run |
| 24 | """ |
| 25 | |
| 26 | from __future__ import annotations |
| 27 | |
| 28 | import datetime as _dt |
| 29 | import json |
| 30 | import os |
| 31 | import threading |
| 32 | import time |
| 33 | from typing import Any, Callable |
| 34 | |
| 35 | from . import adapters |
| 36 | from .nlp import ParseError, parse_duration |
| 37 | from .runtime import Outcome |
| 38 | from .template import TemplateError, render |
| 39 | from .trace import Step, Trace |
| 40 | from .transport import EgressRefused, TransportError |
| 41 | from .tree import Tree |
| 42 | |
| 43 | SYNC_EVERY = 10.0 # how often the watcher looks for trees that changed on disk |
| 44 | FILE_POLL = 2.0 # how often a file watch looks at the file |
| 45 | RETRY_AFTER = 5.0 # after a stream fails, before reconnecting |
| 46 | HISTORY = 20 # fires remembered per watch |
| 47 | |
| 48 | |
| 49 | def every_seconds(value: Any) -> int: |
| 50 | if isinstance(value, (int, float)) and not isinstance(value, bool): |
| 51 | return int(value) |
| 52 | try: |
| 53 | return parse_duration(str(value)) |
| 54 | except ParseError: |
| 55 | return 0 |
| 56 | |
| 57 | |
| 58 | class Watch: |
| 59 | """One watch of one tree, on its own thread.""" |
| 60 | |
| 61 | def __init__(self, watcher: "Watcher", tree_name: str, index: int, spec: dict[str, Any]) -> None: |
| 62 | self.watcher = watcher |
| 63 | self.tree_name = tree_name |
| 64 | self.index = index |
| 65 | self.spec = spec |
| 66 | self.kind = next(k for k in ("every", "file", "queue", "serial") if k in spec) |
| 67 | self.name = str(spec.get("name") or f"{tree_name}#{index}") |
| 68 | self.key = json.dumps([tree_name, index, spec], sort_keys=True, default=str) |
| 69 | self.stop = threading.Event() |
| 70 | self.thread: threading.Thread | None = None |
| 71 | self.fires = 0 |
| 72 | self.errors = 0 |
| 73 | self.last_error: str | None = None |
| 74 | self.last_at: float | None = None |
| 75 | self.last_outcome: dict[str, Any] | None = None |
| 76 | self.last_path: list[str] | None = None |
| 77 | self.path_changed_at: float | None = None |
| 78 | self.previous_path: list[str] | None = None |
| 79 | self.history: list[dict[str, Any]] = [] |
| 80 | self.lock = threading.Lock() |
| 81 | |
| 82 | def start(self) -> None: |
| 83 | self.thread = threading.Thread(target=self._run, name=f"pyrtan-watch-{self.name}", daemon=True) |
| 84 | self.thread.start() |
| 85 | |
| 86 | def _run(self) -> None: |
| 87 | egress = self.watcher.egress |
| 88 | if self.kind in ("file", "serial") and egress != "any": |
| 89 | self._fail(f"{self.kind}: this service reaches only public addresses (--egress {egress}); it watches no file and no serial line") |
| 90 | return |
| 91 | while not self.stop.is_set(): |
| 92 | try: |
| 93 | if self.kind == "every": |
| 94 | self._every() |
| 95 | elif self.kind == "file": |
| 96 | self._file() |
| 97 | elif self.kind == "queue": |
| 98 | adapters.mqtt_stream(self.spec["queue"], lambda m: self.fire({"kind": "queue", **m}), self.stop, egress) |
| 99 | elif self.kind == "serial": |
| 100 | adapters.serial_lines(self.spec["serial"], lambda line: self.fire({"kind": "serial", "device": self.spec["serial"].get("device"), "body": line, "lines": [line]}), self.stop) |
| 101 | return |
| 102 | except EgressRefused as e: |
| 103 | self._fail(str(e)) |
| 104 | return |
| 105 | except TransportError as e: |
| 106 | self._fail(str(e)) |
| 107 | except Exception as e: # a bug, not a line: keep the watch alive, say so |
| 108 | self._fail(f"{type(e).__name__}: {e}") |
| 109 | self.stop.wait(RETRY_AFTER) |
| 110 | |
| 111 | def _fail(self, text: str) -> None: |
| 112 | with self.lock: |
| 113 | self.errors += 1 |
| 114 | self.last_error = text |
| 115 | self.watcher.note(f"watch {self.name}: {text}") |
| 116 | |
| 117 | def _every(self) -> None: |
| 118 | seconds = max(1, every_seconds(self.spec["every"])) |
| 119 | n = 0 |
| 120 | while not self.stop.is_set(): |
| 121 | n += 1 |
| 122 | self.fire({"kind": "every", "n": n, "at": _dt.datetime.now(_dt.timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z"), "every": seconds}) |
| 123 | if self.stop.wait(seconds): |
| 124 | return |
| 125 | |
| 126 | def _file(self) -> None: |
| 127 | path = adapters.absolute_path(self.spec["file"]) |
| 128 | offset: int | None = None # None until the file is first seen: a watch starts at the end |
| 129 | while not self.stop.is_set(): |
| 130 | try: |
| 131 | st = os.stat(path) |
| 132 | except OSError: |
| 133 | offset = None |
| 134 | if self.stop.wait(FILE_POLL): |
| 135 | return |
| 136 | continue |
| 137 | if offset is None or st.st_size < offset: |
| 138 | offset = st.st_size if offset is None else 0 # new file: start at its end; truncated: read it all |
| 139 | if st.st_size < (offset or 0): |
| 140 | offset = 0 |
| 141 | if st.st_size > offset: |
| 142 | with open(path, "rb") as f: |
| 143 | f.seek(offset) |
| 144 | raw = f.read(adapters.MAX_READ) |
| 145 | offset += len(raw) |
| 146 | text = raw.decode("utf-8", errors="replace") |
| 147 | lines = text.splitlines() |
| 148 | if lines: |
| 149 | self.fire({"kind": "file", "path": path, "body": text, "lines": lines, "size": st.st_size}) |
| 150 | if self.stop.wait(FILE_POLL): |
| 151 | return |
| 152 | |
| 153 | def fire(self, event: dict[str, Any]) -> Outcome | None: |
| 154 | """Run the tree for one event. Returns the outcome, or None when the |
| 155 | tree is gone or its inputs do not render.""" |
| 156 | tree = self.watcher.registry.get(self.tree_name) |
| 157 | if tree is None: |
| 158 | self._fail(f"tree {self.tree_name!r} is gone") |
| 159 | return None |
| 160 | scope = {"event": event, "env": self.watcher.env, "now": _now_scope(self.watcher.clock())} |
| 161 | try: |
| 162 | inputs = {k: render(v, scope) for k, v in (self.spec.get("inputs") or {}).items()} |
| 163 | except TemplateError as e: |
| 164 | self._fail(f"inputs: {e}") |
| 165 | return None |
| 166 | trace = Trace(tree.name) |
| 167 | trace.add(Step("watch", "watch", _summary(self.kind, event), observed=_observed(event), next=tree.start)) |
| 168 | outcome = self.watcher.run(tree, inputs, trace) |
| 169 | path = [s.node for s in outcome.trace.steps] if outcome.trace else [] |
| 170 | now = time.time() |
| 171 | with self.lock: |
| 172 | self.fires += 1 |
| 173 | self.last_at = now |
| 174 | self.last_outcome = {k: v for k, v in outcome.to_dict(with_trace=False).items() if k in ("status", "reason", "say", "output")} |
| 175 | if self.last_path is not None and path != self.last_path: |
| 176 | self.previous_path = self.last_path |
| 177 | self.path_changed_at = now |
| 178 | self.last_path = path |
| 179 | self.history.append({"at": now, "status": outcome.status, "say": outcome.say, "path": path, "event": _observed(event)}) |
| 180 | del self.history[:-HISTORY] |
| 181 | self.watcher.after(self, event, outcome) |
| 182 | return outcome |
| 183 | |
| 184 | def to_dict(self) -> dict[str, Any]: |
| 185 | with self.lock: |
| 186 | return { |
| 187 | "tree": self.tree_name, "index": self.index, "name": self.name, "kind": self.kind, "what": _what(self.kind, self.spec), |
| 188 | "fires": self.fires, "errors": self.errors, "last_error": self.last_error, "last_at": self.last_at, |
| 189 | "last": self.last_outcome, "path": self.last_path, "path_changed_at": self.path_changed_at, "previous_path": self.previous_path, |
| 190 | "alive": bool(self.thread and self.thread.is_alive()), |
| 191 | } |
| 192 | |
| 193 | |
| 194 | class Watcher: |
| 195 | """Every watch of every tree in a registry, kept in step with the files |
| 196 | on disk. ``run(tree, inputs, trace)`` is how a run happens (a service |
| 197 | gives its own, so gaps are counted and delegations route); ``after`` |
| 198 | is told about each fire.""" |
| 199 | |
| 200 | def __init__(self, registry: Any, run: Callable[[Tree, dict[str, Any], Trace], Outcome], *, env: dict[str, Any] | None = None, egress: str = "any", clock: Callable[[], _dt.datetime] | None = None, after: Callable[["Watch", dict[str, Any], Outcome], None] | None = None, note: Callable[[str], None] | None = None) -> None: |
| 201 | self.registry = registry |
| 202 | self.run = run |
| 203 | self.env = env or {} |
| 204 | self.egress = egress |
| 205 | self.clock = clock or (lambda: _dt.datetime.now(_dt.timezone.utc)) |
| 206 | self.after = after or (lambda w, e, o: None) |
| 207 | self.note = note or (lambda text: None) |
| 208 | self.watches: dict[str, Watch] = {} |
| 209 | self._stop = threading.Event() |
| 210 | self._thread: threading.Thread | None = None |
| 211 | self._lock = threading.Lock() |
| 212 | |
| 213 | def sync(self) -> tuple[int, int]: |
| 214 | """Start watches for trees that appeared or changed, stop the ones |
| 215 | that vanished. Returns (started, stopped).""" |
| 216 | self.registry.reload() |
| 217 | wanted: dict[str, Watch] = {} |
| 218 | for tree in self.registry.trees(): |
| 219 | for i, spec in enumerate(tree.watches): |
| 220 | w = Watch(self, tree.name, i, spec) |
| 221 | wanted[w.key] = w |
| 222 | started = stopped = 0 |
| 223 | with self._lock: |
| 224 | for key, w in list(self.watches.items()): |
| 225 | if key not in wanted: |
| 226 | w.stop.set() |
| 227 | del self.watches[key] |
| 228 | stopped += 1 |
| 229 | for key, w in wanted.items(): |
| 230 | if key not in self.watches: |
| 231 | self.watches[key] = w |
| 232 | w.start() |
| 233 | started += 1 |
| 234 | return started, stopped |
| 235 | |
| 236 | def start(self) -> "Watcher": |
| 237 | self.sync() |
| 238 | |
| 239 | def loop() -> None: |
| 240 | while not self._stop.wait(SYNC_EVERY): |
| 241 | try: |
| 242 | self.sync() |
| 243 | except Exception as e: |
| 244 | self.note(f"watcher: {e}") |
| 245 | |
| 246 | self._thread = threading.Thread(target=loop, name="pyrtan-watcher", daemon=True) |
| 247 | self._thread.start() |
| 248 | return self |
| 249 | |
| 250 | def stop(self) -> None: |
| 251 | self._stop.set() |
| 252 | with self._lock: |
| 253 | for w in self.watches.values(): |
| 254 | w.stop.set() |
| 255 | |
| 256 | def status(self) -> dict[str, Any]: |
| 257 | with self._lock: |
| 258 | watches = [w.to_dict() for w in self.watches.values()] |
| 259 | watches.sort(key=lambda w: (w["tree"], w["index"])) |
| 260 | return {"count": len(watches), "watches": watches} |
| 261 | |
| 262 | def __len__(self) -> int: |
| 263 | return len(self.watches) |
| 264 | |
| 265 | |
| 266 | def _what(kind: str, spec: dict[str, Any]) -> str: |
| 267 | if kind == "every": |
| 268 | return f"every {spec['every']}" |
| 269 | if kind == "file": |
| 270 | return str(spec["file"]) |
| 271 | if kind == "queue": |
| 272 | return f"{spec['queue'].get('topic')} at {spec['queue'].get('url')}" |
| 273 | return str(spec["serial"].get("device")) |
| 274 | |
| 275 | |
| 276 | def _summary(kind: str, event: dict[str, Any]) -> str: |
| 277 | if kind == "every": |
| 278 | return f"every {event.get('every')} s, run {event.get('n')}" |
| 279 | if kind == "file": |
| 280 | return f"{event.get('path')}: {len(event.get('lines') or [])} new line(s)" |
| 281 | if kind == "queue": |
| 282 | return f"{event.get('topic')}: {len(str(event.get('body') or ''))} B" |
| 283 | return f"{event.get('device')}: {str(event.get('body') or '')[:60]}" |
| 284 | |
| 285 | |
| 286 | def _observed(event: dict[str, Any]) -> dict[str, Any]: |
| 287 | out = {k: v for k, v in event.items() if k != "lines"} |
| 288 | if isinstance(out.get("body"), str) and len(out["body"]) > 200: |
| 289 | out["body"] = out["body"][:200] + "…" |
| 290 | return out |
| 291 | |
| 292 | |
| 293 | def _now_scope(now: _dt.datetime) -> dict[str, Any]: |
| 294 | from .runtime import _now |
| 295 | |
| 296 | return _now(now) |