turbo_broccoli.custom.networkx

NetworkX graph serialization and deserialization.

 1"""NetworkX graph serialization and deserialization."""
 2
 3from typing import Any, Callable, Tuple
 4
 5import networkx as nx
 6
 7from turbo_broccoli.context import Context
 8from turbo_broccoli.exceptions import DeserializationError, TypeNotSupported
 9
10
11def _graph_to_json(obj: nx.Graph, ctx: Context) -> dict:
12    return {
13        "__type__": "networkx.graph",
14        "__version__": 1,
15        "data": nx.adjacency_data(obj),
16    }
17
18
19def _json_to_graph(dct: dict, ctx: Context) -> nx.Graph:
20    decoders = {1: _json_to_graph_v1}
21    return decoders[dct["__version__"]](dct, ctx)
22
23
24def _json_to_graph_v1(dct: dict, ctx: Context) -> nx.Graph:
25    return nx.adjacency_graph(dct["data"])
26
27
28# pylint: disable=missing-function-docstring
29def from_json(dct: dict, ctx: Context) -> nx.Graph:
30    decoders = {
31        "networkx.graph": _json_to_graph,
32    }
33    try:
34        type_name = dct["__type__"]
35        return decoders[type_name](dct, ctx)
36    except KeyError as exc:
37        raise DeserializationError() from exc
38
39
40def to_json(obj: nx.Graph, ctx: Context) -> dict:
41    """
42    Serializes a graph into JSON by cases. The return dict has the following
43    structure
44
45    ```py
46    {
47        "__type__": "networkx.graph",
48        "__version__": 1,
49        "data": {...}
50    }
51    ```
52
53    where the `{...}` is produced by
54    [`networkx.adjacency_data`](https://networkx.org/documentation/stable/reference/readwrite/generated/networkx.readwrite.json_graph.adjacency_data.html#adjacency-data).
55    """
56    encoders: list[Tuple[type, Callable[[Any, Context], dict]]] = [
57        (nx.Graph, _graph_to_json),
58    ]
59    for t, f in encoders:
60        if isinstance(obj, t):
61            return f(obj, ctx)
62    raise TypeNotSupported()
def from_json( dct: dict, ctx: turbo_broccoli.context.Context) -> networkx.classes.graph.Graph: 
30def from_json(dct: dict, ctx: Context) -> nx.Graph:
31    decoders = {
32        "networkx.graph": _json_to_graph,
33    }
34    try:
35        type_name = dct["__type__"]
36        return decoders[type_name](dct, ctx)
37    except KeyError as exc:
38        raise DeserializationError() from exc
def to_json( obj: networkx.classes.graph.Graph, ctx: turbo_broccoli.context.Context) -> dict: 
41def to_json(obj: nx.Graph, ctx: Context) -> dict:
42    """
43    Serializes a graph into JSON by cases. The return dict has the following
44    structure
45
46    ```py
47    {
48        "__type__": "networkx.graph",
49        "__version__": 1,
50        "data": {...}
51    }
52    ```
53
54    where the `{...}` is produced by
55    [`networkx.adjacency_data`](https://networkx.org/documentation/stable/reference/readwrite/generated/networkx.readwrite.json_graph.adjacency_data.html#adjacency-data).
56    """
57    encoders: list[Tuple[type, Callable[[Any, Context], dict]]] = [
58        (nx.Graph, _graph_to_json),
59    ]
60    for t, f in encoders:
61        if isinstance(obj, t):
62            return f(obj, ctx)
63    raise TypeNotSupported()

Serializes a graph into JSON by cases. The return dict has the following structure

{
    "__type__": "networkx.graph",
    "__version__": 1,
    "data": {...}
}

where the {...} is produced by networkx.adjacency_data.