Source code for pypath.inputs.pathwaycommons

#!/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/
#

from future.utils import iteritems

import collections

import pypath.share.curl as curl
import pypath.share.progress as progress
import pypath.resources.urls as urls
import pypath.share.common as common


PathwayCommonsInteraction = collections.namedtuple(
    'PathwayCommonsInteraction',
    [
        'id_a',
        'interaction_type',
        'id_b',
        'resource',
    ],
)


PathwayCommonsResource = collections.namedtuple(
    'PathwayCommonsResource',
    [
        'name',
        'pc_label',
        'version',
    ]
)
PathwayCommonsResource.__new__.__defaults__ = (12,)


pathwaycommons_resources = {
    PathwayCommonsResource('WikiPathways', 'wp', 11),
    PathwayCommonsResource('KEGG', 'kegg'),
    PathwayCommonsResource('BIND', 'bind'),
    PathwayCommonsResource('IntAct', 'intact'),
    PathwayCommonsResource('IntAct', 'intact_complex'),
    PathwayCommonsResource('PANTHER', 'panther'),
    PathwayCommonsResource('NCI-PID', 'pid'),
    PathwayCommonsResource('Reactome', 'reactome'),
    PathwayCommonsResource('DIP', 'dip'),
    PathwayCommonsResource('HPRD', 'hprd'),
    PathwayCommonsResource('INOH', 'inoh'),
    PathwayCommonsResource('NetPath', 'netpath'),
    PathwayCommonsResource('BioGRID', 'biogrid'),
    PathwayCommonsResource('CORUM', 'corum'),
    PathwayCommonsResource('PhosphoSite', 'psp'),
}


pathwaycommons_directed_types = {
    'state-change',
    'controls-state-change-of',
    'controls-transport-of',
    'controls-phosphorylation-of',
}



[docs]
def pathwaycommons_interactions(
        resources = None,
        types = None,
        by_interaction = False,
        version = 12,
    ):

    interactions = collections.defaultdict(set) if by_interaction else []

    types = common.to_set(types)

    resources = {
        res.lower()
        for res in (
            common.to_list(resources) or (
                pc_res.name
                for pc_res in pathwaycommons_resources
            )
        )
    }

    prg = progress.Progress(
        len(resources),
        'Processing PathwayCommons',
        1,
        percent = False,
    )

    url = urls.urls['pwcommons']['url']

    for resource in pathwaycommons_resources:

        if not resources & {resource.pc_label, resource.name.lower()}:

            continue

        prg.step()
        _version = min(resource.version, version)
        resource_url = url % (_version, _version, resource.pc_label)
        c = curl.Curl(resource_url, silent = False, large = True)

        for l in c.result:

            if hasattr(l, 'decode'):

                l = l.decode('ascii')

            l = l.strip('\n\r').split('\t')

            if not types or l[1] in types:

                if by_interaction:

                    a_b = (l[0], l[1], l[2])
                    b_a = (l[2], l[1], l[0])

                    directed = l[1] in pathwaycommons_directed_types

                    key = (
                        b_a
                            if (
                                a_b not in interactions and
                                not directed and
                                b_a in interactions
                            ) else
                        a_b
                    )

                    interactions[key].add(
                        PathwayCommonsInteraction(
                            *key,
                            resource = resource.name
                        )
                    )

                else:

                    l.append(resource.name)
                    interactions.append(PathwayCommonsInteraction(*l))

    return interactions



def _create_single_resource_method(resource):

    def _pc_single_resource(**kwargs):

        kwargs['resources'] = resource.name

        return pathwaycommons_interactions(**kwargs)


    return _pc_single_resource


for res in pathwaycommons_resources:

    method_name = 'pathwaycommons_%s_interactions' % res.name.lower()
    globals()[method_name] = _create_single_resource_method(res)