Skills › Scientific Agent Skills › ontology-term-resolution › resolve_terms.py
resolve_terms.py
264 lines · skills/ontology-term-resolution/scripts/resolve_terms.py at 5f2bc03
1#!/usr/bin/env python3
2# /// script
3# requires-python = ">=3.11"
4# dependencies = []
5# ///
6"""Resolve free-text labels to ontology terms via EBI OLS4.
8Never emit an ontology ID from memory -- run this instead. Each input string is
9searched with an escalating strategy and every returned candidate is labelled
10with how it actually matched, so a partial hit can never be mistaken for an
11exact one.
13Strategies, tried in order until one returns candidates:
15 exact exact=true restricted to label and synonym fields
16 token exact=true across all indexed fields (whole-word match)
17 fulltext unrestricted relevance search
19Examples:
20 # single term, constrained to the ontology that should define it
21 uv run resolve_terms.py "liver" --ontology uberon
23 # a column of tissue names, requiring anatomical entities only
24 uv run resolve_terms.py --input tissues.txt \\
25 --ontology uberon --branch UBERON:0000465 --format tsv -o resolved.tsv
27 # accept exact matches only; unresolved rows are reported, not guessed
28 uv run resolve_terms.py "kidney cortex" "left ventrical" --exact-only
29"""
30from __future__ import annotations
32import argparse
33import csv
34import json
35import sys
36from pathlib import Path
38sys.path.insert(0, str(Path(__file__).resolve().parent))
40from ols_client import ( # noqa: E402
41 OlsError,
42 iri_for,
43 is_curie,
44 rank_candidates,
45 search,
46)
48STRATEGIES = (
49 ("exact", {"exact": True, "query_fields": "label,synonym"}),
50 ("token", {"exact": True, "query_fields": None}),
51 ("fulltext", {"exact": False, "query_fields": None}),
52)
54TSV_COLUMNS = (
55 "query",
56 "rank",
57 "curie",
58 "iri",
59 "label",
60 "ontology",
61 "match_type",
62 "strategy",
63 "defining_ontology",
64)
67def read_inputs(args: argparse.Namespace) -> list[str]:
68 """Collect query strings from positional args, a file, or stdin."""
69 values: list[str] = list(args.text)
70 if args.input:
71 raw = (
72 sys.stdin.read()
73 if args.input == "-"
74 else Path(args.input).read_text(encoding="utf-8")
75 )
76 for line in raw.splitlines():
77 line = line.strip()
78 if line and not line.startswith("#"):
79 values.append(line)
80 if not values and not sys.stdin.isatty():
81 for line in sys.stdin.read().splitlines():
82 line = line.strip()
83 if line and not line.startswith("#"):
84 values.append(line)
85 # Preserve order, drop duplicates.
86 seen: set[str] = set()
87 unique = []
88 for value in values:
89 if value not in seen:
90 seen.add(value)
91 unique.append(value)
92 return unique
95def resolve_one(
96 text: str,
97 *,
98 ontology: str | None,
99 subtree_iri: str | None,
100 rows: int,
101 exact_only: bool,
102) -> dict:
103 """Run the strategy ladder for one string and return ranked candidates."""
104 strategies = STRATEGIES[:1] if exact_only else STRATEGIES
105 for name, options in strategies:
106 docs = search(
107 text,
108 ontology=ontology,
109 subtree_iri=subtree_iri,
110 rows=rows,
111 **options,
112 )
113 ranked = rank_candidates(text, docs)
114 if exact_only:
115 ranked = [candidate for candidate in ranked
116 if candidate["match_type"] in {"exact_label", "exact_synonym"}]
117 if ranked:
118 return {"query": text, "strategy": name, "candidates": ranked[:rows]}
119 return {"query": text, "strategy": strategies[-1][0], "candidates": []}
122def to_rows(results: list[dict]) -> list[dict]:
123 """Flatten results into one row per candidate, or one row if unresolved."""
124 rows: list[dict] = []
125 for result in results:
126 if not result["candidates"]:
127 rows.append(
128 {
129 "query": result["query"],
130 "rank": 1,
131 "curie": "",
132 "iri": "",
133 "label": "",
134 "ontology": "",
135 "match_type": "unresolved",
136 "strategy": result["strategy"],
137 "defining_ontology": "",
138 }
139 )
140 continue
141 for position, candidate in enumerate(result["candidates"], start=1):
142 rows.append(
143 {
144 "query": result["query"],
145 "rank": position,
146 "curie": candidate.get("obo_id", ""),
147 "iri": candidate.get("iri", ""),
148 "label": candidate.get("label", ""),
149 "ontology": candidate.get("ontology_name", ""),
150 "match_type": candidate["match_type"],
151 "strategy": result["strategy"],
152 "defining_ontology": str(
153 bool(candidate.get("is_defining_ontology"))
154 ).lower(),
155 }
156 )
157 return rows
160def write_output(results: list[dict], fmt: str, output: str | None) -> None:
161 """Emit TSV or JSON to a path or stdout."""
162 stream = open(output, "w", encoding="utf-8", newline="") if output else sys.stdout
163 try:
164 if fmt == "json":
165 json.dump(results, stream, indent=2)
166 stream.write("\n")
167 else:
168 writer = csv.DictWriter(
169 stream, fieldnames=TSV_COLUMNS, delimiter="\t", lineterminator="\n"
170 )
171 writer.writeheader()
172 writer.writerows(to_rows(results))
173 finally:
174 if output:
175 stream.close()
178def build_parser() -> argparse.ArgumentParser:
179 parser = argparse.ArgumentParser(
180 description="Resolve free-text labels to ontology terms via EBI OLS4.",
181 formatter_class=argparse.RawDescriptionHelpFormatter,
182 )
183 parser.add_argument("text", nargs="*", help="strings to resolve")
184 parser.add_argument(
185 "--input",
186 help="file with one string per line ('-' for stdin); # lines are comments",
187 )
188 parser.add_argument(
189 "--ontology",
190 help="restrict to OLS ontology ids, comma separated (e.g. uberon,cl)",
191 )
192 parser.add_argument(
193 "--branch",
194 help="require candidates to be descendants of this CURIE (e.g. UBERON:0000465)",
195 )
196 parser.add_argument(
197 "--top", type=int, default=5, help="candidates to report per query (default 5)"
198 )
199 parser.add_argument(
200 "--exact-only",
201 action="store_true",
202 help="only exact label/synonym matches; report anything else as unresolved",
203 )
204 parser.add_argument(
205 "--format", choices=("tsv", "json"), default="tsv", help="output format"
206 )
207 parser.add_argument("-o", "--output", help="write here instead of stdout")
208 return parser
211def main(argv: list[str] | None = None) -> int:
212 args = build_parser().parse_args(argv)
213 if args.top < 1:
214 print("--top must be positive.", file=sys.stderr)
215 return 2
216 queries = read_inputs(args)
217 if not queries:
218 print("No input strings given. See --help.", file=sys.stderr)
219 return 2
221 subtree_iri = None
222 if args.branch:
223 if not is_curie(args.branch):
224 print(f"--branch expects a CURIE, got {args.branch!r}", file=sys.stderr)
225 return 2
226 try:
227 subtree_iri = iri_for(args.branch)
228 except OlsError as exc:
229 print(f"Could not resolve --branch {args.branch}: {exc}", file=sys.stderr)
230 return 2
231 if not subtree_iri:
232 print(f"--branch term not found in OLS: {args.branch}", file=sys.stderr)
233 return 2
235 results = []
236 for query in queries:
237 try:
238 results.append(
239 resolve_one(
240 query,
241 ontology=args.ontology,
242 subtree_iri=subtree_iri,
243 rows=args.top,
244 exact_only=args.exact_only,
245 )
246 )
247 except OlsError as exc:
248 print(f"OLS lookup failed for {query!r}: {exc}", file=sys.stderr)
249 return 2
251 write_output(results, args.format, args.output)
253 unresolved = [r["query"] for r in results if not r["candidates"]]
254 if unresolved:
255 print(
256 f"{len(unresolved)}/{len(results)} unresolved: "
257 + ", ".join(repr(q) for q in unresolved[:10]),
258 file=sys.stderr,
259 )
260 return 0
263if __name__ == "__main__":
264 raise SystemExit(main())