<commit_before>import os
import sys
import logging
import indra
from indra.util import read_unicode_csv
from indra.tools import expand_families
from indra.sources import trips
from bioagents import Bioagent
from indra.databases import get_identifiers_url
from indra.preassembler.hierarchy_manager import hierarchies
from kqml import KQMLModule, KQMLPerformative, KQMLList, KQMLString


logging.basicConfig(format='%(levelname)s: %(name)s - %(message)s',
                    level=logging.INFO)
logger = logging.getLogger('BIOSENSE')


_indra_path = indra.__path__[0]


class BioSense_Module(Bioagent):
    name = 'BioSense'
    tasks = ['CHOOSE-SENSE', 'CHOOSE-SENSE-CATEGORY',
             'CHOOSE-SENSE-IS-MEMBER', 'CHOOSE-SENSE-WHAT-MEMBER']

    def respond_choose_sense(self, content):
        """Return response content to choose-sense request."""
        ekb = content.gets('ekb-term')
        agents, ambiguities = process_ekb(ekb)
        msg = KQMLPerformative('SUCCESS')
        if agents:
            kagents = []
            for term_id, agent_tuple in agents.items():
                kagent = get_kagent(agent_tuple, term_id)
                kagents.append(kagent)
            msg.set('agents', KQMLList(kagents))
        if ambiguities:
            ambiguities_msg = get_ambiguities_msg(ambiguities)
            msg.set('ambiguities', ambiguities_msg)
        return msg

    def respond_choose_sense_category(self, content):
        """Return response content to choose-sense-category request."""
        ekb = content.gets('ekb-term')
        agents, _ = process_ekb(ekb)
        if len(agents) != 1:
            return make_failure('INVALID_AGENT')
        agent = list(agents.values())[0][0]
        category = content.gets('category')
        logger.info("Checking %s for category %s." % (agent, category))
        reg_cat = category.lower().replace('-', ' ').replace('W::', '').replace('w::', '')
        logger.info("Regularized category to \"%s\"." % reg_cat)
        if reg_cat in ['kinase', 'kinase activity']:
            msg = KQMLList('SUCCESS')
            if agent.name in kinase_list:
                msg.set('in-category', 'TRUE')
            else:
                msg.set('in-category', 'FALSE')
        elif reg_cat == 'transcription factor':
            msg = KQMLList('SUCCESS')
            if agent.name in tf_list:
                msg.set('in-category', 'TRUE')
            else:
                msg.set('in-category', 'FALSE')
        elif reg_cat == 'phosphatase':
            msg = KQMLList('SUCCESS')
            if agent.name in phosphatase_list:
                msg.set('in-category', 'TRUE')
            else:
                msg.set('in-category', 'FALSE')
        elif reg_cat == 'enzyme':
            msg = KQMLList('SUCCESS')
            if agent.name in phosphatase_list or agent.name in kinase_list:
                msg.set('in-category', 'TRUE')
            else:
                msg.set('in-category', 'FALSE')
        else:
            msg = make_failure('UNKNOWN_CATEGORY')
            logger.info("Regularized category \"%s\" not recognized: options "
                        "are %s." % (reg_cat, ['kinase', 'kinase activity',
                                               'enzyme', 'transcription factor',
                                               'phosphatase']))
        return msg

    def respond_choose_sense_is_member(self, content):
        """Return response content to choose-sense-is-member request."""
        # Get the member agent first
        ekb = content.gets('ekb-term')
        agents, _ = process_ekb(ekb)
        if len(agents) != 1:
            return make_failure('INVALID_AGENT')
        member_agent = list(agents.values())[0][0]
        # Get the collection next
        ekb = content.gets('collection')
        agents, _ = process_ekb(ekb)
        if len(agents) != 1:
            return make_failure('INVALID_COLLECTION')
        collection_agent = list(agents.values())[0][0]
        if member_agent.isa(collection_agent, hierarchies):
            is_member = 'TRUE'
        else:
            is_member = 'FALSE'
        msg = KQMLList('SUCCESS')
        msg.set('is-member', is_member)
        return msg

    def respond_choose_sense_what_member(self, content):
        """Return response content to choose-sense-what-member request."""
        # Get the collection agent
        ekb = content.gets('collection')
        agents, _ = process_ekb(ekb)
        if len(agents) != 1:
            return make_failure('INVALID_COLLECTION')

        term_id, (agent, ont_type, urls) = list(agents.items())[0]
        members = get_members(agent)
        if members is None:
            return make_failure('COLLECTION_NOT_FAMILY_OR_COMPLEX')

        kagents = [get_kagent((m, 'ONT::PROTEIN', get_urls(m)))
                   for m in members]

        msg = KQMLList('SUCCESS')
        msg.set('members', KQMLList(kagents))
        return msg

    def respond_get_synonyms(self, content):
        """Respond to a query looking for synonyms of a gene."""
        ekb = content.gets('gene')
        agents, _ = process_ekb(ekb)
        if len(agents) != 1:
            return self.make_failure('INVALID_COLLECTION')

        agent = agents[0]
        up_id = agent.db_refs['UP']
        gene_name = up_client.get_gene_name(up_id)
        gene_synonyms = up_client.get_gene_alt_labels(up_id)
        protein_names = up_client.get_gene_alt_names(up_id)
        msg = KQMLList('SUCCESS')
        msg.set('gene_name', gene_name)
        msg.set('gene_synonyms', gene_synonyms)
        msg.set('protein_names', protein_names)
        return msg


