File size: 2,558 Bytes
1c6cb4b
7b3f817
cdc75a8
1c6cb4b
cdc75a8
 
7b3f817
1087a21
a9daa10
cdc75a8
1087a21
cdc75a8
1087a21
cdc75a8
 
 
7b3f817
cdc75a8
 
 
a9daa10
cdc75a8
 
a9daa10
cdc75a8
a9daa10
cdc75a8
 
 
 
1c6cb4b
7b3f817
cdc75a8
1087a21
cdc75a8
1087a21
cdc75a8
 
7b3f817
cdc75a8
 
 
 
1c6cb4b
cdc75a8
 
 
 
 
 
 
 
 
a9daa10
cdc75a8
7b3f817
cdc75a8
 
a9daa10
cdc75a8
a9daa10
cdc75a8
 
 
 
 
 
 
 
7b3f817
 
cdc75a8
7b3f817
cdc75a8
7b3f817
cdc75a8
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
90
91
92
93
94
# genesis/api_clients/umls_api.py
import os
import requests

UMLS_API_KEY = os.getenv("UMLS_API_KEY")
UMLS_AUTH_ENDPOINT = "https://utslogin.nlm.nih.gov/cas/v1/api-key"
UMLS_BASE_URL = "https://uts-ws.nlm.nih.gov/rest"


def get_tgt():
    """
    Get Ticket-Granting Ticket (TGT) for UMLS API authentication.
    """
    params = {"apikey": UMLS_API_KEY}
    r = requests.post(UMLS_AUTH_ENDPOINT, data=params)
    r.raise_for_status()

    # Extract TGT from HTML response
    tgt_url = r.text.split('action="')[1].split('"')[0]
    return tgt_url


def get_service_ticket(tgt):
    """
    Get Service Ticket (ST) from the TGT for making API requests.
    """
    params = {"service": "http://umlsks.nlm.nih.gov"}
    r = requests.post(tgt, data=params)
    r.raise_for_status()
    return r.text


def search_umls(term: str, max_results: int = 10):
    """
    Search UMLS for a term and return matching concepts.
    """
    tgt = get_tgt()
    st = get_service_ticket(tgt)

    url = f"{UMLS_BASE_URL}/search/current"
    params = {"string": term, "ticket": st, "pageSize": max_results}
    r = requests.get(url, params=params)
    r.raise_for_status()

    data = r.json()
    results = []

    for result in data.get("result", {}).get("results", []):
        results.append({
            "ui": result.get("ui", ""),
            "name": result.get("name", ""),
            "rootSource": result.get("rootSource", "")
        })

    return results


def get_concept_details(cui: str):
    """
    Retrieve detailed information for a concept by CUI (Concept Unique Identifier).
    """
    tgt = get_tgt()
    st = get_service_ticket(tgt)

    url = f"{UMLS_BASE_URL}/content/current/CUI/{cui}"
    params = {"ticket": st}
    r = requests.get(url, params=params)
    r.raise_for_status()
    return r.json()


def map_related_concepts(term: str):
    """
    Search UMLS and map related concepts with synonyms and semantic types.
    """
    concepts = search_umls(term, max_results=5)
    mapped = []

    for c in concepts:
        cui = c["ui"]
        if cui and cui != "NONE":
            details = get_concept_details(cui)
            entity = details.get("result", {})

            mapped.append({
                "cui": cui,
                "name": entity.get("name", ""),
                "synonyms": [syn.get("name", "") for syn in entity.get("atoms", [])],
                "semantic_types": [st.get("name", "") for st in entity.get("semanticTypes", [])],
                "rootSource": c.get("rootSource", "")
            })

    return mapped