Source code for pypath.interaction

#!/usr/bin/env python
# -*- coding: utf-8 -*-

#
#  This file is part of the `pypath` python module
#
#  Copyright
#  2014-2020
#  EMBL, EMBL-EBI, Uniklinik RWTH Aachen, Heidelberg University
#
#  File author(s): Dénes Türei (turei.denes@gmail.com)
#                  Nicolàs Palacio
#                  Olga Ivanova
#
#  Distributed under the GPLv3 License.
#  See accompanying file LICENSE.txt or copy at
#      http://www.gnu.org/licenses/gpl-3.0.html
#
#  Website: http://pypath.omnipathdb.org/
#

"""
Here we define one class, the :py:class:`Interaction`` which provides a
rich API for representing and querying molecular interactions. The
interactions serve as the building elements of the network and the
:py:class:`pypath.network.Network` object largely relies on methods
of the :py:class:`Interaction`` objects.
"""

from future.utils import iteritems

import importlib as imp
import collections
import operator
import itertools
import functools

import pypath.evidence as pypath_evidence
import pypath.resource as pypath_resource
import pypath.session_mod as session_mod
import pypath.common as common
import pypath.mapping as mapping
import pypath.entity as entity

_logger = session_mod.Logger(name = 'interaction')
_log = _logger._log


InteractionKey = collections.namedtuple(
    'InteractionKey',
    [
        'entity_a',
        'entity_b',
    ],
)


InteractionDataFrameRecord = collections.namedtuple(
    'InteractionDataFrameRecord',
    [
        'id_a',
        'id_b',
        'type_a',
        'type_b',
        'directed',
        'effect',
        'type',
        'dmodel',
        'sources',
        'references',
    ],
)
InteractionDataFrameRecord.__new__.__defaults__ = (None,) * 8


