Source code for

from __future__ import absolute_import, print_function, unicode_literals
from builtins import dict, str
import pickle
from indra.databases import relevance_client
from indra.assemblers.pysb import PysbAssembler

[docs]def get_subnetwork(statements, nodes, relevance_network=None, relevance_node_lim=10): """Return a PySB model based on a subset of given INDRA Statements. Statements are first filtered for nodes in the given list and other nodes are optionally added based on relevance in a given network. The filtered statements are then assembled into an executable model using INDRA's PySB Assembler. Parameters ---------- statements : list[indra.statements.Statement] A list of INDRA Statements to extract a subnetwork from. nodes : list[str] The names of the nodes to extract the subnetwork for. relevance_network : Optional[str] The UUID of the NDEx network in which nodes relevant to the given nodes are found. relevance_node_lim : Optional[int] The maximal number of additional nodes to add to the subnetwork based on relevance. Returns ------- model : pysb.Model A PySB model object assembled using INDRA's PySB Assembler from the INDRA Statements corresponding to the subnetwork. """ if relevance_network is not None: relevant_nodes = _find_relevant_nodes(nodes, relevance_network, relevance_node_lim) all_nodes = nodes + relevant_nodes else: all_nodes = nodes filtered_statements = _filter_statements(statements, all_nodes) pa = PysbAssembler() pa.add_statements(filtered_statements) model = pa.make_model() return model
def _filter_statements(statements, agents): """Return INDRA Statements which have Agents in the given list. Only statements are returned in which all appearing Agents as in the agents list. Parameters ---------- statements : list[indra.statements.Statement] A list of INDRA Statements to filter. agents : list[str] A list of agent names that need to appear in filtered statements. Returns ------- filtered_statements : list[indra.statements.Statement] The list of filtered INDRA Statements. """ filtered_statements = [] for s in stmts: if all([a is not None for a in s.agent_list()]) and \ all([ in agents for a in s.agent_list()]): filtered_statements.append(s) return filtered_statements def _find_relevant_nodes(query_nodes, relevance_network, relevance_node_lim): """Return a list of nodes that are relevant for the query. Parameters ---------- query_nodes : list[str] A list of node names to query for. relevance_network : str The UUID of the NDEx network to query relevance in. relevance_node_lim : int The number of top relevant nodes to return. Returns ------- nodes : list[str] A list of node names that are relevant for the query. """ all_nodes = relevance_client.get_relevant_nodes(relevance_network, query_nodes) nodes = [n[0] for n in all_nodes[:relevance_node_lim]] return nodes if __name__ == '__main__': genes = ['EGF', 'EGFR', 'ERBB2', 'GRB2', 'SOS1', 'HRAS', 'RAF1', 'MAP2K1', 'MAPK1'] with open('models/rasmachine/rem/model.pkl', 'rb') as f: model = pickle.load(f) stmts = [] for k, v in model.items(): stmts += v rasmachine_network = '50e3dff7-133e-11e6-a039-06603eb7f303' model = get_subnetwork(stmts, genes, rasmachine_network)