In [6]:
from pyontutils.sheets import Sheet
import pandas as pd
KEY_NAME = 'sparc-terms'
SHEET_NAME = 'Minimal information model(MIS)'
class Brainstem(Sheet):
name = KEY_NAME # key name you gave the google sheet id value in secrets.yaml
sheet_name = SHEET_NAME # the actual sheet name on the google sheet
fetch_grid = True # meta data in self.grid that has detials like bolding
brainstem = Brainstem()
df = pd.DataFrame(brainstem.raw_values)
df.columns = df.iloc[0]
df.drop(df.index[0], inplace=True)
df.head()
Out[6]:
In [8]:
list(df.term)[:10]
Out[8]:
In [3]:
# Give "query" a usable parameter to query the databases
from pyontutils.core import query # OntTerm
query(term='brain')
Out[3]:
In [4]:
# similar entities will show
# default limit is 10
query(term='DN1 neuron', limit=2)
Out[4]:
In [5]:
# Faster and more accurate with curie/iri
query(curie='UBERON:0000955')
Out[5]:
In [6]:
entity = query(curie='UBERON:0000955')[0]
# Full result attribute
vars(entity)
Out[6]:
In [7]:
query?
In [8]:
from pyontutils.utils import Async, deferred
from pyontutils.core import OntTerm, ixr, query
from typing import List, Tuple
# query.setup()
def queries(kwargs_list:List[dict]) -> List[Tuple[str, dict]]:
'''Asynchronously query databases to dramatically increase runtime un users end
Examples:
>>> queries([{'term':'Brain'},])
[({'term': 'Brain'},
[OntTerm('HBA:3999', label='brain (hba)'),
OntTerm('FMA:50801', label='Brain'),
OntTerm('UBERON:0000955', label='brain'),
OntTerm('UBERON:6110636', label='adult cerebral ganglion')])]
>>> queries([{'curie':'UBERON:0000955'},])
[({'curie': 'UBERON:0000955'}, [OntTerm('UBERON:0000955', label='brain')])]
Definitions:
kwargs == common name given to dictionary input for function
tuple == a list that you cannot update.
lambda == short-hand for single line function creation (func = lambda <input>: <output>)
Args:
kwargs_list (list): A list of dictionaries that are paramaters for the query function
Returns:
List[tuple]: A list of tuples all being of (<input>, <query_result>).
'''
# create a query function wrapper to return tuple
# kwargs -> (kwargs, query_result)
# We do this in case 2+ queries return the same results & the output WILL NOT have the same input order
gin = lambda kwargs: (kwargs, query(**kwargs))
# run each query instance at the same time
results = Async(use_nest_asyncio=True)(deferred(gin)(kwargs) for kwargs in kwargs_list)
return results
queries([{'curie':'UBERON:0000955'}, {'curie':'UBERON:6110636'}])
Out[8]:
In [ ]:
from pyontutils.utils import Async, deferred
from pyontutils.core import OntTerm, ixr, query
from typing import List, Tuple
def queries(url_list:List[dict]) -> List[Tuple[str, dict]]:
def gin(url):
return requests.get(url).text
# run each query instance at the same time
results = Async(limit=5)(deferred(gin)(url) for url in url_list)
return results
list_tuples(url, html)
In [1]:
# TEST InterLex endpoints
from ilxutils.remotes import interlex_remote_test as ixrt
In [2]:
ixrt.ilx_cli.get_entity('tmp_0738390')
Out[2]:
In [2]:
ixrt.add_entity?
In [3]:
entity = dict(
label = 'offical label', # Can only one unique label per person
type = 'term', # OPTIONS: term, annotation, relationship, cde, fde, pde
definition = 'official definition',
comment = 'helpful misc',
# Optional
subThingOf = '', # WARNING ::: must have at last '', can be blank but please fill this in if you can.
synonyms = ['Encephalon', 'Cerebro'],
predicates = {} # annotations and/or relationships to add
# TODO: existing_ids will be an option later
)
result = ixrt.add_entity(**entity)
print(result)
In [4]:
ixrt.update_entity?
In [8]:
from ilxutils.remotes import interlex_remote_test as ixrt
entity = dict(
ilx_id = 'ILX:0738390',
label = 'Offical label', # Can only one unique label per person
type = 'term', # OPTIONS: term, annotation, relationship, cde, fde, pde
definition = 'official definition',
comment = 'helpful misc',
# Optional
subThingOf = '', # WARNING ::: must have at last '', can be blank but please fill this in if you can.
synonyms = ['Encephalon', 'Cerebro'],
predicates_to_add = {}, # annotations and/or relationships to add
predicates_to_delete = {}, # annotations and/or relationships to del
# TODO: existing_ids will be an option later
)
result = ixrt.update_entity(**entity)
print(result)
In [3]:
# PRODUCTION
from ilxutils.remotes import interlex_remote_production as ixr
# BE CAREFUL :)