[docs]class Interaction(object): """ Represents a unique pair of molecular entities interacting with each other. One :py:class:`Interaction` object might represent multiple interactions i.e. with different direction or effect or type (e.g. transcriptional regulation and post-translational regulation), each supported by different evidences. :arg str,pypath.entity.Entity a,b: The two interacting partners. If an :py:class:`pypath.entity.Entity` objects provided the other attributes (entity_type, id_type, taxon) will be ignored. :arg str id_type_a,id_type_b: The identifier types for partner ``a`` and ``b`` e.g. ``'uniprot'``. :arg str entity_type_a,entity_type_b: The types of the molecular entities ``a`` and ``b`` e.g. ``'protein'``. :arg int taxon_a,taxon_b: The NCBI Taxonomy Identifiers of partner ``a`` and ``b`` e.g. ``9606`` for human. :details: The arguments ``a`` and ``b`` will be assigned to the attribute ``a`` and ``b`` in an alphabetical order, hence it's possible that argument ``a`` becomes attribute ``b``. """ __slots__ = [ 'a', 'b', 'a_b', 'b_a', 'nodes', 'key', 'evidences', 'direction', 'positive', 'negative', 'attrs', ] _get_methods = { 'entities', 'evidences', 'references', 'curation_effort', 'resource_names', 'resources', 'data_models', 'interaction_types', 'interactions', 'interactions_0', 'interactions_directed', 'interactions_undirected', 'interactions_non_directed', 'interactions_undirected_0', 'interactions_non_directed_0', 'interactions_signed', 'interactions_positive', 'interactions_negative', 'interactions_mutual', 'resources_via', 'resource_names_via', } _get_methods_autogen = ( 'references', 'resources', 'resources_via', 'resource_names', 'resource_names_via', 'data_models', 'interaction_types', ) _by_methods = ( 'resource', 'reference', 'data_model', 'interaction_type', 'interaction_type_and_data_model', 'interaction_type_and_data_model_and_resource', ) _count_methods = { 'references', 'resources', 'resources_via', 'resource_names_via', 'resource_names', 'curation_effort', 'entities', 'proteins', 'complexes', 'mirnas', 'interactions', 'interactions_0', 'interactions_directed', 'interactions_signed', 'interactions_positive', 'interactions_negative', 'data_models', 'interaction_types', } _get_method_signature = [ ('direction', None), ('effect', None), ('resources', None), ('data_model', None), ('interaction_type', None), ('via', None), ('references', None), ] _degree_modes = ( 'ALL', 'IN', 'OUT', ) _degree_directions = { 'undirected': (None, None), 'non_directed': (False, None), 'directed': (True, None), 'signed': (True, True), 'positive': (True, 'positive'), 'negative': (True, 'negative'), } _entity_types = { 'protein', ('complex', 'complexes'), 'mirna', 'small_molecule', None, } _entity_values = { 'identifiers', 'labels', None, } def __init__( self, a, b, id_type_a = 'uniprot', id_type_b = 'uniprot', entity_type_a = 'protein', entity_type_b = 'protein', taxon_a = 9606, taxon_b = 9606, ): a = self._get_entity( identifier = a, id_type = id_type_a, entity_type = entity_type_a, taxon = taxon_a, ) b = self._get_entity( identifier = b, id_type = id_type_b, entity_type = entity_type_b, taxon = taxon_b, ) self.nodes = tuple(sorted((a, b))) self.a = self.nodes[0] self.b = self.nodes[1] self.key = self._key self.a_b = (self.nodes[0], self.nodes[1]) self.b_a = (self.nodes[1], self.nodes[0]) self.evidences = pypath_evidence.Evidences() self.direction = { self.a_b: pypath_evidence.Evidences(), self.b_a: pypath_evidence.Evidences(), 'undirected': pypath_evidence.Evidences(), } self.positive = { self.a_b: pypath_evidence.Evidences(), self.b_a: pypath_evidence.Evidences(), } self.negative = { self.a_b: pypath_evidence.Evidences(), self.b_a: pypath_evidence.Evidences(), } self.attrs = {}
[docs] def reload(self): """ Reloads the object from the module level. """ modname = self.__class__.__module__ evmodname = self.evidences.__class__.__module__ enmodname = self.a.__class__.__module__ mod = __import__(modname, fromlist = [modname.split('.')[0]]) evmod = __import__(evmodname, fromlist = [evmodname.split('.')[0]]) enmod = __import__(enmodname, fromlist = [enmodname.split('.')[0]]) imp.reload(mod) imp.reload(evmod) imp.reload(enmod) new = getattr(mod, self.__class__.__name__) evsnew = getattr(evmod, 'Evidences') evnew = getattr(evmod, 'Evidence') ennew = getattr(enmod, 'Entity') setattr(self, '__class__', new) for evs in itertools.chain( (self.evidences,), self.direction.values(), self.positive.values(), self.negative.values(), ): evs.__class__ = evsnew for ev in evs: ev.__class__ = evnew self.a.__class__ = ennew self.b.__class__ = ennew self._generate_get_methods() self._generate_count_methods() self._generate_by_methods()
def _get_entity( self, identifier, id_type = 'uniprot', entity_type = 'protein', taxon = 9606, ): if not isinstance(identifier, entity.Entity): identifier = entity.Entity( identifier = identifier, id_type = id_type, entity_type = entity_type, taxon = taxon, ) return identifier def _check_nodes_key(self, nodes): """Checks if *nodes* is contained in the edge. :arg list nodes: Or [tuple], contains the names of the nodes to be checked. :return: (*bool*) -- ``True`` if all elements in *nodes* are contained in the object :py:attr:`nodes` list. """ return nodes == self.a_b or nodes == self.b_a def _check_direction_key(self, direction): """ Checks if *direction* is ``'undirected'`` or contains the nodes of the current edge. Used internally to check that *di* is a valid key for the object attributes declared on dictionaries. :arg tuple di: Or [str], key to be tested for validity. :return: (*bool*) -- ``True`` if *di* is ``'undirected'`` or a tuple of node names contained in the edge, ``False`` otherwise. """ return ( direction == 'undirected' or ( isinstance(direction, tuple) and self._check_nodes_key(direction) ) ) def id_to_entity(self, identifier): return ( self.a if self.a == identifier else self.b if self.b == identifier else None ) def direction_key(self, direction): if direction == 'undirected': return direction direction = tuple(map(self.id_to_entity, direction)) return ( direction if direction == self.a_b or direction == self.b_a else None ) @staticmethod def direction_key_identifiers(direction): if direction == 'undirected': return direction return tuple(ent.identifier for ent in direction) @staticmethod def _directed_key(direction): return direction is not None and direction != 'undirected'
[docs] def add_evidence( self, evidence, direction = 'undirected', effect = 0, references = None, ): """ Adds directionality information with the corresponding data source named. Modifies self attributes :py:attr:`dirs` and :py:attr:`sources`. :arg resource.NetworkResource,evidence.Evidence evidence: Either a ``pypath.evidence.Evidence`` object or a resource as ``pypath.resource.NetworkResource`` object. In the latter case the references can be provided in a separate argument. :arg tuple direction: Or [str], the directionality key for which the value on :py:attr:`dirs` has to be set ``True``. :arg int effect: The causal effect of the interaction. 1 or 'stimulation' corresponds to a stimulatory, -1 or 'inhibition' to an inhibitory while 0 to an unknown or neutral effect. :arg set,NoneType references: A set of references, used only if the resource have been provided as ``NetworkResource`` object. """ direction = self.direction_key(direction) if direction is None: _log( 'Attempting to add evidence with non matching ' 'interaction partners.' ) return evidence = ( evidence if isinstance( evidence, ( pypath_evidence.Evidence, pypath_evidence.Evidences, ) ) else pypath_evidence.Evidence( resource = evidence, references = references, ) ) self.evidences += evidence self.direction[direction] += evidence if direction != 'undirected': if effect in {1, 'positive', 'stimulation'}: self.positive[direction] += evidence elif effect in {-1, 'negative', 'inhibition'}: self.negative[direction] += evidence
def __hash__(self): return hash(self.key) def __eq__(self, other): return self.key == other.key @property def _key(self): return InteractionKey( self.a.key, self.b.key, ) def __iadd__(self, other): if self != other: _log( 'Attempt to merge interactions with ' 'non matching interaction partners.' ) return self self._merge_evidences(self, other) self.update_attrs(**other.attrs) return self def update_attrs(self, **kwargs): for key, val in iteritems(kwargs): if key in self.attrs: self.attrs[key] = common.combine_attrs((self.attrs[key], val)) else: self.attrs[key] = val def __add__(self, other): new = self.__copy__() new += other return new def __copy__(self): new = Interaction(*self.key) new += self return new @staticmethod def _merge_evidences(one, other): one.evidences += other.evidences for dir_key in one.direction.keys(): one.direction[dir_key] += other.direction[dir_key] for eff_key in one.positive.keys(): one.positive[eff_key] += other.positive[eff_key] for eff_key in one.negative.keys(): one.negative[eff_key] += other.negative[eff_key] def __repr__(self): return '<Interaction: %s %s=%s=%s=%s %s [%s]>' % ( self.a.label or self.a.identifier, '<' if self.direction[self.b_a] else '=', ( '(+-)' if ( self.positive[self.b_a] and self.negative[self.b_a] ) else '(+)=' if self.positive[self.b_a] else '(-)=' if self.negative[self.b_a] else '====' ), ( '(+-)' if ( self.positive[self.a_b] and self.negative[self.a_b] ) else '(+)=' if self.positive[self.a_b] else '(-)=' if self.negative[self.a_b] else '====' ), '>' if self.direction[self.a_b] else '=', self.b.label or self.b.identifier, self.evidences.__repr__().strip('<>'), ) def __contains__(self, other): return ( other == self.a or other == self.b if isinstance(other, entity.Entity) else self.evidences.__contains__(other) ) def has_data_model(self, data_model): return self.evidences.has_data_model(data_model) @property def data_models(self): return { ev.resource.data_model for ev in self.evidences }
[docs] def get_direction( self, direction, resources = False, evidences = False, sources = False, resource_names = False, ): """ Returns the state (or *resources* if specified) of the given *direction*. :arg tuple direction: Or [str] (if ``'undirected'``). Pair of nodes from which direction information is to be retrieved. :arg bool resources: Optional, ``'False'`` by default. Specifies if the :py:attr:`resources` information of the given direction is to be retrieved instead. :return: (*bool* or *set*) -- (if ``resources=True``). Presence/absence of the requested direction (or the list of resources if specified). Returns ``None`` if *direction* is not valid. """ direction = self.direction_key(direction) if direction is not None: return self._select_answer_type( self.direction[direction], resources = resources, evidences = evidences, resource_names = resource_names, sources = sources, ) else: return None
[docs] def get_directions( self, src, tgt, resources = False, evidences = False, resource_names = False, sources = False, ): """ Returns all directions with boolean values or list of sources. :arg str src: Source node. :arg str tgt: Target node. :arg bool resources: Optional, ``False`` by default. Specifies whether to return the :py:attr:`resources` attribute instead of :py:attr:`dirs`. :return: Contains the :py:attr:`dirs` (or :py:attr:`resources` if specified) of the given edge. """ query = (src, tgt) answer_type_args = { 'resources': resources, 'evidences': evidences, 'resource_names': resource_names, 'sources': sources, } query = self.direction_key(query) if query is not None: return [ self._select_answer_type( self.direction[query], **answer_type_args ), self._select_answer_type( self.direction[tuple(reversed(query))], **answer_type_args ), self._select_answer_type( self.direction['undirected'], **answer_type_args ), ] else: return None
def _select_answer_type( self, answer, resources = False, evidences = False, resource_names = False, sources = False, ): return ( answer if evidences else answer.get_resources() if resources else answer.get_resource_names() if sources or resource_names else bool(answer) )
[docs] def which_directions( self, resources = None, effect = None, ): """ Returns the pair(s) of nodes for which there is information about their directionality. :arg str effect: Either *positive* or *negative*. :arg str,set resources: Limits the query to one or more resources. Optional. :return: (*tuple*) -- Tuple of tuples with pairs of nodes where the first element is the source and the second is the target entity, according to the given resources and limited to the effect. """ resources = self._resources_set(resources) effect = self._effect_synonyms(effect) return tuple( _dir for _dir, _evidences in iteritems(self.direction) if _dir != 'undirected' and _evidences and ( not resources or _evidences & resources ) and ( not effect or ( not resources and getattr(self, effect)[_dir] ) or getattr(self, effect)[_dir] & resources ) )
# synonym: old name which_dirs = which_directions
[docs] def which_signs(self, resources = None, effect = None): """ Returns the pair(s) of nodes for which there is information about their effect signs. :param str,set resources: Limits the query to one or more resources. Optional. :param str effect: Either *positive* or *negative*, limiting the query to positive or negative effects; for any other values effects of both signs will be returned. :return: (*tuple*) -- Tuple of tuples with pairs of nodes where the first element is a tuple of the source and the target entity, while the second element is the effect sign, according to the given resources. E.g. ((('A', 'B'), 'positive'),) """ resources = self._resources_set(resources) effect = self._effect_synonyms(effect) effects = (effect,) if effect else ('positive', 'negative') return tuple( (_dir, _effect) for _effect in effects for _dir, _evidences in iteritems(getattr(self, _effect)) if _evidences and ( not resources or _evidences & resources ) )
@staticmethod def _effect_synonyms(effect): if not effect or effect == True: return effect if effect in {'positive', 'stimulation', 'stimulatory'}: return 'positive' if effect in {'negative', 'inhibition', 'inhibitory'}: return 'negative' def _resources_set(self, resources = None): return common.to_set(resources)
[docs] def unset_direction( self, direction, only_sign = False, resource = None, interaction_type = None, via = False, source = None, ): """ Removes directionality and/or source information of the specified *direction*. Modifies attribute :py:attr:`dirs` and :py:attr:`sources`. :arg tuple direction: Or [str] (if ``'undirected'``) the pair of nodes specifying the directionality from which the information is to be removed. :arg set resource: Optional, ``None`` by default. If specified, determines which specific source(s) is(are) to be removed from :py:attr:`sources` attribute in the specified *direction*. """ direction = self.direction_key(direction) if direction is not None: attrs = ( (self._effect_synonyms(only_sign),) if only_sign else ('direction', 'positive', 'negative') ) resource = resource or source for attr in attrs: if resource is not None: getattr(self, attr)[direction].remove( resource = resource, interaction_type = interaction_type, via = via, ) else: getattr(self, attr)[direction] = ( pypath_evidence.Evidences() )
# synonym: old name unset_dir = unset_direction
[docs] def unset_sign( self, direction, sign, resource = None, interaction_type = None, via = False, source = None, ): """ Removes sign and/or source information of the specified *direction* and *sign*. Modifies attribute :py:attr:`positive` and :py:attr:`positive_sources` or :py:attr:`negative` and :py:attr:`negative_sources` (or :py:attr:`positive_attributes`/:py:attr:`negative_sources` only if ``source=True``). :arg tuple direction: The pair of nodes specifying the directionality from which the information is to be removed. :arg str sign: Sign from which the information is to be removed. Must be either ``'positive'`` or ``'negative'``. :arg set source: Optional, ``None`` by default. If specified, determines which source(s) is(are) to be removed from the sources in the specified *direction* and *sign*. """ self.unset_direction( direction = direction, only_sign = sign, resource = resource, interaction_type = interaction_type, via = via, source = source, )
[docs] def unset_interaction_type(self, interaction_type): """ Removes all evidences with a certain ``interaction_type``. """ for ev in tuple(self.evidences): if ev.resource.interaction_type == interaction_type: self.evidences -= ev for attr in ('direction', 'positive', 'negative'): for key, evs in getattr(self, attr): for ev in tuple(evs): if ev.resource.interaction_type == interaction_type: evs -= ev
[docs] def is_directed(self): """ Checks if edge has any directionality information. :return: (*bool*) -- Returns ``True`` if any of the :py:attr:`dirs` attribute values is ``True`` (except ``'undirected'``), ``False`` otherwise. """ return any(self.direction.values())
[docs] def is_directed_by_resources(self, resources = None): """ Checks if edge has any directionality information from some resource(s). :return: (*bool*) -- Returns ``True`` if any of the :py:attr:`dirs` attribute values is ``True`` (except ``'undirected'``), ``False`` otherwise. """ return self._by_resource(resources, op = operator.or_)
def is_mutual(self, resources = None): """ Checks if the edge has mutual directions (both A-->B and B-->A). """ return ( bool(self.direction[self.a_b]) and bool(self.direction[self.b_a]) if not resources else self.is_mutual_by_resources(resources = resources) )
[docs] def is_mutual_by_resources(self, resources = None): """ Checks if the edge has mutual directions (both A-->B and B-->A) according to some resource(s). """ return self._by_resource(resources, op = operator.and_)
[docs] def is_loop(self): """ :returns: ``True`` if the interaction is a loop edge i.e. its endpoints are the same node. """ return self.a == self.b
def _by_resource(self, resources = None, op = operator.or_): resources = self._resources_set(resources) return op( self.direction[self.a_b] & resources, self.direction[self.b_a] & resources, )
[docs] def is_stimulation(self, direction = None, resources = None): """ Checks if any (or for a specific *direction*) interaction is activation (positive interaction). :arg tuple direction: Optional, ``None`` by default. If specified, checks the :py:attr:`positive` attribute of that specific directionality. If not specified, checks both. :return: (*bool*) -- ``True`` if any interaction (or the specified *direction*) is activatory (positive). """ return self._is_effect( sign = 'positive', direction = direction, resources = resources, )
[docs] def is_inhibition(self, direction = None, resources = None): """ Checks if any (or for a specific *direction*) interaction is inhibition (negative interaction). :arg tuple direction: Optional, ``None`` by default. If specified, checks the :py:attr:`negative` attribute of that specific directionality. If not specified, checks both. :return: (*bool*) -- ``True`` if any interaction (or the specified *direction*) is inhibitory (negative). """ return self._is_effect( sign = 'negative', direction = direction, resources = resources, )
def _is_effect(self, sign, direction = None, resources = None): _sign = getattr(self, sign) _resources = self._resources_set(resources) return ( any( bool( _evidences if not _resources else _evidences & _resources ) for _direction, _evidences in iteritems(_sign) if not direction or direction == _direction ) )
[docs] def has_sign(self, direction = None, resources = None): """ Checks whether the edge (or for a specific *direction*) has any signed information (about positive/negative interactions). :arg tuple direction: Optional, ``None`` by default. If specified, only the information of that direction is checked for sign. :return: (*bool*) -- ``True`` if there exist any information on the sign of the interaction, ``False`` otherwise. """ return ( self.is_stimulation(direction = direction, resources = resources) or self.is_inhibition(direction = direction, resources = resources) )
[docs] def add_sign( self, direction, sign, resource = None, resource_name = None, interaction_type = 'PPI', data_model = None, **kwargs ): """ Sets sign and source information on a given direction of the edge. Modifies the attributes :py:attr:`positive` and :py:attr:`positive_sources` or :py:attr:`negative` and :py:attr:`negative_sources` depending on the sign. Direction is also updated accordingly, which also modifies the attributes :py:attr:`dirs` and :py:attr:`sources`. :arg tuple direction: Pair of edge nodes specifying the direction from which the information is to be set/updated. :arg str sign: Specifies the type of interaction. Either ``'positive'`` or ``'negative'``. :arg set resource: Contains the name(s) of the source(s) from which the information was obtained. :arg **kwargs: Passed to ``pypath.resource.NetworkResource`` if ``resource`` is not already a ``NetworkResource`` or ``Evidence`` instance. """ sign = self._effect_synonyms(sign) evidence = ( resource if isinstance(resource, pypath_evidence.Evidence) else pypath_evidence.Evidence( resource = resource, references = references, ) if isinstance(resource, pypath_resource.NetworkResource) else pypath_evidence.Evidence( resource = pypath_resource.NetworkResource( name = resource_name, interaction_type = interaction_type, data_model = data_model, **kwargs, ) ) if resource_name is not None else None ) direction = self.direction_key(direction) if self._directed_key(direction) and evidence is not None: ev_attr = getattr(self, sign) ev_attr += evidence
[docs] def get_sign( self, direction, sign = None, evidences = False, resources = False, resource_names = False, sources = False, ): """ Retrieves the sign information of the edge in the given diretion. If specified in *sign*, only that sign's information will be retrieved. If specified in *sources*, the sources of that information will be retrieved instead. :arg tuple direction: Contains the pair of nodes specifying the directionality of the edge from which th information is to be retrieved. :arg str sign: Optional, ``None`` by default. Denotes whether to retrieve the ``'positive'`` or ``'negative'`` specific information. :arg bool resources: Optional, ``False`` by default. Specifies whether to return the resources instead of sign. :return: (*list*) -- If ``sign=None`` containing [bool] values denoting the presence of positive and negative sign on that direction, if ``sources=True`` the [set] of sources for each of them will be returned instead. If *sign* is specified, returns [bool] or [set] (if ``sources=True``) of that specific direction and sign. """ sign = self._effect_synonyms(sign) answer_type_args = { 'resources': resources, 'evidences': evidences, 'resource_names': resource_names, 'sources': sources, } direction = self.direction_key(direction) if self._directed_key(direction): return ( self._select_answer_type( getattr(self, sign)[direction], **answer_type_args ) if sign else [ self._select_answer_type( self.positive[direction], **answer_type_args ), self._select_answer_type( self.negative[direction], **answer_type_args ) ] )
[docs] def source( self, undirected = False, resources = None, **kwargs ): """ Returns the name(s) of the source node(s) for each existing direction on the interaction. :arg bool undirected: Optional, ``False`` by default. :returns: (*list*) -- Contains the name(s) for the source node(s). This means if the interaction is bidirectional, the list will contain both identifiers on the edge. If the interaction is undirected, an empty list will be returned. """ return self._partner( source_target = 'source', undirected = undirected, resources = resources, **kwargs )
# synonym: old name src = source
[docs] def target( self, undirected = False, resources = None, **kwargs ): """ Returns the name(s) of the target node(s) for each existing direction on the interaction. :arg bool undirected: Optional, ``False`` by default. :returns: (*list*) -- Contains the name(s) for the target node(s). This means if the interaction is bidirectional, the list will contain both identifiers on the edge. If the interaction is undirected, an empty list will be returned. """ return self._partner( source_target = 'target', undirected = undirected, resources = resources, **kwargs )
# synonym: old name tgt = target def _partner( self, source_target, undirected = False, resources = None, **kwargs ): resources = self._resources_set(resources) _slice = slice(0, 1) if source_target == 'source' else slice(1, 2) return tuple(itertools.chain( ( _direction[_slice] if _direction != 'undirected' else self.nodes if undirected else () ) for _direction, _evidences in iteritems(self.direction) if ( ( ( not resources and not kwargs and bool(_evidences) ) or ( any( ev.match( resource = res, **kwargs ) for res in resources or (None,) for ev in _evidences ) ) ) ) ))
[docs] def src_by_resource(self, resource): """ Returns the name(s) of the source node(s) for each existing direction on the interaction for a specific *resource*. :arg str resource: Name of the resource according to which the information is to be retrieved. :return: (*list*) -- Contains the name(s) for the source node(s) according to the specified *resource*. This means if the interaction is bidirectional, the list will contain both identifiers on the edge. If the specified *source* is not found or invalid, an empty list will be returned. """ return [ _dir[0] for _dir, _evidences in iteritems(self.direction) if ( _dir != 'undirected' and resource in _evidences ) ]
[docs] def tgt_by_resource(self, resource): """ Returns the name(s) of the target node(s) for each existing direction on the interaction for a specific *resource*. :arg str resource: Name of the resource according to which the information is to be retrieved. :return: (*list*) -- Contains the name(s) for the target node(s) according to the specified *resource*. This means if the interaction is bidirectional, the list will contain both identifiers on the edge. If the specified *source* is not found or invalid, an empty list will be returned. """ return [ _dir[1] for _dir, _evidences in iteritems(self.direction) if ( _dir != 'undirected' and resource in _evidences ) ]
[docs] def resources_a_b( self, resources = False, evidences = False, resource_names = False, sources = False, ): """ Retrieves the list of resources for the :py:attr:`a_b` direction. :return: (*set*) -- Contains the names of the sources supporting the :py:attr:`a_b` directionality of the edge. """ answer_type_args = { 'resources': resources, 'evidences': evidences, 'resource_names': resource_names, 'sources': sources, } return self._select_answer_type( self.direction[self.a_b], **answer_type_args )
# synonym for old method name sources_straight = resources_a_b
[docs] def resources_b_a( self, resources = False, evidences = False, resource_names = False, sources = False, ): """ Retrieves the list of sources for the :py:attr:`b_a` direction. :return: (*set*) -- Contains the names of the sources supporting the :py:attr:`b_a` directionality of the edge. """ answer_type_args = { 'resources': resources, 'evidences': evidences, 'resource_names': resource_names, 'sources': sources, } return self._select_answer_type( self.direction[self.b_a], **answer_type_args )
# synonym for old method name sources_reverse = resources_b_a
[docs] def resources_undirected( self, resources = False, evidences = False, resource_names = False, sources = False, ): """ Retrieves the list of resources without directed information. :return: (*set*) -- Contains the names of the sources supporting the edge presence but without specific directionality information. """ answer_type_args = { 'resources': resources, 'evidences': evidences, 'resource_names': resource_names, 'sources': sources, } return self._select_answer_type( self.direction['undirected'], **answer_type_args )
sources_undirected = resources_undirected
[docs] def positive_a_b(self): """ Checks if the :py:attr:`a_b` directionality is a positive interaction. :return: (*bool*) -- ``True`` if there is supporting information on the :py:attr:`a_b` direction of the edge as activation. ``False`` otherwise. """ return bool(self.positive[self.a_b])
positive_straight = positive_a_b
[docs] def positive_b_a(self): """ Checks if the :py:attr:`b_a` directionality is a positive interaction. :return: (*bool*) -- ``True`` if there is supporting information on the :py:attr:`b_a` direction of the edge as activation. ``False`` otherwise. """ return bool(self.positive[self.b_a])
positive_reverse = positive_b_a
[docs] def negative_a_b(self): """ Checks if the :py:attr:`a_b` directionality is a negative interaction. :return: (*bool*) -- ``True`` if there is supporting information on the :py:attr:`a_b` direction of the edge as inhibition. ``False`` otherwise. """ return bool(self.negative[self.a_b])
negative_straight = negative_a_b
[docs] def negative_b_a(self): """ Checks if the :py:attr:`b_a` directionality is a negative interaction. :return: (*bool*) -- ``True`` if there is supporting information on the :py:attr:`b_a` direction of the edge as inhibition. ``False`` otherwise. """ return bool(self.negative[self.b_a])
negative_reverse = negative_b_a
[docs] def negative_resources_a_b(self, **kwargs): """ Retrieves the list of resources for the :py:attr:`a_b` direction and negative sign. :return: (*set*) -- Contains the names of the resources supporting the :py:attr:`a_b` directionality of the edge with a negative sign. """ answer_type_args = { 'resource_names': True } answer_type_args.update(kwargs) return self._select_answer_type( self.negative[self.a_b], **answer_type_args )
[docs] def negative_resources_b_a(self, **kwargs): """ Retrieves the list of resources for the :py:attr:`b_a` direction and negative sign. :return: (*set*) -- Contains the names of the resources supporting the :py:attr:`b_a` directionality of the edge with a negative sign. """ answer_type_args = { 'resource_names': True } answer_type_args.update(kwargs) return self._select_answer_type( self.negative[self.b_a], **answer_type_args )
[docs] def positive_resources_a_b(self, **kwargs): """ Retrieves the list of resources for the :py:attr:`a_b` direction and positive sign. :return: (*set*) -- Contains the names of the resources supporting the :py:attr:`a_b` directionality of the edge with a positive sign. """ answer_type_args = { 'resource_names': True } answer_type_args.update(kwargs) return self._select_answer_type( self.positive[self.a_b], **answer_type_args )
[docs] def positive_resources_b_a(self, **kwargs): """ Retrieves the list of resources for the :py:attr:`b_a` direction and positive sign. :return: (*set*) -- Contains the names of the resources supporting the :py:attr:`b_a` directionality of the edge with a positive sign. """ answer_type_args = { 'resource_names': True } answer_type_args.update(kwargs) return self._select_answer_type( self.positive[self.b_a], **answer_type_args )
[docs] def majority_dir( self, only_interaction_type = None, only_primary = False, by_references = False, by_reference_resource_pairs = True, ): """ Infers which is the major directionality of the edge by number of supporting sources. :return: (*tuple*) -- Contains the pair of nodes denoting the consensus directionality. If the number of sources on both directions is equal, ``None`` is returned. If there is no directionality information, ``'undirected'``` will be returned. """ a_b = self.direction[self.a_b] b_a = self.direction[self.b_a] if not a_b and not b_a: return 'undirected' method = ( 'count_references' if by_references else 'count_curation_effort' if by_reference_resource_pairs else 'count_resources' ) n_a_b = getattr(a_b, method)( interaction_type = only_interaction_type, via = False if only_primary else None, ) n_b_a = getattr(b_a, method)( interaction_type = only_interaction_type, via = False if only_primary else None, ) return ( 'undirected' if n_a_b == 0 and n_b_a == 0 else None if n_a_b == n_b_a else self.a_b if n_a_b > n_b_a else self.b_a )
[docs] def majority_sign( self, only_interaction_type = None, only_primary = False, by_references = False, by_reference_resource_pairs = True, ): """ Infers which is the major sign (activation/inhibition) of the edge by number of supporting sources on both directions. :return: (*dict*) -- Keys are the node tuples on both directions (:py:attr:`straight`/:py:attr:`reverse`) and values can be either ``None`` if that direction has no sign information or a list of two [bool] elements corresponding to majority of positive and majority of negative support. In case both elements of the list are ``True``, this means the number of supporting sources for both signs in that direction is equal. """ result = {} method = ( 'count_references' if by_references else 'count_curation_effort' if by_reference_resource_pairs else 'count_resources' ) for _dir in (self.a_b, self.b_a): n_pos = getattr(self.positive[_dir], method)( interaction_type = only_interaction_type, via = False if only_primary else None, ) n_neg = getattr(self.negative[_dir], method)( interaction_type = only_interaction_type, via = False if only_primary else None, ) result[_dir] = [ 0 < n_pos >= n_neg, 0 < n_neg >= n_pos, ] return result
[docs] def consensus( self, only_interaction_type = None, only_primary = False, by_references = False, by_reference_resource_pairs = True, ): """ Infers the consensus edge(s) according to the number of supporting sources. This includes direction and sign. :return: (*list*) -- Contains the consensus edge(s) along with the consensus sign. If there is no major directionality, both are returned. The structure is as follows: ``['<source>', '<target>', '<(un)directed>', '<sign>']`` """ result = [] _dir = self.majority_dir( only_interaction_type = only_interaction_type, only_primary = only_primary, by_references = by_references, by_reference_resource_pairs = by_reference_resource_pairs, ) _effect = self.majority_sign( only_interaction_type = only_interaction_type, only_primary = only_primary, by_references = by_references, by_reference_resource_pairs = by_reference_resource_pairs, ) if _dir == 'undirected': result.append([ self.a_b[0], self.a_b[1], 'undirected', 'unknown', ]) else: dirs = (self.a_b, self.b_a) if _dir is None else (_dir,) for d in dirs: if _effect[d] is not None: # index #0 is positive if _effect[d][0]: result.append([ d[0], d[1], 'directed', 'positive', ]) # can not be elif bc of the case of equal weight of # evidences for both positive and negative if _effect[d][1]: result.append([ d[0], d[1], 'directed', 'negative', ]) # directed with unknown effect else: result.append([ d[0], d[1], 'directed', 'unknown', ]) return result
consensus_edges = consensus
[docs] def merge(self, other): """ Merges current Interaction with another (if and only if they are the same class and contain the same nodes). Updates the attributes :py:attr:`direction`, :py:attr:`positive` and :py:attr:`negative`. :arg pypath.interaction.Interaction other: The new Interaction object to be merged with the current one. """ if not self._check_nodes_key(other.nodes): _log( 'Attempting to merge Interaction instances with different ' 'interacting partners.' ) return self.evidences += other.evidences for attr, _dir in itertools.product( ('direction', 'positive', 'negative'), (self.a_b, self.b_a, 'undirected') ): if attr != 'direction' and _dir == 'undirected': continue getattr(self, attr)[_dir] += getattr(other, attr)[_dir]
[docs] def translate(self, ids, new_attrs = None): """ Translates the node names/identifiers according to the dictionary *ids*. Also is able to change attributes like `id_type`, `taxon` and `entity_type`. :arg dict ids: Dictionary containing (at least) the current names of the nodes as keys and their translation as values. :arg dict new_attrs: Dictionary with new IDs as keys and their dicts of their new attributes as values. For any attribute not provided here the attributes from the original instance will be used. E.g. you can provide `{'1956': {'id_type': 'entrez'}}' if the new ID type for protein EGFR is Entrez Gene ID. :return: (*pypath.main.Direction*) -- The copy of current edge object with translated node names. """ new_a = ids[self.nodes[0]] new_b = ids[self.nodes[1]] new_ids = {'a': new_a, 'b': new_b} to_old = common.swap_dict_simple(ids) all_new_attrs = dict( ( '%s_%s' % (attr, label), new_attrs[new_ids[label]][attr] if ( new_ids[label] in new_attrs and attr in new_attrs[new_ids[label]] ) else getattr(self, '%s_%s' % (attr, label)) ) for attr in ('id_type', 'entity_type', 'taxon') for label in ('a', 'b') ) new = Interaction( id_a = new_a, id_b = new_b, **all_new_attrs ) new.evidences += self.evidences for (old_dir, new_dir), attr in itertools.product( zip( ( (to_old[new.id_a], to_old[new.id_b]), (to_old[new.id_b], to_old[new.id_a]), 'undirected' ), ( new.a_b, new.b_a, 'undirected', ), ), ('direction', 'positive', 'negative'), ): if _dir == 'undirected' and attr != 'direction': continue getattr(self, attr)[new_dir] += getattr(self, attr)[old_dir] return new
def get_evidences( self, direction = None, effect = None, resources = None, data_model = None, interaction_type = None, via = None, references = None, ): effect = self._effect_synonyms(effect) evidences = ( # any signed sum(itertools.chain( self.positive.values(), self.negative.values(), )) if effect == True else # only positive ( self.positive[direction] if direction in self.positive else sum(self.positive.values()) ) if effect == 'positive' else # only negative ( self.negative[direction] if direction in self.negative else sum(self.negative.values()) ) if effect == 'negative' else # any directed sum(self.direction[_dir] for _dir in self.which_dirs()) if direction == True else # one specific direction self.direction[direction] if direction in self.direction else # all evidences (default) self.evidences ) return ( pypath_evidence.Evidences( evidences.filter( resource = resources, interaction_type = interaction_type, via = via, data_model = data_model, references = references, ) ) )
[docs] def get_entities( self, entity_type = None, direction = None, effect = None, resources = None, data_model = None, interaction_type = None, via = None, references = None, return_type = None, ): """ Retrieves the entities involved in interactions matching the criteria. It either returns both interacting entities in a *set* or an empty *set*. This may not sound so useful at the level of this object but becomes more useful once we want to collect entities having certain kind of interactions across a series of `Interaction` objects. :arg str entity_type: The type of the molecular entity. Possible values: `protein`, `complex`, `mirna`, `small_molecule`. :arg str return_type: The type of values to return. Default is py:class:``pypath.entity.Entity`` objects, alternatives are ``labels`` ``identifiers``. """ # TODO: this method could be made slightly more efficient by using # not ``get_interactions`` but a simpler logic as here we don't need # to handle the directions separately; however this is not very # important and for the time being it's good as it is. kwargs = locals() _ = kwargs.pop('self') entity_type = common.to_set(kwargs.pop('entity_type')) return_type = kwargs.pop('return_type') return_types = { 'entity': None, 'entities': None, 'id': 'identifier', 'name': 'identifier', } # allow plurals return_type = ( return_type[:-1] if ( isinstance(return_type, common.basestring) and return_type[-1] == 's' ) else return_type ) # allow some synonyms return_type = ( return_types[return_type] if return_type in return_types else return_type ) return ( set( ( getattr(en, return_type) if return_type and hasattr(en, return_type) else en ) for en in itertools.chain( *self.get_interactions(**kwargs) ) if not entity_type or en.entity_type in entity_type ) )
@classmethod def _generate_entity_methods(cls): def _create_entity_method(entity_type, return_type): def _entity_method(*args, **kwargs): self = args[0] kwargs['entity_type'] = entity_type kwargs['return_type'] = return_type return self.get_entities(*args[1:], **kwargs) return _entity_method for etype, vtype in itertools.product( cls._entity_types, cls._entity_values, ): if etype is None and vtype is None: continue entity_type = etype[0] if isinstance(etype, tuple) else etype return_type = vtype etype_part = ( '' if not entity_type else entity_type if vtype else etype[1] if isinstance(etype, tuple) else '%ss' % entity_type ) vtype_part = '%s' % vtype if vtype else '' _method_name = '%s%s%s' % ( etype_part, '_' if etype_part and vtype_part else '', vtype_part, ) method_name = 'get_%s' % _method_name method = _create_entity_method( entity_type = entity_type, return_type = return_type, ) cls._add_method( method_name, method, signature = ( ['self', ('entity_type', None)] + cls._get_method_signature + [('return_type', None)] ), doc = cls.get_entities.__doc__, ) cls._get_methods.add(_method_name) cls._count_methods.add(_method_name)
[docs] def get_interactions( self, direction = None, effect = None, resources = None, data_model = None, interaction_type = None, via = None, references = None, ): """ Returns one or two tuples of the interacting partners: one if only one direction, two if both directions match the query criteria. The tuple will be empty if no evidence matches the criteria. :arg NontType,bool,tuple direction: If `None` both undirected and directed, if `True` only directed, if a *tuple* of entities only the interactions with that specific direction will be considered. Unless you set this parameter to `True` this method will return both directions if one or more undirected resources present. If `False`, only the undirected interactions will be considered, and if any resource annotates this interaction as undirected both directions will be returned. However the ``count_interactions_undirected`` method will return `1` in this case. :arg NoneType,bool,str effect: If `None` also interactions without effect, if `True` only the ones with any effect, if a string naming an effect only the interactions with that specific effect will be considered. :arg NontType,str,set resources: Optionally limit the query to one or more resources. :arg NontType,str,set data_model: Optionally limit the query to one or more data models e.g. `activity_flow`. :arg NontType,str,set interaction_type: Optionally limit the query to one or more interaction types e.g. `PPI`. :arg NontType,bool,str,set via: Optionally limit the query to certain secondary databases or if `False` consider only data from primary databases. """ effect = self._effect_synonyms(effect) direction = ( self.direction_key(direction) if isinstance(direction, tuple) else direction ) return tuple( # direction key _dir # possible directions for _dir in (self.a_b, self.b_a) # conditions by selecting and evaluating evidence collections if self.evaluate_evidences( this_direction = _dir, direction = direction, effect = effect, resources = resources, data_model = data_model, interaction_type = interaction_type, via = via, references = references, ) )
[docs] def evaluate_evidences( self, this_direction, direction = None, effect = None, resources = None, data_model = None, interaction_type = None, via = None, references = None, ): """ Selects the evidence collections matching the direction and effect criteria and then evaluates if any of the evidences in these collections match the evidence criteria. """ kwargs = locals() _ = kwargs.pop('self') return any(self.iter_match_evidences(**kwargs))
[docs] def iter_match_evidences( self, this_direction, direction = None, effect = None, resources = None, data_model = None, interaction_type = None, via = None, references = None, ): """ Selects the evidence collections matching the direction and effect criteria and yields collections matching the evidence criteria. """ for evs in self.iter_evidences( this_direction = this_direction, direction = direction, effect = effect, ): if evs.match( resource = resources, data_model = data_model, interaction_type = interaction_type, via = via, references = references, ): yield evs
[docs] def iter_evidences( self, this_direction, direction = None, effect = None, ): """ Selects and yields evidence collections matching the direction and effect criteria. """ # evidence keys for evs_key in ('undirected', this_direction): # evidence dicts for this_effect in ('direction', 'positive', 'negative'): if ( # only undirected ( direction == False and evs_key == 'undirected' and this_effect == 'direction' ) or # undirected ( direction is None and not effect and this_effect == 'direction' ) or # directed ( direction != False and evs_key != 'undirected' and this_effect == 'direction' and not effect and ( # any direction direction == True or # specific direction direction == this_direction ) ) or # with effect ( direction != False and evs_key != 'undirected' and this_effect != 'direction' and ( # any effect effect == True or # specific effect effect == this_effect ) ) ): # getting the evidence dict and the key from it yield getattr(self, this_effect)[evs_key]
[docs] def get_interactions_0(self, **kwargs): """ Returns unique interacting pairs without being aware of the direction. """ kwargs['direction'] = None kwargs['effect'] = None result = self.get_interactions(**kwargs) return result[:1] if result else ()
[docs] def get_interactions_directed(self, **kwargs): """ **kwargs: see the docs of method ``get_interactions``. """ if 'direction' not in kwargs or kwargs['direction'] is None: kwargs['direction'] = True return self.get_interactions(**kwargs)
[docs] def get_interactions_undirected(self, **kwargs): """ Only the undirected interactions will be considered, if any resource annotates this interaction as undirected both directions will be returned, no matter if certain resources provide direction. However the ``count_interactions_undirected`` method will return `1` in this case. **kwargs: see the docs of method ``get_interactions``. """ kwargs['direction'] = False return self.get_interactions(**kwargs)
[docs] def get_interactions_undirected_0(self, **kwargs): """ Only the undirected interactions will be considered, if any resource annotates this interaction as undirected the interacting pair as a sorted tuple will be returned inside a one element tuple. **kwargs: see the docs of method ``get_interactions``. """ undir = self.get_interactions_undirected(**kwargs) return undir[:1] if undir else ()
[docs] def get_interactions_non_directed(self, **kwargs): """ Only the undirected interactions will be considered, if any resource annotates this interaction as undirected both directions will be returned, but only if no resource provide direction. However the ``count_interactions_non_directed`` method will return `1` in this case. **kwargs: see the docs of method ``get_interactions``. """ kwargs['direction'] = True return ( self.get_interactions_undirected(**kwargs) if not self.get_interactions(**kwargs) else () )
[docs] def get_interactions_non_directed_0(self, **kwargs): """ Only the undirected interactions will be considered, if any resource annotates this interaction as undirected and none as directed, the interacting pair as a sorted tuple will be returned inside a one element tuple. **kwargs: see the docs of method ``get_interactions``. """ nondir = self.get_interactions_non_directed(**kwargs) return nondir[:1] if nondir else ()
[docs] def get_interactions_signed(self, **kwargs): """ **kwargs: see the docs of method ``get_interactions``. """ if 'effect' not in kwargs or kwargs['effect'] is None: kwargs['effect'] = True return self.get_interactions(**kwargs)
[docs] def get_interactions_positive(self, **kwargs): """ **kwargs: see the docs of method ``get_interactions``. """ kwargs['effect'] = 'positive' return self.get_interactions(**kwargs)
[docs] def get_interactions_negative(self, **kwargs): """ **kwargs: see the docs of method ``get_interactions``. """ kwargs['effect'] = 'negative' return self.get_interactions(**kwargs)
[docs] def get_interactions_mutual(self, **kwargs): """ Note: undirected interactions does not count as mutual but only interactions with explicit direction information for both directions. **kwargs: see the docs of method ``get_interactions``. """ if 'direction' not in kwargs or kwargs['direction'] is None: kwargs['direction'] = True interactions = self.get_interactions(**kwargs) return interactions if len(interactions) == 2 else ()
[docs] def is_mutual(self, **kwargs): """ Note: undirected interactions does not count as mutual but only interactions with explicit direction information for both directions. **kwargs: see the docs of method ``get_interactions``. """ return bool(self.get_interactions_mutual(**kwargs))
[docs] def count_interactions_mutual(self, **kwargs): """ Note: undirected interactions does not count as mutual but only interactions with explicit direction information for both directions. **kwargs: see the docs of method ``get_interactions``. """ return int(self.is_mutual(**kwargs))
[docs] def count_interactions_undirected(self, **kwargs): """ Returns `True` if any resource annotates this interaction without direction. **kwargs: see the docs of method ``get_interactions``. """ return bool(self.get_interactions_undirected(**kwargs))
[docs] def count_interactions_non_directed(self, **kwargs): """ Returns `True` if any resource annotates this interaction without and no resource with direction. **kwargs: see the docs of method ``get_interactions``. """ return bool(self.get_interactions_non_directed(**kwargs))
[docs] def get_degrees( self, mode, direction = None, effect = None, resources = None, data_model = None, interaction_type = None, via = None, references = None, ): """ Returns a *set* of nodes with the connections matching the direction, effect and evidence criteria. E.g. if the query concerns the incoming degrees with positive effect and the matching evidences show A activates B, but not the other way around, only "B" will be returned. :arg str mode: The type of degrees to be considered. Three possible values are ``'IN'``, `'OUT'`` and ``'ALL'`` for incoming, outgoing and all connections, respectively. If the ``direction`` is ``False`` the only possible mode is ``ALL``. If the ``direction`` is ``None`` and also directed evidence(s) match the criteria these will overwrite the undirected evidences and only the directed result will be returned. """ kwargs = locals() _ = kwargs.pop('self') mode = kwargs.pop('mode') idx = { 'ALL': (0, 2), 'OUT': (0, 1), 'IN': (1, 2), } if direction == False: mode = 'ALL' if direction is None and not effect: _ = kwargs.pop('direction') return ( self.get_degrees(mode = mode, direction = True, **kwargs) or self.get_degrees(mode = mode, direction = False, **kwargs) ) result = set() node_pairs = self.get_interactions(**kwargs) for pair in node_pairs: result.update( pair[ idx[mode][0]: idx[mode][1] ] ) return result
def get_curation_effort(self, **kwargs): for ref in self.get_references(): return (self.a, self.b, ref) @staticmethod def _get(self, method, **kwargs): return getattr( self.get_evidences( **kwargs ), 'get_%s' % method, )() @staticmethod def _count(method): @functools.wraps(method) def count_method(*args, **kwargs): return len(method(*args, **kwargs)) return count_method @staticmethod def _by(method, by = 'resources'): by = (by,) if isinstance(by, common.basestring) else by @functools.wraps(method) def by_method(*args, name_keys = True, **kwargs): self = args[0] for _by in by: _ = kwargs.pop(_by, None) levels_methods = ( 'get_%s%ss' % ( _by[:-1] if _by in {'resources', 'references'} else _by, '_name' if _by == 'resources' and name_keys else '' ) for _by in by ) levels = list(itertools.product(*( getattr(self, levels_method)() for levels_method in levels_methods ))) result = dict( ( _levels if len(_levels) > 1 else _levels[0], method( *args, **dict(zip(by, _levels)), **kwargs ) ) for _levels in levels ) return dict((k, v) for k, v in iteritems(result) if v) return by_method @classmethod def _by_resource(cls, method): return cls._by(method, by = 'resources') @classmethod def _by_data_model(cls, method): return cls._by(method, by = 'data_model') @classmethod def _by_interaction_type_and_data_model(cls, method): return cls._by(method, by = ('interaction_type', 'data_model')) @classmethod def _by_interaction_type_and_data_model_and_resource(cls, method): return cls._by( method, by = ('interaction_type', 'data_model', 'resources'), ) @classmethod def _by_interaction_type(cls, method): return cls._by(method, by = 'interaction_type') @classmethod def _by_reference(cls, method): return cls._by(method, by = 'references') @classmethod def _generate_get_methods(cls): def _create_get_method(method): @functools.wraps(method) def _get_method(*args, **kwargs): return cls._get(self = args[0], method = method, **kwargs) return _get_method for _get in cls._get_methods_autogen: cls._add_method( method_name = 'get_%s' % _get, method = _create_get_method(_get), # this is not always correct, to be fixed later signature = cls._get_method_signature, doc = ( 'Retrieves %s matching the criteria.' % ( _get.replace('_', ' ') ) ), ) @classmethod def _generate_degree_methods(cls): def _create_degree_method(mode, direction, effect): wrap_args = (mode, direction, effect) @functools.wraps(wrap_args) def _degree_method(*args, **kwargs): mode, direction, effect = wrap_args kwargs['direction'] = direction kwargs['effect'] = effect return cls.get_degrees(self = args[0], mode = mode, **kwargs) return _degree_method for mode, (dir_label, dir_args) in itertools.product( cls._degree_modes, iteritems(cls._degree_directions) ): if dir_label in {'undirected', 'non_directed'} and mode != 'ALL': continue method_name = 'degrees_%s%s' % ( dir_label, '_%s' % mode.lower() if mode != 'ALL' else '' ) cls._count_methods.add(method_name) cls._get_methods.add(method_name) method = _create_degree_method(mode, *dir_args) _method_name = 'get_%s' % method_name cls._add_method( method_name = _method_name, method = method, signature = ['mode'] + cls._get_method_signature, doc = cls.get_degrees.__doc__, ) @classmethod def _generate_count_methods(cls): for _get in cls._count_methods: _get_method = getattr(cls, 'get_%s' % _get) cls._add_method( method_name = 'count_%s' % _get, method = cls._count(_get_method), signature = cls._get_method_signature, doc = _get_method.__doc__, ) @classmethod def _generate_by_methods(cls): for _get, _by in itertools.product( cls._get_methods, cls._by_methods, ): _get_method = getattr(cls, 'get_%s' % _get) method_name = '%s_by_%s' % (_get, _by) method = getattr(cls, '_by_%s' % _by)(_get_method) cls._add_method( method_name = method_name, method = method, signature = cls._get_method_signature, doc = _get_method.__doc__, ) @classmethod def _add_method(cls, method_name, method, signature = None, doc = None): common._add_method( cls, method_name, method, signature = signature, doc = doc, )
[docs] def generate_df_records(self, by_source = False, with_references = False): """ Yields interaction records. It is a generator because one edge can be represented by one or more records depending on the signs and directions and other parameters :arg bool by_source: Yield separate records by resources. This way the node pairs will be redundant and you need to group later if you want unique interacting pairs. By default is ``False`` because for most applications unique interactions are preferred. If ``False`` the *refrences* field will still be present but with ``None`` values. :arg bool with_references: Include the literature references. By default is ``False`` because you rarely need these and they increase the data size significantly. """ def source_add_via(source, via): return '%s%s' % (source, '_%s' % via if via else '') def iter_sources(evs): sources = evs.get_resource_names_via() if by_source: for source, via in sources: refs = ( { ref.pmid for ref in self.get_references( resources = source, interaction_type = interaction_type, via = via, ) } if with_references else None ) _source = source_add_via(source, via) yield _source, refs else: _sources = { source_add_via(source, via) for source, via in sources } refs = ( { ref.pmid for ref in self.get_references( resources = {s[0] for s in sources}, interaction_type = interaction_type, ) } if with_references else None ) if _sources: yield _sources, refs for interaction_type in self.get_interaction_types(): dmodels = ( self.get_data_models(interaction_type = interaction_type) ) dmodels = dmodels if by_source else (dmodels,) for data_model in dmodels: evs_undirected = self.get_evidences( direction = 'undirected', interaction_type = interaction_type, data_model = data_model, ) for _dir in (self.a_b, self.b_a): evs_dir = self.get_evidences( direction = _dir, interaction_type = interaction_type, data_model = data_model, ) evs_without_sign = evs_dir.__copy__() for _effect, effect in zip( (1, -1), ('positive', 'negative') ): evs_sign = self.get_evidences( direction = _dir, effect = effect, interaction_type = interaction_type, data_model = data_model, ) # to make sure we keep all references: evs_sign += evs_sign.intersection(evs_dir) evs_without_sign -= evs_sign evs_undirected -= evs_sign for sources, refs in iter_sources(evs_sign): yield InteractionDataFrameRecord( id_a = _dir[0].identifier, id_b = _dir[1].identifier, type_a = _dir[0].entity_type, type_b = _dir[1].entity_type, directed = True, effect = _effect, type = interaction_type, dmodel = data_model, sources = sources, references = refs, ) if evs_without_sign: evs_undirected -= evs_without_sign for sources, refs in iter_sources(evs_without_sign): yield InteractionDataFrameRecord( id_a = _dir[0].identifier, id_b = _dir[1].identifier, type_a = _dir[0].entity_type, type_b = _dir[1].entity_type, directed = True, effect = 0, type = interaction_type, dmodel = data_model, sources = sources, references = refs, ) if evs_undirected: for sources, refs in iter_sources(evs_undirected): yield InteractionDataFrameRecord( id_a = self.a.identifier, id_b = self.b.identifier, type_a = self.a.entity_type, type_b = self.b.entity_type, directed = False, effect = 0, type = interaction_type, dmodel = data_model, sources = sources, references = refs, )
Interaction._generate_entity_methods() Interaction._generate_get_methods() Interaction._generate_degree_methods() Interaction._generate_count_methods() Interaction._generate_by_methods()