Skills › Scientific Agent Skills › ontology-term-resolution › lookup_prefix.py
lookup_prefix.py
210 lines · skills/ontology-term-resolution/scripts/lookup_prefix.py at 5f2bc03
1#!/usr/bin/env python3
2# /// script
3# requires-python = ">=3.11"
4# dependencies = []
5# ///
6"""Look up ontology prefixes and check CURIEs against Bioregistry.
8Use this when the *shape* of an identifier is in doubt — ``HPO`` vs ``HP``,
9a local id that does not match the recorded pattern, or which landing page
10to open. It does not say whether the term exists; run ``validate_terms.py``
11for that.
13Examples:
14 uv run lookup_prefix.py HP HPO HP:0001250 HPO:0001250
15 uv run lookup_prefix.py --input prefixes.txt --format tsv
16"""
17from __future__ import annotations
19import argparse
20import csv
21import json
22import sys
23from pathlib import Path
25sys.path.insert(0, str(Path(__file__).resolve().parent))
27from id_client import ( # noqa: E402
28 IdError,
29 NotFoundError,
30 classify_prefix_query,
31 compact_id_from_identifiers_url,
32 get_reference,
33 get_resource,
34 is_prefix,
35 landing_page_urls,
36 resolve_identifiers,
37 split_query,
38)
40TSV_COLUMNS = (
41 "query",
42 "status",
43 "preferred_prefix",
44 "canonical_curie",
45 "pattern",
46 "example",
47 "name",
48 "default_iri",
49 "identifiers_org",
50 "ontobee",
51 "ols_id",
52 "detail",
53)
55FAIL_STATUSES = {"unknown_prefix", "invalid_local", "malformed"}
58def read_inputs(args: argparse.Namespace) -> list[str]:
59 """Collect prefix/CURIE strings from positional args, a file, or stdin."""
60 values: list[str] = list(args.value)
61 if args.input:
62 raw = (
63 sys.stdin.read()
64 if args.input == "-"
65 else Path(args.input).read_text(encoding="utf-8")
66 )
67 for line in raw.splitlines():
68 line = line.strip()
69 if line and not line.startswith("#"):
70 values.append(line)
71 if not values and not sys.stdin.isatty():
72 for line in sys.stdin.read().splitlines():
73 line = line.strip()
74 if line and not line.startswith("#"):
75 values.append(line)
76 seen: set[str] = set()
77 unique = []
78 for value in values:
79 if value not in seen:
80 seen.add(value)
81 unique.append(value)
82 return unique
85def lookup_one(value: str) -> dict:
86 """Resolve one prefix or CURIE through Bioregistry, then Identifiers.org."""
87 prefix, local = split_query(value)
88 if not is_prefix(prefix) and local is None:
89 return {
90 "query": value,
91 "status": "malformed",
92 "preferred_prefix": "",
93 "canonical_curie": "",
94 "pattern": "",
95 "example": "",
96 "name": "",
97 "default_iri": "",
98 "identifiers_org": "",
99 "ontobee": "",
100 "ols_id": "",
101 "detail": "not a prefix or PREFIX:local CURIE",
102 }
104 resource = None
105 reference = None
106 reference_detail = None
107 try:
108 resource = get_resource(prefix)
109 except NotFoundError as exc:
110 return classify_prefix_query(value, None, local=local, reference_detail=exc.detail)
112 if local is not None:
113 try:
114 reference = get_reference(value)
115 except NotFoundError as exc:
116 reference_detail = exc.detail
118 result = classify_prefix_query(
119 value, resource, local=local, reference_detail=reference_detail
120 )
121 if result["status"] not in {"ok", "synonym_prefix"}:
122 return result
124 identifiers, ontobee = landing_page_urls(
125 resource, reference, result["default_iri"] or None
126 )
127 result["identifiers_org"] = identifiers
128 result["ontobee"] = ontobee
130 compact = compact_id_from_identifiers_url(result["identifiers_org"])
131 if not compact:
132 return result
134 try:
135 payload = resolve_identifiers(compact)
136 except IdError:
137 extra = f"Identifiers.org resolver unavailable; landing page for {compact!r} unverified"
138 result["detail"] = f"{result['detail']}; {extra}" if result["detail"] else extra
139 return result
140 if payload.get("errorMessage"):
141 result["identifiers_org"] = ""
142 extra = f"Identifiers.org rejected {compact!r}: {payload['errorMessage']}"
143 result["detail"] = f"{result['detail']}; {extra}" if result["detail"] else extra
144 return result
147def write_output(results: list[dict], fmt: str, output: str | None) -> None:
148 stream = open(output, "w", encoding="utf-8", newline="") if output else sys.stdout
149 try:
150 if fmt == "json":
151 json.dump(results, stream, indent=2)
152 stream.write("\n")
153 else:
154 writer = csv.DictWriter(
155 stream, fieldnames=TSV_COLUMNS, delimiter="\t", lineterminator="\n"
156 )
157 writer.writeheader()
158 writer.writerows(results)
159 finally:
160 if output:
161 stream.close()
164def build_parser() -> argparse.ArgumentParser:
165 parser = argparse.ArgumentParser(
166 description=(
167 "Look up ontology prefixes and check CURIEs against Bioregistry. "
168 "Does not say whether the term exists — use validate_terms.py for that."
169 ),
170 formatter_class=argparse.RawDescriptionHelpFormatter,
171 )
172 parser.add_argument("value", nargs="*", help="prefixes or CURIEs")
173 parser.add_argument(
174 "--input",
175 help="file with one prefix or CURIE per line ('-' for stdin); # lines are comments",
176 )
177 parser.add_argument(
178 "--format", choices=("tsv", "json"), default="tsv", help="output format"
179 )
180 parser.add_argument("-o", "--output", help="write here instead of stdout")
181 return parser
184def main(argv: list[str] | None = None) -> int:
185 args = build_parser().parse_args(argv)
186 values = read_inputs(args)
187 if not values:
188 print("No prefixes or CURIEs given. See --help.", file=sys.stderr)
189 return 2
191 results = []
192 for value in values:
193 try:
194 results.append(lookup_one(value))
195 except IdError as exc:
196 print(f"Prefix lookup failed for {value!r}: {exc}", file=sys.stderr)
197 return 2
199 write_output(results, args.format, args.output)
200 failed = [r for r in results if r["status"] in FAIL_STATUSES]
201 synonyms = [r for r in results if r["status"] == "synonym_prefix"]
202 print(
203 f"{len(results)} checked, {len(failed)} failed, {len(synonyms)} synonym prefixes",
204 file=sys.stderr,
205 )
206 return 1 if failed else 0
209if __name__ == "__main__":
210 raise SystemExit(main())