|
from collections import deque, Counter
|
|
import warnings
|
|
import pandas as pd
|
|
import numpy as np
|
|
from xml.etree import ElementTree as ET
|
|
import math
|
|
|
|
BIOLOGICAL_PROCESS = 'GO:0008150'
|
|
MOLECULAR_FUNCTION = 'GO:0003674'
|
|
CELLULAR_COMPONENT = 'GO:0005575'
|
|
FUNC_DICT = {
|
|
'cc': CELLULAR_COMPONENT,
|
|
'mf': MOLECULAR_FUNCTION,
|
|
'bp': BIOLOGICAL_PROCESS}
|
|
|
|
NAMESPACES = {
|
|
'cc': 'cellular_component',
|
|
'mf': 'molecular_function',
|
|
'bp': 'biological_process'
|
|
}
|
|
|
|
EXP_CODES = set([
|
|
'EXP', 'IDA', 'IPI', 'IMP', 'IGI', 'IEP', 'TAS', 'IC',
|
|
'HTP', 'HDA', 'HMP', 'HGI', 'HEP'])
|
|
|
|
|
|
CAFA_TARGETS = set([
|
|
'287', '3702', '4577', '6239', '7227', '7955', '9606', '9823', '10090',
|
|
'10116', '44689', '83333', '99287', '226900', '243273', '284812', '559292'])
|
|
|
|
|
|
def is_cafa_target(org):
|
|
return org in CAFA_TARGETS
|
|
|
|
|
|
def is_exp_code(code):
|
|
return code in EXP_CODES
|
|
|
|
|
|
def get_goplus_defs(filename='data/definitions.txt'):
|
|
plus_defs = {}
|
|
with open(filename) as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
go_id, definition = line.split(': ')
|
|
go_id = go_id.replace('_', ':')
|
|
definition = definition.replace('_', ':')
|
|
plus_defs[go_id] = set(definition.split(' and '))
|
|
return plus_defs
|
|
|
|
|
|
class Ontology(object):
|
|
|
|
def __init__(self, filename='data/go.obo', with_rels=False):
|
|
self.ont = self.load(filename, with_rels)
|
|
self.ic = None
|
|
self.ic_norm = 0.0
|
|
|
|
def has_term(self, term_id):
|
|
return term_id in self.ont
|
|
|
|
def get_term(self, term_id):
|
|
if self.has_term(term_id):
|
|
return self.ont[term_id]
|
|
return None
|
|
|
|
def calculate_ic(self, annots):
|
|
cnt = Counter()
|
|
for x in annots:
|
|
cnt.update(x)
|
|
self.ic = {}
|
|
for go_id, n in cnt.items():
|
|
parents = self.get_parents(go_id)
|
|
if len(parents) == 0:
|
|
min_n = n
|
|
else:
|
|
min_n = min([cnt[x] for x in parents])
|
|
|
|
self.ic[go_id] = math.log(min_n / n, 2)
|
|
self.ic_norm = max(self.ic_norm, self.ic[go_id])
|
|
|
|
def get_ic(self, go_id):
|
|
if self.ic is None:
|
|
raise Exception('Not yet calculated')
|
|
if go_id not in self.ic:
|
|
return 0.0
|
|
return self.ic[go_id]
|
|
|
|
def get_norm_ic(self, go_id):
|
|
return self.get_ic(go_id) / self.ic_norm
|
|
|
|
def load(self, filename, with_rels):
|
|
ont = dict()
|
|
obj = None
|
|
with open(filename, 'r') as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
if line == '[Term]':
|
|
if obj is not None:
|
|
ont[obj['id']] = obj
|
|
obj = dict()
|
|
obj['is_a'] = list()
|
|
obj['part_of'] = list()
|
|
obj['regulates'] = list()
|
|
obj['alt_ids'] = list()
|
|
obj['is_obsolete'] = False
|
|
continue
|
|
elif line == '[Typedef]':
|
|
if obj is not None:
|
|
ont[obj['id']] = obj
|
|
obj = None
|
|
else:
|
|
if obj is None:
|
|
continue
|
|
l = line.split(": ")
|
|
if l[0] == 'id':
|
|
obj['id'] = l[1]
|
|
elif l[0] == 'alt_id':
|
|
obj['alt_ids'].append(l[1])
|
|
elif l[0] == 'namespace':
|
|
obj['namespace'] = l[1]
|
|
elif l[0] == 'is_a':
|
|
obj['is_a'].append(l[1].split(' ! ')[0])
|
|
elif with_rels and l[0] == 'relationship':
|
|
it = l[1].split()
|
|
|
|
obj['is_a'].append(it[1])
|
|
elif l[0] == 'name':
|
|
obj['name'] = l[1]
|
|
elif l[0] == 'is_obsolete' and l[1] == 'true':
|
|
obj['is_obsolete'] = True
|
|
if obj is not None:
|
|
ont[obj['id']] = obj
|
|
for term_id in list(ont.keys()):
|
|
for t_id in ont[term_id]['alt_ids']:
|
|
ont[t_id] = ont[term_id]
|
|
if ont[term_id]['is_obsolete']:
|
|
del ont[term_id]
|
|
for term_id, val in ont.items():
|
|
if 'children' not in val:
|
|
val['children'] = set()
|
|
for p_id in val['is_a']:
|
|
if p_id in ont:
|
|
if 'children' not in ont[p_id]:
|
|
ont[p_id]['children'] = set()
|
|
ont[p_id]['children'].add(term_id)
|
|
|
|
return ont
|
|
|
|
def get_anchestors(self, term_id):
|
|
if term_id not in self.ont:
|
|
return set()
|
|
term_set = set()
|
|
q = deque()
|
|
q.append(term_id)
|
|
while (len(q) > 0):
|
|
t_id = q.popleft()
|
|
if t_id not in term_set:
|
|
term_set.add(t_id)
|
|
for parent_id in self.ont[t_id]['is_a']:
|
|
if parent_id in self.ont:
|
|
q.append(parent_id)
|
|
return term_set
|
|
|
|
def get_prop_terms(self, terms):
|
|
prop_terms = set()
|
|
|
|
for term_id in terms:
|
|
prop_terms |= self.get_anchestors(term_id)
|
|
return prop_terms
|
|
|
|
def get_parents(self, term_id):
|
|
if term_id not in self.ont:
|
|
return set()
|
|
term_set = set()
|
|
for parent_id in self.ont[term_id]['is_a']:
|
|
if parent_id in self.ont:
|
|
term_set.add(parent_id)
|
|
return term_set
|
|
|
|
def get_namespace_terms(self, namespace):
|
|
terms = set()
|
|
for go_id, obj in self.ont.items():
|
|
if obj['namespace'] == namespace:
|
|
terms.add(go_id)
|
|
return terms
|
|
|
|
def get_namespace(self, term_id):
|
|
return self.ont[term_id]['namespace']
|
|
|
|
def get_term_set(self, term_id):
|
|
if term_id not in self.ont:
|
|
return set()
|
|
term_set = set()
|
|
q = deque()
|
|
q.append(term_id)
|
|
while len(q) > 0:
|
|
t_id = q.popleft()
|
|
if t_id not in term_set:
|
|
term_set.add(t_id)
|
|
for ch_id in self.ont[t_id]['children']:
|
|
q.append(ch_id)
|
|
return term_set
|
|
|
|
|
|
def read_fasta(filename):
|
|
seqs = list()
|
|
info = list()
|
|
seq = ''
|
|
inf = ''
|
|
with open(filename, 'r') as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if line.startswith('>'):
|
|
if seq != '':
|
|
seqs.append(seq)
|
|
info.append(inf)
|
|
seq = ''
|
|
inf = line[1:].split()[0]
|
|
else:
|
|
seq += line
|
|
seqs.append(seq)
|
|
info.append(inf)
|
|
return info, seqs
|
|
|
|
|
|
class DataGenerator(object):
|
|
|
|
def __init__(self, batch_size, is_sparse=False):
|
|
self.batch_size = batch_size
|
|
self.is_sparse = is_sparse
|
|
|
|
def fit(self, inputs, targets=None):
|
|
self.start = 0
|
|
self.inputs = inputs
|
|
self.targets = targets
|
|
if isinstance(self.inputs, tuple) or isinstance(self.inputs, list):
|
|
self.size = self.inputs[0].shape[0]
|
|
else:
|
|
self.size = self.inputs.shape[0]
|
|
self.has_targets = targets is not None
|
|
|
|
def __next__(self):
|
|
return self.next()
|
|
|
|
def reset(self):
|
|
self.start = 0
|
|
|
|
def next(self):
|
|
if self.start < self.size:
|
|
batch_index = np.arange(
|
|
self.start, min(self.size, self.start + self.batch_size))
|
|
if isinstance(self.inputs, tuple) or isinstance(self.inputs, list):
|
|
res_inputs = []
|
|
for inp in self.inputs:
|
|
if self.is_sparse:
|
|
res_inputs.append(
|
|
inp[batch_index, :].toarray())
|
|
else:
|
|
res_inputs.append(inp[batch_index, :])
|
|
else:
|
|
if self.is_sparse:
|
|
res_inputs = self.inputs[batch_index, :].toarray()
|
|
else:
|
|
res_inputs = self.inputs[batch_index, :]
|
|
self.start += self.batch_size
|
|
if self.has_targets:
|
|
if self.is_sparse:
|
|
labels = self.targets[batch_index, :].toarray()
|
|
else:
|
|
labels = self.targets[batch_index, :]
|
|
return (res_inputs, labels)
|
|
return res_inputs
|
|
else:
|
|
self.reset()
|
|
return self.next()
|
|
|
|
|
|
|