def get_kagent(agent_tuple, term_id=None):
    agent, ont_type, urls = agent_tuple
    db_refs = '|'.join('%s:%s' % (k, v) for k, v in
                       agent.db_refs.items())
    kagent = KQMLList(term_id) if term_id else KQMLList()
    kagent.sets('name', agent.name)
    kagent.sets('ids', db_refs)
    url_parts = [KQMLList([':name', KQMLString(k),
                           ':dblink', KQMLString(v)])
                 for k, v in urls.items()]
    url_list = KQMLList()
    for url_part in url_parts:
        url_list.append(url_part)
    kagent.set('id-urls', url_list)<commit_msg>Make response to getting synonyms work<commit_after>import os
import sys
import logging
import indra
from indra.util import read_unicode_csv
from indra.tools import expand_families
from indra.sources import trips
from indra.sources.trips.processor import TripsProcessor
from bioagents import Bioagent
from indra.databases import get_identifiers_url, uniprot_client
from indra.preassembler.hierarchy_manager import hierarchies
from kqml import KQMLModule, KQMLPerformative, KQMLList, KQMLString


logging.basicConfig(format='%(levelname)s: %(name)s - %(message)s',
                    level=logging.INFO)
logger = logging.getLogger('BIOSENSE')


_indra_path = indra.__path__[0]


