#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# This file is part of the `pypath` python module
#
# Copyright 2014-2023
# EMBL, EMBL-EBI, Uniklinik RWTH Aachen, Heidelberg University
#
# Authors: see the file `README.rst`
# Contact: Dénes Türei (turei.denes@gmail.com)
#
# Distributed under the GPLv3 License.
# See accompanying file LICENSE.txt or copy at
# https://www.gnu.org/licenses/gpl-3.0.html
#
# Website: https://pypath.omnipathdb.org/
#
"""
Old server class working from a ``pypath.legacy.main.PyPath`` object.
"""
import json
import pypath.omnipath.server.run as server
[docs]
class PypathServer(server.BaseServer):
[docs]
def __init__(self, pypath):
self.p = pypath
self.g = pypath.graph
self.isLeaf = True
server.BaseServer.__init__(self)
def network(self, req):
hdr = ['nodes', 'edges', 'is_directed', 'sources']
val = [
self.g.vcount(), self.g.ecount(), int(self.g.is_directed()),
self.p.sources
]
if b'format' in req.args and req.args[b'format'] == b'json':
return json.dumps(dict(zip(hdr, val)))
else:
return '%s\n%s' % ('\t'.join(hdr), '\t'.join(
[str(v) if type(v) is not list else ';'.join(v) for v in val]))
def interactions(self, req):
fields = [b'sources', b'references']
result = []
elist = self._get_eids(req)
res = []
hdr = [
'source', 'target', 'is_directed', 'is_stimulation',
'is_inhibition'
]
if b'fields' in req.args:
hdr += [
f.decode('utf-8')
for f in fields
if f in b','.join(req.args[b'fields']).split(b',')
]
if (
b'genesymbols' in req.args and
self._parse_arg(req.args[b'genesymbols'])
):
genesymbols = True
hdr.insert(2, 'source_genesymbol')
hdr.insert(3, 'target_genesymbol')
else:
genesymbols = False
all_sources = set([])
for eid in elist:
e = self.g.es[eid]
all_sources = all_sources | e['sources']
for d in ['straight', 'reverse']:
uniprots = getattr(e['dirs'], d)
if e['dirs'].dirs[uniprots]:
thisEdge = [
uniprots[0], uniprots[1]
]
if genesymbols:
thisEdge.extend([
self.g.vs[self.p.nodDct[uniprots[0]]]['label'],
self.g.vs[self.p.nodDct[uniprots[1]]]['label']
])
thisEdge.extend([
1,
int(e['dirs'].is_stimulation(uniprots)),
int(e['dirs'].is_inhibition(uniprots))
])
dsources = e['dirs'].get_dir(uniprots, sources=True)
dsources = dsources | e['dirs'].get_dir(
'undirected', sources=True)
if 'sources' in hdr:
thisEdge.append(list(dsources))
if 'references' in hdr:
thisEdge.append([
r.pmid
for r in flat_list([
rs for s, rs in iteritems(e['refs_by_source'])
if s in dsources
])
])
thisEdge.append(self._dip_urls(e))
res.append(thisEdge)
if not e['dirs'].is_directed():
thisEdge = [e['dirs'].nodes[0], e['dirs'].nodes[1]]
if genesymbols:
thisEdge.extend([
self.g.vs[self.p.nodDct[e['dirs'].nodes[0]]]['label'],
self.g.vs[self.p.nodDct[e['dirs'].nodes[1]]]['label']
])
thisEdge.extend([0, 0, 0])
if 'sources' in hdr:
thisEdge.append(list(e['sources']))
if 'references' in hdr:
thisEdge.append([r.pmid for r in e['references']])
thisEdge.append(self._dip_urls(e))
res.append(thisEdge)
if 'DIP' in all_sources:
hdr.append('dip_url')
else:
res = map(lambda r: r[:-1], res)
if b'format' in req.args and req.args[b'format'] == b'json':
return json.dumps([dict(zip(hdr, r)) for r in res])
else:
return self._table_output(res, hdr, req)
def _table_output(self, res, hdr, req):
return '%s%s' % ('' if not bool(req.args[b'header']) else
'%s\n' % '\t'.join(hdr), '\n'.join([
'\t'.join([
';'.join(f) if type(f) is list or
type(f) is set else str(f) for f in r
]) for r in res
]))
def _get_eids(self, req):
names = None if len(req.postpath) <= 1 else req.postpath[1].split(',')
ids = range(
0, self.g.vcount()) if names is None else self.p.names2vids(names)
ids = set(ids)
elist = set(range(0, self.g.ecount())) if names is None else set([])
if names is not None:
alist = dict(zip(ids, [self.g.neighbors(i) for i in ids]))
for one, others in iteritems(alist):
for two in others:
e = self.g.get_eid(one, two, directed=True, error=False)
if e != -1:
elist.add(e)
if self.g.is_directed():
e = self.g.get_eid(
two, one, directed=True, error=False)
if e != -1:
elist.add(e)
return elist
def ptms(self, req):
fields = [
b'is_stimulation', b'is_inhibition', b'sources', b'references'
]
result = []
elist = self._get_eids(req)
res = []
hdr = [
'enzyme', 'substrate', 'residue_type', 'residue_offset',
'modification'
]
if (
b'genesymbols' in req.args and
self._parse_arg(req.args[b'genesymbols'])
):
genesymbols = True
hdr.insert(2, 'enzyme_genesymbol')
hdr.insert(3, 'substrate_genesymbol')
else:
genesymbols = False
if b'fields' in req.args:
hdr += [
f.decode('utf-8')
for f in fields
if f in b','.join(req.args[b'fields']).split(b',')
]
if 'ptm' in self.g.es.attributes():
for eid in elist:
e = self.g.es[eid]
for ptm in e['ptm']:
if 'ptmtype' not in req.args or ptm.ptm.typ in req.args[
'ptmtype']:
thisPtm = [
ptm.domain.protein, ptm.ptm.protein
]
if genesymbols:
thisPtm.extend([
self.g.vs[self.p.nodDct[
ptm.domain.protein]]['label'],
self.g.vs[self.p.nodDct[
ptm.ptm.protein]]['label']
])
thisPtm.extend([
ptm.ptm.residue.name, ptm.ptm.residue.number,
ptm.ptm.typ
])
if 'is_stimulation' in hdr:
thisPtm.append(
int(e['dirs'].is_stimulation((
ptm.domain.protein, ptm.ptm.protein))))
if 'is_inhibition' in hdr:
thisPtm.append(
int(e['dirs'].is_inhibition((
ptm.domain.protein, ptm.ptm.protein))))
if 'sources' in hdr:
thisPtm.append(list(ptm.ptm.sources))
if 'references' in hdr:
thisPtm.append(list(ptm.refs))
res.append(thisPtm)
if b'format' in req.args and req.args[b'format'] == b'json':
return json.dumps([dict(zip(hdr, r)) for r in res])
else:
return self._table_output(res, hdr, req)
def resources(self, req):
hdr = [
'database', 'proteins', 'interactions', 'directions',
'stimulations', 'inhibitions', 'signs'
]
res = [[
# database name
s,
# number of proteins
len([1 for v in self.g.vs if s in v['sources']]),
# number of interacting pairs
len([1 for e in self.g.es if s in e['sources']]),
# number of directions
sum([s in e['dirs'].sources[d] \
for e in self.g.es for d in e['dirs'].which_dirs()]),
# number of stimulations
sum([s in e['dirs'].positive_sources[d] \
for e in self.g.es for d in e['dirs'].which_dirs()]),
# number of inhibitions
sum([s in e['dirs'].negative_sources[d] \
for e in self.g.es for d in e['dirs'].which_dirs()]),
# number of signs
sum([s in sg for e in self.g.es for d in e['dirs'].which_dirs() \
for sg in [e['dirs'].positive_sources[d],
e['dirs'].negative_sources[d]]]),
]
for s in self.p.sources]
if b'format' in req.args and req.args[b'format'] == b'json':
return json.dumps([dict(zip(hdr, r)) for r in res])
else:
return self._table_output(res, hdr, req)
def _dip_urls(self, e):
result = []
if 'dip_id' in e.attributes():
for dip_id in e['dip_id']:
try:
result.append(urls.urls['dip']['ik'] %
int(dip_id.split('-')[1][:-1]))
except:
print(dip_id)
return ';'.join(result)
