File size: 2,840 Bytes
6e9bb07
87c5f3b
8e4dca9
78f5084
87c5f3b
6e9bb07
87c5f3b
 
8e4dca9
 
 
 
6e9bb07
87c5f3b
 
6e9bb07
8e4dca9
 
 
 
 
87c5f3b
 
 
8e4dca9
87c5f3b
 
 
 
8e4dca9
87c5f3b
6e9bb07
8e4dca9
 
 
 
 
87c5f3b
 
 
8e4dca9
87c5f3b
 
 
 
8e4dca9
87c5f3b
6e9bb07
8e4dca9
 
 
 
78f5084
 
 
8e4dca9
78f5084
8e4dca9
 
78f5084
 
 
8e4dca9
 
78f5084
8e4dca9
78f5084
 
 
 
 
 
 
 
 
 
 
 
 
 
8e4dca9
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
# genesis/ontology.py
"""
Ontology Expansion & Merging for GENESIS-AI
Uses UMLS & BioPortal to find related terms, synonyms, and hierarchies.
"""

import os
import requests
import logging
from typing import List, Dict, Union

logging.basicConfig(level=logging.INFO)

UMLS_API_KEY = os.getenv("UMLS_API_KEY")
BIOPORTAL_API_KEY = os.getenv("BIOPORTAL_API_KEY")


# -------------------------
# BioPortal Expansion
# -------------------------
def expand_with_bioportal(term: str) -> Union[List[str], Dict]:
    """Expand term using BioPortal ontology search."""
    try:
        url = f"https://data.bioontology.org/search?q={term}&apikey={BIOPORTAL_API_KEY}"
        r = requests.get(url, timeout=20)
        r.raise_for_status()
        data = r.json()
        return list({res["prefLabel"] for res in data.get("collection", []) if "prefLabel" in res})
    except Exception as e:
        logging.error(f"[Ontology] BioPortal expansion failed: {e}")
        return {"error": str(e)}


# -------------------------
# UMLS Expansion
# -------------------------
def expand_with_umls(term: str) -> Union[List[str], Dict]:
    """Expand term using UMLS API."""
    try:
        url = f"https://uts-ws.nlm.nih.gov/rest/search/current?string={term}&apiKey={UMLS_API_KEY}"
        r = requests.get(url, timeout=20)
        r.raise_for_status()
        data = r.json()
        return [res["name"] for res in data.get("result", {}).get("results", []) if "name" in res]
    except Exception as e:
        logging.error(f"[Ontology] UMLS expansion failed: {e}")
        return {"error": str(e)}


# -------------------------
# Merge Ontology Terms
# -------------------------
def merge_ontology_terms(query_term: str,
                         umls_terms: Union[List[str], Dict, None] = None,
                         bioportal_terms: Union[List[str], Dict, None] = None) -> List[str]:
    """
    Merge the original query term with expanded terms from UMLS and BioPortal.

    Args:
        query_term: Original search term.
        umls_terms: List or error dict from expand_with_umls().
        bioportal_terms: List or error dict from expand_with_bioportal().

    Returns:
        List of unique merged terms (strings).
    """
    logging.info(f"[Ontology] Merging terms for query: {query_term}")
    results = set()

    if query_term:
        results.add(query_term.strip())

    if isinstance(umls_terms, list):
        results.update(t.strip() for t in umls_terms if t and isinstance(t, str))

    if isinstance(bioportal_terms, list):
        results.update(t.strip() for t in bioportal_terms if t and isinstance(t, str))

    merged_list = list(results)
    logging.info(f"[Ontology] Merged {len(merged_list)} unique terms.")
    return merged_list


__all__ = [
    "expand_with_bioportal",
    "expand_with_umls",
    "merge_ontology_terms"
]