Skills › Scientific Agent Skills › ontology-term-resolution › map_terms.py
map_terms.py
233 lines · skills/ontology-term-resolution/scripts/map_terms.py at 5f2bc03
1#!/usr/bin/env python3
2# /// script
3# requires-python = ">=3.11"
4# dependencies = []
5# ///
6"""Map lab shorthand to ontology terms via EBI ZOOMA.
8Use this after ``resolve_terms.py`` returns ``unresolved`` or only ``partial``
9hits on strings like ``PBMC`` or ``WT``. Every HIGH/GOOD hit is still only a
10candidate — run ``validate_terms.py`` on the CURIE before writing it down.
12``--ontology`` is required. Unfiltered ZOOMA annotate returns FOODON, XAO, and
13BTO alongside UBERON for ``liver``, all at HIGH confidence.
15Examples:
16 uv run map_terms.py PBMC --ontology cl
17 uv run map_terms.py liver --ontology uberon --property-type "organism part"
18 uv run map_terms.py --input shorthand.txt --ontology uberon,cl --high-confidence-only
19"""
20from __future__ import annotations
22import argparse
23import csv
24import json
25import sys
26from pathlib import Path
28sys.path.insert(0, str(Path(__file__).resolve().parent))
30from zooma_client import ( # noqa: E402
31 ZoomaError,
32 annotate,
33 flatten_hit,
34)
36TSV_COLUMNS = (
37 "query",
38 "rank",
39 "curie",
40 "iri",
41 "confidence",
42 "safe",
43 "evidence",
44 "source",
45 "property_type",
46 "match_type",
47)
50def read_inputs(args: argparse.Namespace) -> list[str]:
51 """Collect query strings from positional args, a file, or stdin."""
52 values: list[str] = list(args.text)
53 if args.input:
54 raw = (
55 sys.stdin.read()
56 if args.input == "-"
57 else Path(args.input).read_text(encoding="utf-8")
58 )
59 for line in raw.splitlines():
60 line = line.strip()
61 if line and not line.startswith("#"):
62 values.append(line)
63 if not values and not sys.stdin.isatty():
64 for line in sys.stdin.read().splitlines():
65 line = line.strip()
66 if line and not line.startswith("#"):
67 values.append(line)
68 seen: set[str] = set()
69 unique = []
70 for value in values:
71 if value not in seen:
72 seen.add(value)
73 unique.append(value)
74 return unique
77def map_one(
78 text: str,
79 *,
80 ontologies: list[str],
81 property_type: str | None,
82 top: int,
83 safe_only: bool,
84) -> dict:
85 """Annotate one string and return ranked, flattened candidates."""
86 hits = annotate(text, ontologies=ontologies, property_type=property_type)
87 candidates: list[dict] = []
88 for hit in hits:
89 candidates.extend(flatten_hit(hit))
90 if safe_only:
91 candidates = [row for row in candidates if row["safe"]]
92 # HIGH before GOOD before MEDIUM/LOW; preserve server order within a tier.
93 rank = {"HIGH": 0, "GOOD": 1, "MEDIUM": 2, "LOW": 3}
94 candidates.sort(key=lambda row: rank.get(row["confidence"], 9))
95 return {"query": text, "candidates": candidates[:top]}
98def to_rows(results: list[dict]) -> list[dict]:
99 rows: list[dict] = []
100 for result in results:
101 if not result["candidates"]:
102 rows.append(
103 {
104 "query": result["query"],
105 "rank": 1,
106 "curie": "",
107 "iri": "",
108 "confidence": "",
109 "safe": "",
110 "evidence": "",
111 "source": "",
112 "property_type": "",
113 "match_type": "unresolved",
114 }
115 )
116 continue
117 for position, candidate in enumerate(result["candidates"], start=1):
118 rows.append(
119 {
120 "query": result["query"],
121 "rank": position,
122 "curie": candidate["curie"],
123 "iri": candidate["iri"],
124 "confidence": candidate["confidence"],
125 "safe": str(candidate["safe"]).lower(),
126 "evidence": candidate["evidence"],
127 "source": candidate["source"],
128 "property_type": candidate["property_type"],
129 "match_type": "zooma_safe" if candidate["safe"] else "zooma_weak",
130 }
131 )
132 return rows
135def write_output(results: list[dict], fmt: str, output: str | None) -> None:
136 stream = open(output, "w", encoding="utf-8", newline="") if output else sys.stdout
137 try:
138 if fmt == "json":
139 json.dump(results, stream, indent=2)
140 stream.write("\n")
141 else:
142 writer = csv.DictWriter(
143 stream, fieldnames=TSV_COLUMNS, delimiter="\t", lineterminator="\n"
144 )
145 writer.writeheader()
146 writer.writerows(to_rows(results))
147 finally:
148 if output:
149 stream.close()
152def build_parser() -> argparse.ArgumentParser:
153 parser = argparse.ArgumentParser(
154 description=(
155 "Map lab shorthand to ontology terms via EBI ZOOMA. "
156 "--ontology is required. Validate every CURIE with validate_terms.py "
157 "before writing it down."
158 ),
159 formatter_class=argparse.RawDescriptionHelpFormatter,
160 )
161 parser.add_argument("text", nargs="*", help="strings to map")
162 parser.add_argument(
163 "--input",
164 help="file with one string per line ('-' for stdin); # lines are comments",
165 )
166 parser.add_argument(
167 "--ontology",
168 required=True,
169 help="OLS ontology ids to filter on, comma separated (e.g. uberon,cl)",
170 )
171 parser.add_argument(
172 "--property-type",
173 help='ZOOMA property type, e.g. "organism part" or "cell type"',
174 )
175 parser.add_argument(
176 "--top", type=int, default=5, help="candidates to report per query (default 5)"
177 )
178 parser.add_argument(
179 "--high-confidence-only", "--exact-only", dest="exact_only",
180 action="store_true",
181 help="only HIGH/GOOD candidates (legacy --exact-only alias; not lexical exactness or validation)",
182 )
183 parser.add_argument(
184 "--format", choices=("tsv", "json"), default="tsv", help="output format"
185 )
186 parser.add_argument("-o", "--output", help="write here instead of stdout")
187 return parser
190def main(argv: list[str] | None = None) -> int:
191 args = build_parser().parse_args(argv)
192 if args.top < 1:
193 print("--top must be positive.", file=sys.stderr)
194 return 2
195 queries = read_inputs(args)
196 if not queries:
197 print("No input strings given. See --help.", file=sys.stderr)
198 return 2
200 ontologies = [item.strip() for item in args.ontology.split(",") if item.strip()]
201 if not ontologies:
202 print("--ontology needs at least one OLS ontology id.", file=sys.stderr)
203 return 2
205 results = []
206 for query in queries:
207 try:
208 results.append(
209 map_one(
210 query,
211 ontologies=ontologies,
212 property_type=args.property_type,
213 top=args.top,
214 safe_only=args.exact_only,
215 )
216 )
217 except ZoomaError as exc:
218 print(f"ZOOMA lookup failed for {query!r}: {exc}", file=sys.stderr)
219 return 2
221 write_output(results, args.format, args.output)
222 unresolved = [r["query"] for r in results if not r["candidates"]]
223 if unresolved:
224 print(
225 f"{len(unresolved)}/{len(results)} unresolved: "
226 + ", ".join(repr(q) for q in unresolved[:10]),
227 file=sys.stderr,
228 )
229 return 0
232if __name__ == "__main__":
233 raise SystemExit(main())