Source code for indra.sources.medscan.processor

from urllib.parse import unquote

import re
import os
import glob
import time
import shutil
import tempfile
import logging
from math import floor
import lxml.etree
import collections

from indra.databases import go_client, mesh_client
from indra.statements import *
from indra.databases.chebi_client import get_chebi_id_from_cas, \
from indra.databases.hgnc_client import get_hgnc_from_entrez, get_uniprot_id, \
from indra.util import read_unicode_csv
from indra.sources.reach.processor import ReachProcessor, Site

from .fix_csxml_character_encoding import fix_character_encoding

logger = logging.getLogger(__name__)

MedscanEntity = collections.namedtuple('MedscanEntity', ['name', 'urn', 'type',
                                                         'ch_start', 'ch_end'])

MedscanProperty = collections.namedtuple('MedscanProperty',
                                         ['type', 'name', 'urn'])

def _read_famplex_map():
    fname = os.path.join(os.path.dirname(os.path.abspath(__file__)),
    famplex_map = {}
    csv_rows = read_unicode_csv(fname, delimiter='\t')
    for row in csv_rows:
        source_ns = row[0]
        source_id = row[1]
        be_id = row[2]
        famplex_map[(source_ns, source_id)] = be_id
    return famplex_map

famplex_map = _read_famplex_map()

def _fix_different_refs(a1, a2, ref_key):
    if all(ref_key in a.db_refs for a in [a1, a2]) \
           and a1.db_refs[ref_key] != a2.db_refs[ref_key]: = a1.db_refs[ref_key] = a2.db_refs[ref_key]
        return True
    return False

def _is_statement_in_list(new_stmt, old_stmt_list):
    """Return True of given statement is equivalent to on in a list

    Determines whether the statement is equivalent to any statement in the
    given list of statements, with equivalency determined by Statement's
    equals method.

    new_stmt : indra.statements.Statement
        The statement to compare with
    old_stmt_list : list[indra.statements.Statement]
        The statement list whose entries we compare with statement

    in_list : bool
        True if statement is equivalent to any statements in the list
    for old_stmt in old_stmt_list:
        if old_stmt.equals(new_stmt):
            return True
        elif old_stmt.evidence_equals(new_stmt) and old_stmt.matches(new_stmt):
            # If we're comparing a complex, make sure the agents are sorted.
            if isinstance(new_stmt, Complex):
                agent_pairs = zip(old_stmt.sorted_members(),
                agent_pairs = zip(old_stmt.agent_list(), new_stmt.agent_list())

            # Compare agent-by-agent.
            for ag_old, ag_new in agent_pairs:
                s_old = set(ag_old.db_refs.items())
                s_new = set(ag_new.db_refs.items())

                # If they're equal this isn't the one we're interested in.
                if s_old == s_new:

                # If the new statement has nothing new to offer, just ignore it
                if s_old > s_new:
                    return True

                # If the new statement does have something new, add it to the
                # existing statement. And then ignore it.
                if s_new > s_old:
                    return True

                # If this is a case where different CHEBI ids were mapped to
                # the same entity, set the agent name to the CHEBI id.
                if _fix_different_refs(ag_old, ag_new, 'CHEBI'):
                    # Check to make sure the newly described statement does
                    # not match anything.
                    return _is_statement_in_list(new_stmt, old_stmt_list)

                # If this is a case, like above, but with UMLS IDs, do the same
                # thing as above. This will likely never be improved.
                if _fix_different_refs(ag_old, ag_new, 'UMLS'):
                    # Check to make sure the newly described statement does
                    # not match anything.
                    return _is_statement_in_list(new_stmt, old_stmt_list)

                logger.warning("Found an unexpected kind of duplicate. "
                               "Ignoring it.")
                return True

            # This means all the agents matched, which can happen if the
            # original issue was the ordering of agents in a Complex.
            return True

        elif old_stmt.get_hash(True, True) == new_stmt.get_hash(True, True):
            # Check to see if we can improve the annotation of the existing
            # statement.
            e_old = old_stmt.evidence[0]
            e_new = new_stmt.evidence[0]
            if e_old.annotations['last_verb'] is None:
                e_old.annotations['last_verb'] = e_new.annotations['last_verb']

            # If the evidence is "the same", modulo annotations, just ignore it
            if e_old.get_source_hash(True) == e_new.get_source_hash(True):
                return True

    return False

[docs]class ProteinSiteInfo(object): """Represent a site on a protein, extracted from a StateEffect event. Parameters ---------- site_text : str The site as a string (ex. S22) object_text : str The protein being modified, as the string that appeared in the original sentence """ def __init__(self, site_text, object_text): self.site_text = site_text self.object_text = object_text
[docs] def get_sites(self): """Parse the site-text string and return a list of sites. Returns ------- sites : list[Site] A list of position-residue pairs corresponding to the site-text """ st = self.site_text suffixes = [' residue', ' residues', ',', '/'] for suffix in suffixes: if st.endswith(suffix): st = st[:-len(suffix)] assert(not st.endswith(',')) # Strip parentheses st = st.replace('(', '') st = st.replace(')', '') st = st.replace(' or ', ' and ') # Treat end and or the same sites = [] parts = st.split(' and ') for part in parts: if part.endswith(','): part = part[:-1] if len(part.strip()) > 0: sites.extend(ReachProcessor._parse_site_text(part.strip())) return sites
# These normalized verbs are mapped to IncreaseAmount statements INCREASE_AMOUNT_VERBS = ['ExpressionControl-positive', 'MolSynthesis-positive', 'CellExpression', 'QuantitativeChange-positive', 'PromoterBinding'] # These normalized verbs are mapped to DecreaseAmount statements DECREASE_AMOUNT_VERBS = ['ExpressionControl-negative', 'MolSynthesis-negative', 'miRNAEffect-negative', 'QuantitativeChange-negative'] # These normalized verbs are mapped to Activation statements (indirect) ACTIVATION_VERBS = ['UnknownRegulation-positive', 'Regulation-positive'] # These normalized verbs are mapped to Activation statements (direct) D_ACTIVATION_VERBS = ['DirectRegulation-positive', 'DirectRegulation-positive--direct interaction'] # All activation verbs ALL_ACTIVATION_VERBS = ACTIVATION_VERBS + D_ACTIVATION_VERBS # These normalized verbs are mapped to Inhibition statements (indirect) INHIBITION_VERBS = ['UnknownRegulation-negative', 'Regulation-negative'] # These normalized verbs are mapped to Inhibition statements (direct) D_INHIBITION_VERBS = ['DirectRegulation-negative', 'DirectRegulation-negative--direct interaction'] # All inhibition verbs ALL_INHIBITION_VERBS = INHIBITION_VERBS + D_INHIBITION_VERBS PMID_PATT = re.compile('info:pmid/(\d+)')
[docs]class MedscanProcessor(object): """Processes Medscan data into INDRA statements. The special StateEffect event conveys information about the binding site of a protein modification. Sometimes this is paired with additional event information in a seperate SVO. When we encounter a StateEffect, we don't process into an INDRA statement right away, but instead store the site information and use it if we encounter a ProtModification event within the same sentence. Attributes ---------- statements : list<str> A list of extracted INDRA statements sentence_statements : list<str> A list of statements for the sentence we are currently processing. Deduplicated and added to the main statement list when we finish processing a sentence. num_entities : int The total number of subject or object entities the processor attempted to resolve num_entities_not_found : int The number of subject or object IDs which could not be resolved by looking in the list of entities or tagged phrases. last_site_info_in_sentence : SiteInfo Stored protein site info from the last StateEffect event within the sentence, allowing us to combine information from StateEffect and ProtModification events within a single sentence in a single INDRA statement. This is reset at the end of each sentence """ def __init__(self): self.statements = [] self.sentence_statements = [] self.num_entities_not_found = 0 self.num_entities = 0 self.last_site_info_in_sentence = None self.files_processed = 0 self._gen = None self._tmp_dir = None self._pmids_handled = set() self._sentences_handled = set() self.__f = None return def iter_statements(self, populate=True): if self._gen is None and not self.statements: raise InputError("No generator has been initialized. Use " "`process_directory` or `process_file` first.") if self.statements and not self._gen: for stmt in self.statements: yield stmt else: for stmt in self._gen: if populate: self.statements.append(stmt) yield stmt def process_directory(self, directory_name, lazy=False): # Process each file glob_pattern = os.path.join(directory_name, '*.csxml') files = glob.glob(glob_pattern) self._gen = self._iter_over_files(files) if not lazy: for stmt in self._gen: self.statements.append(stmt) return def _iter_over_files(self, files): # Create temporary directory into which to put the csxml files with # normalized character encodings self.__tmp_dir = tempfile.mkdtemp('indra_medscan_processor') tmp_file = os.path.join(self.__tmp_dir, 'fixed_char_encoding') num_files = float(len(files)) percent_done = 0 start_time_s = time.time()"%d files to read" % int(num_files)) for filename in files:'Processing %s' % filename) fix_character_encoding(filename, tmp_file) with open(tmp_file, 'rb') as self.__f: for stmt in self._iter_through_csxml_file_from_handle(): yield stmt percent_done_now = floor(100.0 * self.files_processed / num_files) if percent_done_now > percent_done: percent_done = percent_done_now ellapsed_s = time.time() - start_time_s ellapsed_min = ellapsed_s / 60.0 msg = 'Processed %d of %d files (%f%% complete, %f minutes)' % \ (self.files_processed, num_files, percent_done, ellapsed_min) # Delete the temporary directory shutil.rmtree(self.__tmp_dir) return
[docs] def process_csxml_file(self, filename, interval=None, lazy=False): """Processes a filehandle to MedScan csxml input into INDRA statements. The CSXML format consists of a top-level `<batch>` root element containing a series of `<doc>` (document) elements, in turn containing `<sec>` (section) elements, and in turn containing `<sent>` (sentence) elements. Within the `<sent>` element, a series of additional elements appear in the following order: * `<toks>`, which contains a tokenized form of the sentence in its text attribute * `<textmods>`, which describes any preprocessing/normalization done to the underlying text * `<match>` elements, each of which contains one of more `<entity>` elements, describing entities in the text with their identifiers. The local IDs of each entities are given in the `msid` attribute of this element; these IDs are then referenced in any subsequent SVO elements. * `<svo>` elements, representing subject-verb-object triples. SVO elements with a `type` attribute of `CONTROL` represent normalized regulation relationships; they often represent the normalized extraction of the immediately preceding (but unnormalized SVO element). However, in some cases there can be a "CONTROL" SVO element without its parent immediately preceding it. Parameters ---------- filename : string The path to a Medscan csxml file. interval : (start, end) or None Select the interval of documents to read, starting with the `start`th document and ending before the `end`th document. If either is None, the value is considered undefined. If the value exceeds the bounds of available documents, it will simply be ignored. lazy : bool If True, only create a generator which can be used by the `get_statements` method. If True, populate the statements list now. """ if interval is None: interval = (None, None) tmp_fname = tempfile.mktemp(os.path.basename(filename)) fix_character_encoding(filename, tmp_fname) self.__f = open(tmp_fname, 'rb') self._gen = self._iter_through_csxml_file_from_handle(*interval) if not lazy: for stmt in self._gen: self.statements.append(stmt) return
def _iter_through_csxml_file_from_handle(self, start=None, stop=None): pmid = None sec = None tagged_sent = None doc_idx = 0 entities = {} match_text = None in_prop = False last_relation = None property_entities = [] property_name = None # Go through the document again and extract statements good_relations = [] skipping_doc = False skipping_sent = False for event, elem in lxml.etree.iterparse(self.__f, events=('start', 'end'), encoding='utf-8', recover=True): if elem.tag in ['attr', 'toks']: continue # If opening up a new doc, set the PMID if event == 'start' and elem.tag == 'doc': if start is not None and doc_idx < start:"Skipping document number %d." % doc_idx) skipping_doc = True continue if stop is not None and doc_idx >= stop:"Reach the end of the allocated docs.") break uri = elem.attrib.get('uri') re_pmid = PMID_PATT.match(uri) if re_pmid is None: logger.warning("Could not extract pmid from: %s." % uri) skipping_doc = True pmid = pmid_num = int(pmid) if pmid_num in self._pmids_handled: logger.warning("Skipping repeated pmid: %s from %s." % (pmid, skipping_doc = True # If getting a section, set the section type elif event == 'start' and elem.tag == 'sec' and not skipping_doc: sec = elem.attrib.get('type') # Set the sentence context elif event == 'start' and elem.tag == 'sent' and not skipping_doc: tagged_sent = elem.attrib.get('msrc') h = hash(tagged_sent) if h in self._sentences_handled: skipping_sent = True continue skipping_sent = False # Reset last_relation between sentences, since we will only be # interested in the relation immediately preceding a CONTROL # statement but within the same sentence. last_relation = None entities = {} elif event == 'end' and elem.tag == 'sent' and not skipping_doc \ and not skipping_sent: # End of sentence; deduplicate and copy statements from this # sentence to the main statements list for s in self.sentence_statements: yield s self.sentence_statements = [] self._sentences_handled.add(h) good_relations = [] # Reset site info self.last_site_info_in_sentence = None elif event == 'start' and elem.tag == 'match' and not skipping_doc\ and not skipping_sent: match_text = elem.attrib.get('chars') match_start = int(elem.attrib.get('coff')) match_end = int(elem.attrib.get('clen')) + match_start elif event == 'start' and elem.tag == 'entity' \ and not skipping_doc and not skipping_sent: if not in_prop: ent_id = elem.attrib['msid'] ent_urn = elem.attrib.get('urn') ent_type = elem.attrib['type'] entities[ent_id] = MedscanEntity(match_text, ent_urn, ent_type, {}, match_start, match_end) else: ent_type = elem.attrib['type'] ent_urn = elem.attrib['urn'] ent_name = elem.attrib['name'] property_entities.append(MedscanEntity(ent_name, ent_urn, ent_type, None, None, None)) elif event == 'start' and elem.tag == 'svo' and not skipping_doc \ and not skipping_sent: subj = elem.attrib.get('subj') verb = elem.attrib.get('verb') obj = elem.attrib.get('obj') svo_type = elem.attrib.get('type') # Aggregate information about the relation relation = MedscanRelation(pmid=pmid, sec=sec, uri=uri, tagged_sentence=tagged_sent, entities=entities, subj=subj, verb=verb, obj=obj, svo_type=svo_type) if svo_type == 'CONTROL': good_relations.append(relation) self.process_relation(relation, last_relation) else: # Sometimes a CONTROL SVO can be after an unnormalized SVO # that is a more specific but less uniform version of the # same extracted statement. last_relation = relation elif event == 'start' and elem.tag == 'prop' and not skipping_doc \ and not skipping_sent: in_prop = True property_name = elem.attrib.get('name') property_entities = [] elif event == 'end' and elem.tag == 'prop' and not skipping_doc \ and not skipping_sent: in_prop = False entities[ent_id].properties[property_name] = property_entities elif event == 'end' and elem.tag == 'doc': doc_idx += 1 # Give a status update if doc_idx % 100 == 0:"Processed %d documents" % doc_idx) self._pmids_handled.add(pmid_num) self._sentences_handled = set() # Solution for memory leak found here: # elem.clear() self.files_processed += 1 self.__f.close() return def _add_statement(self, stmt): if not _is_statement_in_list(stmt, self.sentence_statements): self.sentence_statements.append(stmt) return
[docs] def process_relation(self, relation, last_relation): """Process a relation into an INDRA statement. Parameters ---------- relation : MedscanRelation The relation to process (a CONTROL svo with normalized verb) last_relation : MedscanRelation The relation immediately proceding the relation to process within the same sentence, or None if there are no preceding relations within the same sentence. This proceeding relation, if available, will refer to the same interaction but with an unnormalized (potentially more specific) verb, and is used when processing protein modification events. """ subj_res = self.agent_from_entity(relation, relation.subj) obj_res = self.agent_from_entity(relation, relation.obj) if subj_res is None or obj_res is None: # Don't extract a statement if the subject or object cannot # be resolved return subj, subj_bounds = subj_res obj, obj_bounds = obj_res # Make evidence object untagged_sentence = _untag_sentence(relation.tagged_sentence) if last_relation: last_verb = last_relation.verb else: last_verb = None # Get the entity information with the character coordinates annotations = {'verb': relation.verb, 'last_verb': last_verb, 'agents': {'coords': [subj_bounds, obj_bounds]}} epistemics = dict() epistemics['direct'] = False # Overridden later if needed ev = [Evidence(source_api='medscan', source_id=relation.uri, pmid=relation.pmid, text=untagged_sentence, annotations=annotations, epistemics=epistemics)] if relation.verb in INCREASE_AMOUNT_VERBS: # If the normalized verb corresponds to an IncreaseAmount statement # then make one self._add_statement(IncreaseAmount(subj, obj, evidence=ev)) elif relation.verb in DECREASE_AMOUNT_VERBS: # If the normalized verb corresponds to a DecreaseAmount statement # then make one self._add_statement(DecreaseAmount(subj, obj, evidence=ev)) elif relation.verb in ALL_ACTIVATION_VERBS: # If the normalized verb corresponds to an Activation statement, # then make one if relation.verb in D_ACTIVATION_VERBS: ev[0].epistemics['direction'] = True self._add_statement(Activation(subj, obj, evidence=ev)) elif relation.verb in ALL_INHIBITION_VERBS: # If the normalized verb corresponds to an Inhibition statement, # then make one if relation.verb in D_INHIBITION_VERBS: ev[0].epistemics['direct'] = True self._add_statement(Inhibition(subj, obj, evidence=ev)) elif relation.verb == 'ProtModification': # The normalized verb 'ProtModification' is too vague to make # an INDRA statement. We look at the unnormalized verb in the # previous svo element, if available, to decide what type of # INDRA statement to construct. if last_relation is None: # We cannot make a statement unless we have more fine-grained # information on the relation type from a preceding # unnormalized SVO return # Map the unnormalized verb to an INDRA statement type if last_relation.verb == 'TK{phosphorylate}': statement_type = Phosphorylation elif last_relation.verb == 'TK{dephosphorylate}': statement_type = Dephosphorylation elif last_relation.verb == 'TK{ubiquitinate}': statement_type = Ubiquitination elif last_relation.verb == 'TK{acetylate}': statement_type = Acetylation elif last_relation.verb == 'TK{methylate}': statement_type = Methylation elif last_relation.verb == 'TK{deacetylate}': statement_type = Deacetylation elif last_relation.verb == 'TK{demethylate}': statement_type = Demethylation elif last_relation.verb == 'TK{hyperphosphorylate}': statement_type = Phosphorylation elif last_relation.verb == 'TK{hydroxylate}': statement_type = Hydroxylation elif last_relation.verb == 'TK{sumoylate}': statement_type = Sumoylation elif last_relation.verb == 'TK{palmitoylate}': statement_type = Palmitoylation elif last_relation.verb == 'TK{glycosylate}': statement_type = Glycosylation elif last_relation.verb == 'TK{ribosylate}': statement_type = Ribosylation elif last_relation.verb == 'TK{deglycosylate}': statement_type = Deglycosylation elif last_relation.verb == 'TK{myristylate}': statement_type = Myristoylation elif last_relation.verb == 'TK{farnesylate}': statement_type = Farnesylation elif last_relation.verb == 'TK{desumoylate}': statement_type = Desumoylation elif last_relation.verb == 'TK{geranylgeranylate}': statement_type = Geranylgeranylation elif last_relation.verb == 'TK{deacylate}': statement_type = Deacetylation else: # This unnormalized verb is not handled, do not extract an # INDRA statement return obj_text = obj.db_refs['TEXT'] last_info = self.last_site_info_in_sentence if last_info is not None and obj_text == last_info.object_text: for site in self.last_site_info_in_sentence.get_sites(): r = site.residue p = site.position s = statement_type(subj, obj, residue=r, position=p, evidence=ev) self._add_statement(s) else: self._add_statement(statement_type(subj, obj, evidence=ev)) elif relation.verb == 'Binding': # The Binding normalized verb corresponds to the INDRA Complex # statement. self._add_statement( Complex([subj, obj], evidence=ev) ) elif relation.verb == 'ProtModification-negative': pass # TODO? These occur so infrequently so maybe not worth it elif relation.verb == 'Regulation-unknown': pass # TODO? These occur so infrequently so maybe not worth it elif relation.verb == 'StateEffect-positive': pass # self._add_statement( # ActiveForm(subj, obj, evidence=ev) # ) # TODO: disabling for now, since not sure whether we should set # the is_active flag elif relation.verb == 'StateEffect': self.last_site_info_in_sentence = \ ProteinSiteInfo(, object_text=obj.db_refs['TEXT']) return
[docs] def agent_from_entity(self, relation, entity_id): """Create a (potentially grounded) INDRA Agent object from a given Medscan entity describing the subject or object. Uses helper functions to convert a Medscan URN to an INDRA db_refs grounding dictionary. If the entity has properties indicating that it is a protein with a mutation or modification, then constructs the needed ModCondition or MutCondition. Parameters ---------- relation : MedscanRelation The current relation being processed entity_id : str The ID of the entity to process Returns ------- agent : indra.statements.Agent A potentially grounded INDRA agent representing this entity """ # Extract sentence tags mapping ids to the text. We refer to this # mapping only if the entity doesn't appear in the grounded entity # list tags = _extract_sentence_tags(relation.tagged_sentence) if entity_id is None: return None self.num_entities += 1 entity_id = _extract_id(entity_id) if entity_id not in relation.entities and \ entity_id not in tags: # Could not find the entity in either the list of grounded # entities of the items tagged in the sentence. Happens for # a very small percentage of the dataset. self.num_entities_not_found += 1 return None if entity_id not in relation.entities: # The entity is not in the grounded entity list # Instead, make an ungrounded entity, with TEXT corresponding to # the words with the given entity id tagged in the sentence. entity_data = tags[entity_id] db_refs = {'TEXT': entity_data['text']} ag = Agent(normalize_medscan_name(db_refs['TEXT']), db_refs=db_refs) return ag, entity_data['bounds'] else: entity = relation.entities[entity_id] bounds = (entity.ch_start, entity.ch_end) prop = if len(prop.keys()) == 2 and 'Protein' in prop \ and 'Mutation' in prop: # Handle the special case where the entity is a protein # with a mutation or modification, with those details # described in the entity properties protein = prop['Protein'] assert(len(protein) == 1) protein = protein[0] mutation = prop['Mutation'] assert(len(mutation) == 1) mutation = mutation[0] db_refs, db_name = _urn_to_db_refs(protein.urn) if db_refs is None: return None db_refs['TEXT'] = if db_name is None: agent_name = db_refs['TEXT'] else: agent_name = db_name # Check mutation.type. Only some types correspond to situations # that can be represented in INDRA; return None if we cannot # map to an INDRA statement (which will block processing of # the statement in process_relation). if mutation.type == 'AASite': # Do not handle this # Example: # MedscanEntity(name='D1', urn='urn:agi-aa:D1', # type='AASite', properties=None) return None elif mutation.type == 'Mutation': # Convert mutation properties to an INDRA MutCondition r_old, pos, r_new = _parse_mut_string( if r_old is None: logger.warning('Could not parse mutation string: ' + # Don't create an agent return None else: try: cond = MutCondition(pos, r_old, r_new) ag = Agent(normalize_medscan_name(agent_name), db_refs=db_refs, mutations=[cond]) return ag, bounds except BaseException: logger.warning('Could not parse mutation ' + 'string: ' + return None elif mutation.type == 'MethSite': # Convert methylation site information to an INDRA # ModCondition res, pos = _parse_mod_string( if res is None: return None cond = ModCondition('methylation', res, pos) ag = Agent(normalize_medscan_name(agent_name), db_refs=db_refs, mods=[cond]) return ag, bounds # Example: # MedscanEntity(name='R457', # urn='urn:agi-s-llid:R457-2185', type='MethSite', # properties=None) elif mutation.type == 'PhosphoSite': # Convert phosphorylation site information to an INDRA # ModCondition res, pos = _parse_mod_string( if res is None: return None cond = ModCondition('phosphorylation', res, pos) ag = Agent(normalize_medscan_name(agent_name), db_refs=db_refs, mods=[cond]) return ag, bounds # Example: # MedscanEntity(name='S455', # urn='urn:agi-s-llid:S455-47', type='PhosphoSite', # properties=None) pass elif mutation.type == 'Lysine': # Ambiguous whether this is a methylation or # demethylation; skip # Example: # MedscanEntity(name='K150', # urn='urn:agi-s-llid:K150-5624', type='Lysine', # properties=None) return None else: logger.warning('Processor currently cannot process ' + 'mutations of type ' + mutation.type) else: # Handle the more common case where we just ground the entity # without mutation or modification information db_refs, db_name = _urn_to_db_refs(entity.urn) if db_refs is None: return None db_refs['TEXT'] = if db_name is None: agent_name = db_refs['TEXT'] else: agent_name = db_name ag = Agent(normalize_medscan_name(agent_name), db_refs=db_refs) return ag, bounds
[docs]class MedscanRelation(object): """A structure representing the information contained in a Medscan SVO xml element as well as associated entities and properties. Attributes ---------- pmid : str The URI of the current document (such as a PMID) sec : str The section of the document the relation occurs in entities : dict A dictionary mapping entity IDs from the same sentence to MedscanEntity objects. tagged_sentence : str The sentence from which the relation was extracted, with some tagged phrases and annotations. subj : str The entity ID of the subject verb : str The verb in the relationship between the subject and the object obj : str The entity ID of the object svo_type : str The type of SVO relationship (for example, CONTROL indicates that the verb is normalized) """ def __init__(self, pmid, uri, sec, entities, tagged_sentence, subj, verb, obj, svo_type): self.pmid = pmid self.uri = uri self.sec = sec self.entities = entities self.tagged_sentence = tagged_sentence self.subj = subj self.verb = verb self.obj = obj self.svo_type = svo_type
[docs]def normalize_medscan_name(name): """Removes the "complex" and "complex complex" suffixes from a medscan agent name so that it better corresponds with the grounding map. Parameters ---------- name: str The Medscan agent name Returns ------- norm_name: str The Medscan agent name with the "complex" and "complex complex" suffixes removed. """ suffix = ' complex' for i in range(2): if name.endswith(suffix): name = name[:-len(suffix)] return name
MOD_PATT = re.compile('([A-Za-z])+([0-9]+)') def _parse_mod_string(s): """Parses a string referring to a protein modification of the form (residue)(position), such as T47. Parameters ---------- s : str A string representation of a protein residue and position being modified Returns ------- residue : str The residue being modified (example: T) position : str The position at which the modification is happening (example: 47) """ m = MOD_PATT.match(s) assert m is not None return m.groups() MUT_PATT = re.compile('([A-Za-z]+)([0-9]+)([A-Za-z]+)') def _parse_mut_string(s): """ A string representation of a protein mutation of the form (old residue)(position)(new residue). Example: T34U. Parameters ---------- s : str The string representation of the protein mutation Returns ------- old_residue : str The old residue, or None of the mutation string cannot be parsed position : str The position at which the mutation occurs, or None if the mutation string cannot be parsed new_residue : str The new residue, or None if the mutation string cannot be parsed """ m = MUT_PATT.match(s) if m is None: # Mutation string does not fit this pattern, other patterns not # currently supported return None, None, None else: return m.groups() URN_PATT = re.compile('urn:([^:]+):([^:]+)') def _urn_to_db_refs(urn): """Converts a Medscan URN to an INDRA db_refs dictionary with grounding information. Parameters ---------- urn : str A Medscan URN Returns ------- db_refs : dict A dictionary with grounding information, mapping databases to database identifiers. If the Medscan URN is not recognized, returns an empty dictionary. db_name : str The Famplex name, if available; otherwise the HGNC name if available; otherwise None """ # Convert a urn to a db_refs dictionary if urn is None: return {}, None m = URN_PATT.match(urn) if m is None: return None, None urn_type, urn_id = m.groups() db_refs = {} db_name = None # TODO: support more types of URNs if urn_type == 'agi-cas': # Identifier is CAS, convert to CHEBI chebi_id = get_chebi_id_from_cas(urn_id) if chebi_id: db_refs['CHEBI'] = 'CHEBI:%s' % chebi_id db_name = get_chebi_name_from_id(chebi_id) elif urn_type == 'agi-llid': # This is an Entrez ID, convert to HGNC hgnc_id = get_hgnc_from_entrez(urn_id) if hgnc_id is not None: db_refs['HGNC'] = hgnc_id # Convert the HGNC ID to a Uniprot ID uniprot_id = get_uniprot_id(hgnc_id) if uniprot_id is not None: db_refs['UP'] = uniprot_id # Try to lookup HGNC name; if it's available, set it to the # agent name db_name = get_hgnc_name(hgnc_id) elif urn_type in ['agi-meshdis', 'agi-ncimorgan', 'agi-ncimtissue', 'agi-ncimcelltype']: if urn_id.startswith('C') and urn_id[1:].isdigit(): # Identifier is probably UMLS db_refs['UMLS'] = urn_id else: # Identifier is MESH urn_mesh_name = unquote(urn_id) mesh_id, mesh_name = mesh_client.get_mesh_id_name(urn_mesh_name) if mesh_id: db_refs['MESH'] = mesh_id db_name = mesh_name else: db_name = urn_mesh_name elif urn_type == 'agi-gocomplex': # Identifier is GO db_refs['GO'] = 'GO:%s' % urn_id elif urn_type == 'agi-go': # Identifier is GO db_refs['GO'] = 'GO:%s' % urn_id # If we have a GO or MESH grounding, see if there is a corresponding # Famplex grounding db_sometimes_maps_to_famplex = ['GO', 'MESH'] for db in db_sometimes_maps_to_famplex: if db in db_refs: key = (db, db_refs[db]) if key in famplex_map: db_refs['FPLX'] = famplex_map[key] # If the urn corresponds to an eccode, groudn to famplex if that eccode # is in the Famplex equivalences table if urn.startswith('urn:agi-enz'): tokens = urn.split(':') eccode = tokens[2] key = ('ECCODE', eccode) if key in famplex_map: db_refs['FPLX'] = famplex_map[key] # If the Medscan URN itself maps to a Famplex id, add a Famplex grounding key = ('MEDSCAN', urn) if key in famplex_map: db_refs['FPLX'] = famplex_map[key] # If there is a Famplex grounding, use Famplex for entity name if 'FPLX' in db_refs: db_name = db_refs['FPLX'] elif 'GO' in db_refs: db_name = go_client.get_go_label(db_refs['GO']) return db_refs, db_name TAG_PATT = re.compile('ID{([0-9,]+)=([^}]+)}') JUNK_PATT = re.compile('(CONTEXT|GLOSSARY){[^}]+}+') ID_PATT = re.compile('ID\\{([0-9]+)\\}') def _extract_id(id_string): """Extracts the numeric ID from the representation of the subject or object ID that appears as an attribute of the svo element in the Medscan XML document. Parameters ---------- id_string : str The ID representation that appears in the svo element in the XML document (example: ID{123}) Returns ------- id : str The numeric ID, extracted from the svo element's attribute (example: 123) """ matches = ID_PATT.match(id_string) assert matches is not None return def _untag_sentence(tagged_sentence): """Removes all tags in the sentence, returning the original sentence without Medscan annotations. Parameters ---------- tagged_sentence : str The tagged sentence Returns ------- untagged_sentence : str Sentence with tags and annotations stripped out """ untagged_sentence = TAG_PATT.sub('\\2', tagged_sentence) clean_sentence = JUNK_PATT.sub('', untagged_sentence) return clean_sentence.strip() def _extract_sentence_tags(tagged_sentence): """Given a tagged sentence, extracts a dictionary mapping tags to the words or phrases that they tag. Parameters ---------- tagged_sentence : str The sentence with Medscan annotations and tags Returns ------- tags : dict A dictionary mapping tags to the words or phrases that they tag. """ untagged_sentence = _untag_sentence(tagged_sentence) decluttered_sentence = JUNK_PATT.sub('', tagged_sentence) tags = {} # Iteratively look for all matches of this pattern endpos = 0 while True: match =, pos=endpos) if not match: break endpos = match.end() text = text = text.replace('CONTEXT', '') text = text.replace('GLOSSARY', '') text = text.strip() start = untagged_sentence.index(text) stop = start + len(text) tag_key = if ',' in tag_key: for sub_key in tag_key.split(','): if sub_key == '0': continue tags[sub_key] = {'text': text, 'bounds': (start, stop)} else: tags[tag_key] = {'text': text, 'bounds': (start, stop)} return tags