pyrtan / sourcev0.19.0 · main

pyrtan / src / pyrtan / registry.py

5,918 bytes · last changed 2026-10-25 in Read sentences with rules: rewrites, typed slots, synonyms, typos, pyrtan match

1"""The trees a service knows about, and how a sentence finds one.
2
3A registry loads every ``*.toml`` under one or more directories, reloads a
4file when its modification time changes (so trees can be edited while a
5service runs), and matches intents against the triggers of everything
6loaded.
7"""
8
9from __future__ import annotations
10
11import os
12import threading
13from dataclasses import dataclass, field
14from pathlib import Path
15from typing import Any, Iterable
16
17import datetime as _dt
18
19from .intent import Match, Reading, Trigger, read
20from .tree import Tree, TreeError, load_tree
21
22
23class Ambiguous(LookupError):
24 def __init__(self, text: str, matches: list[Match]) -> None:
25 self.text = text
26 self.matches = matches
27 names = ", ".join(f"{m.tree} ({m.trigger.pattern!r})" for m in matches)
28 super().__init__(f"{text!r} matches more than one tree equally well: {names}")
29
30
31class NoMatch(LookupError):
32 pass
33
34
35@dataclass
36class _Loaded:
37 tree: Tree
38 path: Path | None
39 mtime: float
40
41
42@dataclass
43class Registry:
44 directories: list[Path] = field(default_factory=list)
45 _trees: dict[str, _Loaded] = field(default_factory=dict, init=False, repr=False)
46 _lock: threading.RLock = field(default_factory=threading.RLock, init=False, repr=False)
47 errors: dict[str, str] = field(default_factory=dict, init=False)
48
49 @classmethod
50 def from_paths(cls, paths: Iterable[str | Path]) -> "Registry":
51 reg = cls()
52 for p in paths:
53 p = Path(p)
54 if p.is_dir():
55 reg.directories.append(p)
56 else:
57 reg.add_file(p)
58 reg.reload()
59 return reg
60
61 # ------------------------------------------------------------ loading
62
63 def add(self, tree: Tree) -> None:
64 with self._lock:
65 self._trees[tree.name] = _Loaded(tree, None, 0.0)
66
67 def add_file(self, path: str | Path) -> Tree:
68 path = Path(path)
69 tree = load_tree(path)
70 with self._lock:
71 self._trees[tree.name] = _Loaded(tree, path, path.stat().st_mtime)
72 self.errors.pop(str(path), None)
73 return tree
74
75 def reload(self) -> None:
76 """Load new files, reload changed ones, drop deleted ones."""
77 with self._lock:
78 seen: set[Path] = set()
79 for d in self.directories:
80 for path in sorted(d.rglob("*.toml")):
81 if path.name == "pyproject.toml":
82 continue
83 seen.add(path)
84 try:
85 mtime = path.stat().st_mtime
86 except OSError:
87 continue
88 current = next((l for l in self._trees.values() if l.path == path), None)
89 if current is not None and current.mtime == mtime:
90 continue
91 try:
92 tree = load_tree(path)
93 except TreeError as e:
94 self.errors[str(path)] = str(e)
95 continue
96 self.errors.pop(str(path), None)
97 if current is not None and current.tree.name != tree.name:
98 del self._trees[current.tree.name]
99 self._trees[tree.name] = _Loaded(tree, path, mtime)
100 for name, loaded in list(self._trees.items()):
101 if loaded.path is not None and loaded.path not in seen and any(loaded.path.is_relative_to(d) for d in self.directories):
102 if not loaded.path.exists():
103 del self._trees[name]
104
105 # ------------------------------------------------------------ lookup
106
107 def names(self) -> list[str]:
108 with self._lock:
109 return sorted(self._trees)
110
111 def trees(self) -> list[Tree]:
112 with self._lock:
113 return [l.tree for _, l in sorted(self._trees.items())]
114
115 def get(self, name: str) -> Tree | None:
116 with self._lock:
117 loaded = self._trees.get(name)
118 return loaded.tree if loaded else None
119
120 def __contains__(self, name: object) -> bool:
121 return isinstance(name, str) and self.get(name) is not None
122
123 def __len__(self) -> int:
124 return len(self._trees)
125
126 def candidates(self) -> list[tuple[str, Trigger]]:
127 return [(t.name, trig) for t in self.trees() for trig in t.triggers]
128
129 def read(self, text: str, now: _dt.datetime | _dt.date | None = None) -> Reading:
130 """How ``text`` reads against every trigger here: the rewrites, the
131 verdicts, the winner or the tie."""
132 return read(text, self.candidates(), now)
133
134 def match(self, text: str, now: _dt.datetime | _dt.date | None = None) -> Match:
135 """The one tree that answers ``text``, or :class:`NoMatch` /
136 :class:`Ambiguous`."""
137 matches = self.read(text, now).matches
138 if not matches:
139 raise NoMatch(f"no tree answers {text!r}")
140 if len(matches) > 1:
141 # Same tree matched twice at equal specificity: take the first.
142 if len({m.tree for m in matches}) == 1:
143 return matches[0]
144 raise Ambiguous(text, matches)
145 return matches[0]
146
147 def capabilities(self) -> list[dict[str, Any]]:
148 return [t.capability() for t in self.trees()]
149
150 def gather(self, root: str | Path) -> int:
151 """Add a directory and load it. Returns how many trees are loaded."""
152 root = Path(root)
153 if root not in self.directories:
154 self.directories.append(root)
155 self.reload()
156 return len(self)
157
158
159def default_paths() -> list[Path]:
160 """Where ``pyrtan`` looks when no path is given: ``$PYRTAN_TREES`` (colon
161 separated) or ``./trees``."""
162 env = os.environ.get("PYRTAN_TREES")
163 if env:
164 return [Path(p) for p in env.split(os.pathsep) if p]
165 return [Path("trees")]