# -*- coding: utf-8 -*-
"""Streamable BEL as JSON."""
import gzip
import json
from typing import Any, Iterable, List, TextIO, Union
from networkx.utils import open_file
from .nodelink import _augment_node, _prepare_graph_dict, _recover_graph_dict
from ..constants import CITATION, SOURCE_MODIFIER, TARGET_MODIFIER
from ..language import CitationDict
from ..struct.graph import BELGraph, _handle_modifier
from ..tokens import parse_result_to_dsl
from ..utils import hash_edge
__all__ = [
'to_sbel_file',
'to_sbel',
'to_sbel_gz',
'from_sbel',
'from_sbel_gz',
'from_sbel_file',
]
SBEL = Any
[docs]@open_file(1, mode='w')
def to_sbel_file(graph: BELGraph, path: Union[str, TextIO], separators=(',', ':'), **kwargs) -> None:
"""Write this graph as BEL JSONL to a file.
:param graph: A BEL graph
:param separators: The separators used in :func:`json.dumps`
:param path: A path or file-like
"""
for i in iterate_sbel(graph):
print(json.dumps(i, ensure_ascii=False, separators=separators, **kwargs), file=path)
[docs]def to_sbel_gz(graph: BELGraph, path: str, separators=(',', ':'), **kwargs) -> None:
"""Write a graph as BEL JSONL to a gzip file.
:param graph: A BEL graph
:param separators: The separators used in :func:`json.dumps`
:param path: A path for a gzip file
"""
with gzip.open(path, 'wt') as file:
to_sbel_file(graph, file, separators=separators, **kwargs)
[docs]def to_sbel(graph: BELGraph) -> List[SBEL]:
"""Create a list of JSON dictionaries corresponding to lines in BEL JSONL."""
return list(iterate_sbel(graph))
def iterate_sbel(graph: BELGraph) -> Iterable[SBEL]:
"""Iterate over JSON dictionaries corresponding to lines in BEL JSONL."""
g = graph.graph.copy()
_prepare_graph_dict(g)
yield g
for u, v, k, d in graph.edges(data=True, keys=True):
yield {
'source': _augment_node(u),
'target': _augment_node(v),
'key': k,
**d,
}
[docs]def from_sbel(it: Iterable[SBEL], includes_metadata: bool = True) -> BELGraph:
"""Load a BEL graph from an iterable of dictionaries corresponding to lines in BEL JSONL.
:param it: An iterable of dictionaries.
:param includes_metadata: By default, interprets the first element of the iterable as the graph's metadata.
Switch to ``False`` to disable.
:return: A BEL graph
"""
it = iter(it)
rv = BELGraph()
if includes_metadata:
rv.graph.update(next(it))
_recover_graph_dict(rv)
add_sbel(rv, it)
return rv
def add_sbel(graph: BELGraph, it: Iterable[SBEL]) -> None:
"""Add dictionaries to a BEL graph.
:param graph: A BEL graph
:param it: An iterable of dictionaries.
"""
for data in it:
add_sbel_row(graph, data)
def add_sbel_row(graph: BELGraph, data: SBEL) -> str:
"""Add a single SBEL data dictionary to a graph."""
u = parse_result_to_dsl(data['source'])
v = parse_result_to_dsl(data['target'])
edge_data = {
k: v
for k, v in data.items()
if k not in {'source', 'target', 'key'}
}
for side in (SOURCE_MODIFIER, TARGET_MODIFIER):
side_data = edge_data.get(side)
if side_data:
_handle_modifier(side_data)
if CITATION in edge_data:
edge_data[CITATION] = CitationDict(**edge_data[CITATION])
return graph.add_edge(u, v, key=hash_edge(u, v, edge_data), **edge_data)
[docs]@open_file(0, mode='r')
def from_sbel_file(path: Union[str, TextIO]) -> BELGraph:
"""Build a graph from the BEL JSONL contained in the given file.
:param path: A path or file-like
"""
return from_sbel((
json.loads(line)
for line in path
))
[docs]def from_sbel_gz(path: str) -> BELGraph:
"""Read a graph as BEL JSONL from a gzip file."""
with gzip.open(path, 'rt') as file:
return from_sbel_file(file)