utilities.py

#!/usr/bin/python

"""
Utility functions.

"""


import time
import logging
import requests
import json
from config import settings
from Authentication import Authentication


# API-kEY FOR UMLS REST TICKET SERVICES
umls_api = settings['apis']['umls']
# UMLS REST SERVICES INITIALIZATION OF CLIENT AND TICKET
# GRANTING SERVICE TO BE USED IN ALL CASES
AuthClient = Authentication(umls_api)
tgt = AuthClient.gettgt()

# To supress some kind of warning?!
logging.getLogger("requests").setLevel(logging.WARNING)
logging.getLogger("urllib3").setLevel(logging.WARNING)



# logging.basicConfig(
#     format="%(asctime)s [%(threadName)-12.12s] [%(levelname)-5.5s]  %(message)s",
#     handlers=[
#         #logging.FileHandler("%s" % settings['log_path']),
#         logging.StreamHandler()
#     ])
# logging.info('lala')

# # create logger
# logger = logging.getLogger()
# formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
# ch = logging.StreamHandler()
# ch.setFormatter(formatter)
# logger.addHandler(ch)

# fh = logging.FileHandler(settings['log_path'])
# fh.setFormatter(formatter)
# logger.addHandler(fh)


def get_umls_ticket2(tgt=tgt, AuthClient=AuthClient, apikey=umls_api):
    """
    Get a single use ticket for the UMLS REST services.
    It is supposed that an Author Client and a Ticket
    Granting Service have already been set-up in case
    the apikey = None. If an api-key is given, create the
    above needed instances and generate a new ticket.
    Input:
        - apikey: str,
        UMLS REST services api-key. Default is None and
        the already establised service is used
    Output:
        - string of the generated ticket
    """

    # Get ticket from the already establised service
    if not(tgt) and not(AuthClient):
        AuthClient = Authentication(umls_api)
        tgt = AuthClient.gettgt()
    return AuthClient.getst(tgt)


def get_umls_ticket(apikey=None, AuthClient=AuthClient, tgt=tgt):
    """
    Get a single use ticket for the UMLS REST services.
    It is supposed that an Author Client and a Ticket
    Granting Service have already been set-up in case
    the apikey = None. If an api-key is given, create the
    above needed instances and generate a new ticket.
    Input:
        - apikey: str,
        UMLS REST services api-key. Default is None and
        the already establised service is used
    Output:
        - string of the generated ticket
    """

    # Get ticket from the already establised service
    if apikey is None:
        return AuthClient.getst(tgt)
    else:
        # Establish new Client and Ticket granting service
        AuthClient = Authentication(apikey)
        tgt = AuthClient.gettgt()
        return AuthClient.getst(tgt)


def time_log(phrase, time_start=None):
    """
    A time_logger function so as to print info with time since elapsed if wanted,
    alongside with the current logging config.
    """

    # If we want to also print time_elapsed
    if time_start:
        logging.info('%s in : %.2f seconds.' % (phrase, time.time() - time_start))
    else:
        logging.info('%s' % (phrase))
    return 1


def get_concept_from_source(source_id, source, apikey=tgt):
    """
    Function that maps an entity from another source to UMLS concepts.
    Input:
        - source_id: str,
        string of the unique id from the source knowledge base
        - source: str,
        string code-name of the knowledge base, as used by the
        UMLS REST services (e.g. drugbank -> DRUGBANK, MESH->MSH)
        - apikey: str,
        UMLS REST services api-key. Default is None and
        the already establised service is used. Check get_umls_ticket
        function for details
    Output:
        - concepts: list,
        list of dictionaries representing concepts mapped to the
        source_id entity. Each dictionary has keys "label", "cuid", "sem_types"
        Check get_concept_from_cui for more details
    """

    ticket = get_umls_ticket2()
    #ticket = get_umls_ticket(apikey)
    url = "https://uts-ws.nlm.nih.gov/rest/search/current"
    passed = False
    times = 0
    while not(passed):
        params = {'string': source_id, 'sabs': source, 'searchType': 'exact',
              'inputType': 'sourceUi', 'ticket': ticket}
        r = requests.get(url, params=params)
        r.encoding = 'utf-8'
        concepts = []
        if r.ok:
            items = json.loads(r.text)
            jsonData = items["result"]
            # Get cuis related to source_id
            cuis = [res['ui']for res in jsonData['results']]
            # Get concepts from cuis
            concepts = [get_concept_from_cui(cui) for cui in cuis if cui != 'NONE']
            passed = True
        else:
            time_log(r.url)
            time_log('Error getting concept from: Source %s   | ID: %s' % (source, source_id))
            time_log('~'*25 + ' GETTING NEW TICKET SERVICE' + '~'*24)
            ticket = get_umls_ticket2(None, None, umls_api)
            times += 1
            if times >= 2:
                passed = True
                time_log('Error getting concept from: Source %s   | ID: %s' % (source, source_id))
                time_log('~'*25 + ' EXITING AFTER TRYING TWICE WITH NEW TICKET  ' + '~'*25)
                #exit(1)
                raise Exception
    return concepts