class BioSense_Module(Bioagent):
    name = 'BioSense'
    tasks = ['CHOOSE-SENSE', 'CHOOSE-SENSE-CATEGORY',
             'CHOOSE-SENSE-IS-MEMBER', 'CHOOSE-SENSE-WHAT-MEMBER']

    def respond_choose_sense(self, content):
        """Return response content to choose-sense request."""
        ekb = content.gets('ekb-term')
        agents, ambiguities = process_ekb(ekb)
        msg = KQMLPerformative('SUCCESS')
        if agents:
            kagents = []
            for term_id, agent_tuple in agents.items():
                kagent = get_kagent(agent_tuple, term_id)
                kagents.append(kagent)
            msg.set('agents', KQMLList(kagents))
        if ambiguities:
            ambiguities_msg = get_ambiguities_msg(ambiguities)
            msg.set('ambiguities', ambiguities_msg)
        return msg

    def respond_choose_sense_category(self, content):
        """Return response content to choose-sense-category request."""
        ekb = content.gets('ekb-term')
        agents, _ = process_ekb(ekb)
        if len(agents) != 1:
            return make_failure('INVALID_AGENT')
        agent = list(agents.values())[0][0]
        category = content.gets('category')
        logger.info("Checking %s for category %s." % (agent, category))
        reg_cat = category.lower().replace('-', ' ').replace('W::', '').replace('w::', '')
        logger.info("Regularized category to \"%s\"." % reg_cat)
        if reg_cat in ['kinase', 'kinase activity']:
            msg = KQMLList('SUCCESS')
            if agent.name in kinase_list:
                msg.set('in-category', 'TRUE')
            else:
                msg.set('in-category', 'FALSE')
        elif reg_cat == 'transcription factor':
            msg = KQMLList('SUCCESS')
            if agent.name in tf_list:
                msg.set('in-category', 'TRUE')
            else:
                msg.set('in-category', 'FALSE')
        elif reg_cat == 'phosphatase':
            msg = KQMLList('SUCCESS')
            if agent.name in phosphatase_list:
                msg.set('in-category', 'TRUE')
            else:
                msg.set('in-category', 'FALSE')
        elif reg_cat == 'enzyme':
            msg = KQMLList('SUCCESS')
            if agent.name in phosphatase_list or agent.name in kinase_list:
                msg.set('in-category', 'TRUE')
            else:
                msg.set('in-category', 'FALSE')
        else:
            msg = make_failure('UNKNOWN_CATEGORY')
            logger.info("Regularized category \"%s\" not recognized: options "
                        "are %s." % (reg_cat, ['kinase', 'kinase activity',
                                               'enzyme', 'transcription factor',
                                               'phosphatase']))
        return msg

    def respond_choose_sense_is_member(self, content):
        """Return response content to choose-sense-is-member request."""
        # Get the member agent first
        ekb = content.gets('ekb-term')
        agents, _ = process_ekb(ekb)
        if len(agents) != 1:
            return make_failure('INVALID_AGENT')
        member_agent = list(agents.values())[0][0]
        # Get the collection next
        ekb = content.gets('collection')
        agents, _ = process_ekb(ekb)
        if len(agents) != 1:
            return make_failure('INVALID_COLLECTION')
        collection_agent = list(agents.values())[0][0]
        if member_agent.isa(collection_agent, hierarchies):
            is_member = 'TRUE'
        else:
            is_member = 'FALSE'
        msg = KQMLList('SUCCESS')
        msg.set('is-member', is_member)
        return msg

    def respond_choose_sense_what_member(self, content):
        """Return response content to choose-sense-what-member request."""
        # Get the collection agent
        ekb = content.gets('collection')
        agents, _ = process_ekb(ekb)
        if len(agents) != 1:
            return make_failure('INVALID_COLLECTION')

        term_id, (agent, ont_type, urls) = list(agents.items())[0]
        members = get_members(agent)
        if members is None:
            return make_failure('COLLECTION_NOT_FAMILY_OR_COMPLEX')

        kagents = [get_kagent((m, 'ONT::PROTEIN', get_urls(m)))
                   for m in members]

        msg = KQMLList('SUCCESS')
        msg.set('members', KQMLList(kagents))
        return msg

    def respond_get_synonyms(self, content):
        """Respond to a query looking for synonyms of a protein."""
        ekb = content.gets('protein')
        try:
            agent = self._get_agent(ekb)
        except Exception as e:
            return self.make_failure('INVALID_AGENT')
        if agent is None:
            return self.make_failure('INVALID_AGENT')

        up_id = agent.db_refs.get('UP')
        if not up_id:
            return self.make_failure('INVALID_AGENT')

        synonyms = uniprot_client.get_synonyms(up_id)
        syns = KQMLList([KQMLString(s) for s in synonyms])
        msg = KQMLList('SUCCESS')
        msg.set('synonyms', syns)
        return msg

    @staticmethod
    def _get_agent(agent_ekb):
        tp = TripsProcessor(agent_ekb)
        terms = tp.tree.findall('TERM')
        term_id = terms[0].attrib['id']
        agent = tp._get_agent_by_id(term_id, None)
        return agent


def get_kagent(agent_tuple, term_id=None):
    agent, ont_type, urls = agent_tuple
    db_refs = '|'.join('%s:%s' % (k, v) for k, v in
                       agent.db_refs.items())
    kagent = KQMLList(term_id) if term_id else KQMLList()
    kagent.sets('name', agent.name)
    kagent.sets('ids', db_refs)
    url_parts = [KQMLList([':name', KQMLString(k),
                           ':dblink', KQMLString(v)])
                 for k, v in urls.items()]
    url_list = KQMLList()
    for url_part in url_parts:
        url_list.append(url_part)
    kagent.set('id-urls', url_list)
{'fraction_non_alphanumeric': 0.11937377690802348, 'fraction_numerical': 0.0021135029354207436, 'mean_word_length': 2.248410882278159, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2348832', 'n_tokens_mistral': 4018, 'n_tokens_neox': 3849, 'n_words': 846}