Source code for indra.sources.ndex_cx.api

from __future__ import absolute_import, print_function, unicode_literals
from builtins import dict, str
import json
import logging
import ndex2.client
from .processor import NdexCxProcessor


logger = logging.getLogger(__name__)


[docs]def process_cx_file(file_name, require_grounding=True): """Process a CX JSON file into Statements. Parameters ---------- file_name : str Path to file containing CX JSON. require_grounding: bool Whether network nodes lacking grounding information should be included among the extracted Statements (default is True). Returns ------- NdexCxProcessor Processor containing Statements. """ with open(file_name, 'rt') as fh: json_list = json.load(fh) return process_cx(json_list, require_grounding=require_grounding)
[docs]def process_ndex_network(network_id, username=None, password=None, require_grounding=True): """Process an NDEx network into Statements. Parameters ---------- network_id : str NDEx network ID. username : str NDEx username. password : str NDEx password. require_grounding: bool Whether network nodes lacking grounding information should be included among the extracted Statements (default is True). Returns ------- NdexCxProcessor Processor containing Statements. Returns None if there if the HTTP status code indicates an unsuccessful request. """ nd = ndex2.client.Ndex2(username=username, password=password) res = nd.get_network_as_cx_stream(network_id) if res.status_code != 200: logger.error('Problem downloading network: status code %s' % res.status_code) logger.error('Response: %s' % res.text) return None json_list = res.json() summary = nd.get_network_summary(network_id) return process_cx(json_list, summary=summary, require_grounding=require_grounding)
[docs]def process_cx(cx_json, summary=None, require_grounding=True): """Process a CX JSON object into Statements. Parameters ---------- cx_json : list CX JSON object. summary : Optional[dict] The network summary object which can be obtained via get_network_summary through the web service. THis contains metadata such as the owner and the creation time of the network. require_grounding: bool Whether network nodes lacking grounding information should be included among the extracted Statements (default is True). Returns ------- NdexCxProcessor Processor containing Statements. """ ncp = NdexCxProcessor(cx_json, summary=summary, require_grounding=require_grounding) ncp.get_statements() return ncp