def get_concept_from_cui(cui, apikey=None):
    """
    Function that fetches a concept's attributes from the corresponding cui.
    Input:
        - cui: str,
        string of cui that will be looked up
        - apikey: str,
        UMLS REST services api-key. Default is None and
        the already establised service is used. Check get_umls_ticket
        function for details
    Output:
        - res: dictionary,
        dictionary with the concepts attributes as fetched. Specifically,
        "label", "cuid"(cui) and "sem_types"(comma delimited string of
        the semantic types us returned)
    """

    ticket = get_umls_ticket2()
    #ticket = get_umls_ticket(apikey)
    url = "https://uts-ws.nlm.nih.gov/rest/content/current/CUI/" + cui
    passed = False
    times = 0
    while not(passed):
        try:
            r = requests.get(url, params={'ticket': ticket}, timeout=120)
            passed = True
        except requests.exceptions.Timeout:
            time_log('~'*25 + ' TIMEOUT ERROR 120 SECONDS'+'~'*25)
            time_log('~'*25 + ' GETTING NEW TICKET SERVICE' + '~'*24)
            ticket = get_umls_ticket2(None, None, umls_api)
            times += 1
            if times >= 2:
                passed = True
                time_log('Error getting concept from: CUI %s' % cui)
                time_log('~'*25 + ' EXITING AFTER TRYING TWICE WITH NEW TICKET  ' + '~'*25)
                raise Exception
                #exit(1)
    r.encoding = 'utf-8'
    res = {}
    if r.ok:
        items = json.loads(r.text)
        jsonData = items["result"]
        res = {'label': jsonData['name'], 'cuid': cui}
        sem_types = []
        # For each semantic type of the entity
        for stys in jsonData["semanticTypes"]:
            # Keep only the TUI code from the uri e.g.
            # https://uts-ws.nlm.nih.gov/rest/semantic-network/current/TUI/T116
            code_tui = stys['uri'].split('/')[-1]
            # Fetch the abbreviation of this TUI code
            sem_types.append(get_sem_type_abbr(code_tui))
        # Comma separated string
        sem_types = ",".join(sem_types)
        res['sem_types'] = sem_types
    else:
        time_log(r.url)
        time_log('Error getting concept from cui : %s' % cui)
        raise ValueError
    return res


def get_sem_type_abbr(code_tui, apikey=None):
    """
    Function that fetches a semantic-type's abbreviation.
    Input:
        - code_tui: str,
        string of TUI code that will be looked up
        - apikey: str,
        UMLS REST services api-key. Default is None and
        the already establised service is used. Check get_umls_ticket
        function for details
    Output:
        string, abbreviation of the code (e.g. "gngm")
    """
    ticket = get_umls_ticket2()
    #ticket = get_umls_ticket(apikey)
    url = "https://uts-ws.nlm.nih.gov/rest/semantic-network/current/TUI/" + code_tui
    passed = False
    times = 0
    while not(passed):
        try:
            r = requests.get(url, params={'ticket': ticket}, timeout=120)
            passed = True
        except requests.exceptions.Timeout:
            time_log('~'*25 + ' TIMEOUT ERROR 120 SECONDS'+'~'*25)
            time_log('~'*25 + ' GETTING NEW TICKET SERVICE' + '~'*24)
            ticket = get_umls_ticket2(None, None, umls_api)
        times += 1
        if times >= 2:
            passed = True
            time_log('Error getting semantic type abbreviation: %s' % code_tui)
            time_log('~'*25 + ' EXITING AFTER TRYING TWICE WITH NEW TICKET  ' + '~'*25)
            raise Exception
            #exit(1)
    r.encoding = 'utf-8'
    res = ' '
    if r.ok:
        items = json.loads(r.text)
        jsonData = items["result"]
        res = jsonData['abbreviation']
    else:
        time_log(r.url)
        time_log('Error getting sem-type from TUI : %s' % code_tui)
        raise ValueError
    return res