Skills › Scientific Agent Skills › ontology-term-resolution › zooma_client.py
zooma_client.py
144 lines · skills/ontology-term-resolution/scripts/zooma_client.py at 5f2bc03
1#!/usr/bin/env python3
2# /// script
3# requires-python = ">=3.11"
4# dependencies = []
5# ///
6"""EBI ZOOMA annotate client.
8ZOOMA maps free text to ontology IRIs using curated annotation history. It is
9a fallback when OLS lexical search fails on lab shorthand (``PBMC``, ``WT``),
10not a replacement for OLS.
12Unfiltered annotate is unusable — ``propertyValue=liver`` returns FOODON,
13XAO, BTO, and UBERON as equally HIGH hits. Always pass an ontology filter.
15Standard library only. Reuses ``iri_to_curie`` from ``ols_client`` so IRI
16shapes stay consistent with the rest of the skill.
17"""
18from __future__ import annotations
20import json
21import time
22import urllib.error
23import urllib.parse
24import urllib.request
25from typing import Any
27from ols_client import iri_to_curie
29ZOOMA_ANNOTATE = "https://www.ebi.ac.uk/spot/zooma/v2/api/services/annotate"
30USER_AGENT = "scientific-agent-skills-ontology-term-resolution/1.4"
31TIMEOUT = 60
32MAX_ATTEMPTS = 3
33RETRY_STATUS = {429, 500, 502, 503, 504}
35# Legacy output field ``safe`` means this confidence bucket only. Neither HIGH
36# nor GOOD proves semantic equivalence, curation, currency or correct context.
37SAFE_CONFIDENCE = {"HIGH", "GOOD"}
40class ZoomaError(RuntimeError):
41 """A request to ZOOMA failed in a way the caller cannot paper over."""
44def ontology_filter(ontologies: list[str]) -> str:
45 """Build the ``filter`` query value ZOOMA requires.
47 Empty required sources permit all curated sources. Ontology ids are
48 lowercase OLS ids (``uberon``, ``cl``); defining_only excludes imports.
49 """
50 ids = ",".join(item.strip().lower() for item in ontologies if item.strip())
51 if not ids:
52 raise ZoomaError("ZOOMA annotate requires at least one ontology id")
53 if "none" in ids.split(","):
54 raise ZoomaError("'none' removes the server's ontology restriction; use a real ontology id")
55 if any(not item or not all(c.isalnum() or c in "_.-" for c in item) for item in ids.split(",")):
56 raise ZoomaError("invalid OLS ontology id")
57 return f"ontologies:[{ids}],defining_only:[true]"
60def annotate(
61 text: str,
62 *,
63 ontologies: list[str],
64 property_type: str | None = None,
65) -> list[dict]:
66 """Call ``/annotate`` and return the raw hit list.
68 ``ontologies`` is required. Calling this without a filter is how you get
69 ``FOODON:03309772`` for ``liver``.
70 """
71 params: dict[str, Any] = {
72 "propertyValue": text,
73 "filter": ontology_filter(ontologies),
74 }
75 if property_type:
76 params["propertyType"] = property_type
77 url = f"{ZOOMA_ANNOTATE}?{urllib.parse.urlencode(params)}"
78 request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT, "Accept": "application/json"})
80 last: Exception | None = None
81 for attempt in range(MAX_ATTEMPTS):
82 try:
83 with urllib.request.urlopen(request, timeout=TIMEOUT) as response:
84 payload = json.load(response)
85 if not isinstance(payload, list):
86 raise ZoomaError(f"ZOOMA returned a non-list body for {text!r}")
87 return payload
88 except urllib.error.HTTPError as exc:
89 last = exc
90 if exc.code not in RETRY_STATUS:
91 raise ZoomaError(f"ZOOMA HTTP {exc.code} for {text!r}") from exc
92 except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc:
93 last = exc
94 if attempt < MAX_ATTEMPTS - 1:
95 time.sleep(1.5 * (attempt + 1))
96 raise ZoomaError(f"ZOOMA request failed after {MAX_ATTEMPTS} attempts: {url} ({last})")
99def flatten_hit(hit: dict) -> list[dict]:
100 """Turn one ZOOMA annotation into one row per semantic tag."""
101 confidence = (hit.get("confidence") or "").upper()
102 prop = hit.get("annotatedProperty") or {}
103 chain = []
104 origin = hit
105 while isinstance(origin, dict):
106 if origin.get("provenance"):
107 chain.append(origin["provenance"])
108 origin = origin.get("derivedFrom")
109 # The v2 compatibility wrapper always says INFERRED_FROM_CURATED, even for
110 # lexical or embedding hits. Report the underlying source's evidence.
111 provenance = chain[-1] if chain else {}
112 source = provenance.get("source") or {}
113 rows = []
114 tags = hit.get("semanticTags") or []
115 if not tags:
116 return [
117 {
118 "iri": "",
119 "curie": "",
120 "confidence": confidence or "UNKNOWN",
121 "safe": False,
122 "evidence": provenance.get("evidence") or "",
123 "source": source.get("name") or "",
124 "property_type": prop.get("propertyType") or "",
125 "property_value": prop.get("propertyValue") or "",
126 "provenance_chain": chain,
127 }
128 ]
129 for iri in tags:
130 iri = str(iri)
131 rows.append(
132 {
133 "iri": iri,
134 "curie": iri_to_curie(iri) or "",
135 "confidence": confidence or "UNKNOWN",
136 "safe": confidence in SAFE_CONFIDENCE,
137 "evidence": provenance.get("evidence") or "",
138 "source": source.get("name") or "",
139 "property_type": prop.get("propertyType") or "",
140 "property_value": prop.get("propertyValue") or "",
141 "provenance_chain": chain,
142 }
143 )
144 return rows