Source code for pybel.io.sbel

# -*- 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)