Skills › Scientific Agent Skills › ontology-term-resolution › validate_terms.py
validate_terms.py
324 lines · skills/ontology-term-resolution/scripts/validate_terms.py at 5f2bc03
1#!/usr/bin/env python3
2# /// script
3# requires-python = ">=3.11"
4# dependencies = []
5# ///
6"""Validate ontology CURIEs against EBI OLS4 before they leave the machine.
8This is the gate that catches invented IDs. For each term it reports whether the
9ID exists at all, whether it has been obsoleted (and what replaced it), whether a
10claimed label actually belongs to it, and optionally whether it sits in the
11branch and ontology the metadata field requires.
13Statuses: ``ok`` and ``matched_synonym`` pass, ``not_found``, ``obsolete``,
14``label_mismatch``, ``wrong_branch``, ``wrong_ontology`` fail, and
15``not_a_class`` warns. Exit code is 1 if anything failed, 0 otherwise, 2 on
16usage or network trouble. ``--strict`` promotes warnings to failures.
18Examples:
19 # spot-check three IDs
20 uv run validate_terms.py UBERON:0002107 CL:0000182 UBERON:9999999
22 # verify id/label pairs a pipeline wrote, as a CI gate
23 uv run validate_terms.py --input metadata.tsv --strict
25 # a tissue column must hold anatomical entities from UBERON
26 uv run validate_terms.py --input tissue_ids.tsv \\
27 --branch UBERON:0000465 --expect-ontology uberon
28"""
29from __future__ import annotations
31import argparse
32import csv
33import json
34import sys
35from pathlib import Path
37sys.path.insert(0, str(Path(__file__).resolve().parent))
39from ols_client import ( # noqa: E402
40 OlsError,
41 ancestor_curies,
42 curie_to_ontology_id,
43 is_curie,
44 iri_to_curie,
45 normalize_label,
46 synonym_scope,
47 term_detail,
48)
50FAIL_STATUSES = {
51 "not_found",
52 "obsolete",
53 "label_mismatch",
54 "wrong_branch",
55 "wrong_ontology",
56 "malformed_curie",
57}
58WARN_STATUSES = {"not_a_class", "matched_synonym", "imported_only"}
60TSV_COLUMNS = ("id", "status", "actual_label", "ontology", "replacement", "detail")
62ID_HEADERS = {"id", "curie", "term_id", "ontology_term_id", "obo_id"}
63LABEL_HEADERS = {"label", "term_label", "name", "ontology_term_label"}
66def read_pairs(args: argparse.Namespace) -> list[tuple[str, str | None]]:
67 """Collect (curie, expected_label) pairs from positional args or a file."""
68 pairs: list[tuple[str, str | None]] = [(value, None) for value in args.term]
69 if not args.input:
70 return pairs
72 raw = (
73 sys.stdin.read()
74 if args.input == "-"
75 else Path(args.input).read_text(encoding="utf-8")
76 )
77 lines = [line for line in raw.splitlines() if line.strip()]
78 if not lines:
79 return pairs
81 delimiter = "\t" if "\t" in lines[0] else ","
82 rows = list(csv.reader(lines, delimiter=delimiter))
83 header = [cell.strip().lower() for cell in rows[0]]
84 id_index, label_index = 0, 1
86 if any(cell in ID_HEADERS for cell in header):
87 id_index = next(i for i, cell in enumerate(header) if cell in ID_HEADERS)
88 label_index = next(
89 (i for i, cell in enumerate(header) if cell in LABEL_HEADERS), None
90 )
91 rows = rows[1:]
93 for row in rows:
94 if not row or row[0].lstrip().startswith("#"):
95 continue
96 if len(row) <= id_index or not row[id_index].strip():
97 raise ValueError("input row has a missing or empty identifier column")
98 curie = row[id_index].strip()
99 if curie.startswith("#"):
100 continue
101 label = None
102 if label_index is not None and len(row) > label_index:
103 label = row[label_index].strip() or None
104 pairs.append((curie, label))
105 return pairs
108def check_term(
109 curie: str,
110 expected_label: str | None,
111 *,
112 branch: str | None,
113 expect_ontologies: set[str] | None,
114 branch_relation: str = "hierarchical",
115) -> dict:
116 """Validate one CURIE and return a result record."""
117 result = {
118 "id": curie,
119 "status": "ok",
120 "actual_label": "",
121 "ontology": "",
122 "replacement": "",
123 "detail": "",
124 }
126 if not is_curie(curie):
127 result["status"] = "malformed_curie"
128 result["detail"] = "not of the form PREFIX:local"
129 return result
131 term = term_detail(curie)
132 if term is None:
133 result["status"] = "not_found"
134 result["detail"] = "no such term in the ontology this prefix names"
135 return result
137 result["actual_label"] = term.get("label") or ""
138 result["ontology"] = term.get("ontology_name") or ""
140 if term.get("is_obsolete"):
141 replacements = term.get("term_replaced_by") or []
142 if isinstance(replacements, str):
143 replacements = [replacements]
144 replacement = "; ".join(iri_to_curie(iri) or iri for iri in replacements)
145 result["status"] = "obsolete"
146 result["replacement"] = replacement
147 result["detail"] = (
148 f"obsolete; replaced by {replacement}"
149 if replacement
150 else "obsolete with no stated replacement"
151 )
152 return result
154 namespace_ontology = curie_to_ontology_id(curie)
155 if expect_ontologies and namespace_ontology not in expect_ontologies:
156 result["status"] = "wrong_ontology"
157 result["detail"] = (
158 f"identifier namespace maps to {namespace_ontology!r}, expected one of "
159 + ",".join(sorted(expect_ontologies))
160 )
161 return result
163 warnings: list[tuple[str, str]] = []
165 if expected_label is not None:
166 wanted = normalize_label(expected_label)
167 if normalize_label(result["actual_label"]) != wanted:
168 scope = synonym_scope(expected_label, term)
169 if scope:
170 warnings.append(
171 (
172 "matched_synonym",
173 f"{expected_label!r} is a {scope} synonym; primary label is "
174 f"{result['actual_label']!r}",
175 )
176 )
177 else:
178 result["status"] = "label_mismatch"
179 result["detail"] = (
180 f"claimed {expected_label!r}, actual {result['actual_label']!r}"
181 )
182 return result
184 if branch:
185 ancestors = ancestor_curies(curie, relation=branch_relation) if branch != curie else set()
186 if branch not in ancestors and branch != curie:
187 result["status"] = "wrong_branch"
188 result["detail"] = f"not a descendant of {branch}"
189 return result
191 # Importers can hold stale copies; lack of a defining copy requires review,
192 # but does not prove deletion in the source ontology.
193 home = term.get("_home_ontology")
194 if not term.get("is_defining_ontology"):
195 warnings.append(
196 (
197 "imported_only",
198 f"no defining copy verified for {home}; found a copy in "
199 f"{result['ontology']!r}",
200 )
201 )
203 if term.get("type") not in (None, "class"):
204 warnings.append(
205 ("not_a_class", f"term type is {term.get('type')!r}, not a class")
206 )
208 if warnings:
209 order = {"imported_only": 0, "not_a_class": 1, "matched_synonym": 2}
210 warnings.sort(key=lambda item: order[item[0]])
211 result["status"] = warnings[0][0]
212 result["detail"] = "; ".join(detail for _, detail in warnings)
214 return result
217def write_output(results: list[dict], fmt: str, output: str | None) -> None:
218 """Emit TSV or JSON to a path or stdout."""
219 stream = open(output, "w", encoding="utf-8", newline="") if output else sys.stdout
220 try:
221 if fmt == "json":
222 json.dump(results, stream, indent=2)
223 stream.write("\n")
224 else:
225 writer = csv.DictWriter(
226 stream, fieldnames=TSV_COLUMNS, delimiter="\t", lineterminator="\n"
227 )
228 writer.writeheader()
229 writer.writerows(results)
230 finally:
231 if output:
232 stream.close()
235def build_parser() -> argparse.ArgumentParser:
236 parser = argparse.ArgumentParser(
237 description="Validate ontology CURIEs against EBI OLS4.",
238 formatter_class=argparse.RawDescriptionHelpFormatter,
239 )
240 parser.add_argument("term", nargs="*", help="CURIEs to validate")
241 parser.add_argument(
242 "--input",
243 help="TSV/CSV with an id column and an optional label column ('-' for stdin)",
244 )
245 parser.add_argument(
246 "--branch", help="require every term to be a descendant of this CURIE"
247 )
248 parser.add_argument(
249 "--branch-relation", choices=("hierarchical", "is-a"), default="hierarchical",
250 help="hierarchical includes part-of/develops-from; is-a checks subclass ancestry only",
251 )
252 parser.add_argument(
253 "--expect-ontology",
254 help="comma-separated OLS ontology ids every term must come from",
255 )
256 parser.add_argument(
257 "--strict", action="store_true", help="treat warnings as failures"
258 )
259 parser.add_argument(
260 "--format", choices=("tsv", "json"), default="tsv", help="output format"
261 )
262 parser.add_argument("-o", "--output", help="write here instead of stdout")
263 return parser
266def main(argv: list[str] | None = None) -> int:
267 args = build_parser().parse_args(argv)
268 try:
269 pairs = read_pairs(args)
270 except (ValueError, OSError) as exc:
271 print(f"Invalid input: {exc}", file=sys.stderr)
272 return 2
273 if not pairs:
274 print("No terms given. See --help.", file=sys.stderr)
275 return 2
277 if args.branch and not is_curie(args.branch):
278 print(f"--branch expects a CURIE, got {args.branch!r}", file=sys.stderr)
279 return 2
280 if args.branch:
281 try:
282 root = term_detail(args.branch)
283 except OlsError as exc:
284 print(f"Could not resolve --branch: {exc}", file=sys.stderr)
285 return 2
286 if not root or root.get("is_obsolete"):
287 print("--branch must resolve to a current OLS term.", file=sys.stderr)
288 return 2
290 expect_ontologies = (
291 {value.strip().lower() for value in args.expect_ontology.split(",") if value.strip()}
292 if args.expect_ontology
293 else None
294 )
296 results = []
297 for curie, label in pairs:
298 try:
299 results.append(
300 check_term(
301 curie,
302 label,
303 branch=args.branch,
304 expect_ontologies=expect_ontologies,
305 branch_relation=args.branch_relation,
306 )
307 )
308 except OlsError as exc:
309 print(f"OLS lookup failed for {curie}: {exc}", file=sys.stderr)
310 return 2
312 write_output(results, args.format, args.output)
314 failed = [r for r in results if r["status"] in FAIL_STATUSES]
315 warned = [r for r in results if r["status"] in WARN_STATUSES]
316 summary = f"{len(results)} checked, {len(failed)} failed, {len(warned)} warned"
317 print(summary, file=sys.stderr)
318 if failed or (args.strict and warned):
319 return 1
320 return 0
323if __name__ == "__main__":
324 raise SystemExit(main())