json_brain_server.py

#!/usr/bin/env jython -J-Xmx2000m

from uk.ac.ebi.brain.core import Brain
import json
from owltools.graph import OWLGraphWrapper
import re
import sys

## This script defines a class for running Brain queries using input specified as JSON
## It may be useful for designing an http server for running Brian queries via JSON post requests

class ont_server():

    def __init__(self, ont_url):
        """ont_url = ontology URL as string.  
        Note, this will also take a plain file path."""
        self.brain = Brain()
        self.brain.learn(ont_url)
        self.ont = self.brain.getOntology() # Plain, OWL-API ontology object.
        self.ogw = OWLGraphWrapper(self.ont)

    def __string__(self):
        ontid = self.ont.getOntologyID()
        iri = ontid.getVersionIRI()
        return "ont_server running " + iri.toString()

    def get_syns(self, owl_id):
        """Specify and OWL shortForm ID, return a list of synonyms strings"""
        obo_id = re.sub('_', ':', owl_id) 
        clazo = self.ogw.getOWLClassByIdentifier(obo_id)
        return list(self.ogw.getOBOSynonymStrings(clazo, []))

    def query(self, query_string):
        """query_string = JSON *string*
        Spec:  { "query_type": 'descendants|instances'
        , "query" : "MS string"
        . "return": ['name', 'synonyms', ...]}
        """
        query = json.loads(query_string)
        out = {}
        query_results = []
        if query['query_type'] == 'descendants':
            query_results = self.brain.getSubClasses(query['query'], 0)
            out = self.roll_results(query_results, query['return'])
        elif  query['query_type'] == 'instances':
            query_results = self.brain.getInstances(query['query'], 0)
            out = self.roll_results(query_results, query['return'])
        else: 
            out['query_status'] = "Unknown query type %s" %  query['query_type']
        return json.dumps(out, sort_keys=True, indent=4) # Probably want to prettify a bit
    
    def roll_results(self, query_results, query_return):
        out = {}
        if not query_results:
            out['query_status'] = 'Empty'
        else:
            out['query_status'] = 'Success'
            rdl = []
            for r in query_results:
                rd = {}
                rd['ID'] = r
                if 'name' in query_return:
                    rd['name'] = self.brain.getLabel(r)
                if 'synonyms' in query_return:
                    syn = self.get_syns(r)
                    if syn: rd['synonyms'] = self.get_syns(r)
                rdl.append(rd)
            out['query_results'] = rdl
        return out


## For testing.
os = ont_server("http://purl.obolibrary.org/obo/fbbt/fbbt-simple.owl")
test_query1  = { 'query_type' : 'descendants', 
                'query' : 'FBbt_00005106 that RO_0002131 some FBbt_00003679', 
                'return': ['name', 'synonyms']}

print os.query(json.dumps(test_query1))