turbo_broccoli.custom.bokeh

Bokeh objects (de)serialization utilities.

  1"""Bokeh objects (de)serialization utilities."""
  2
  3from typing import Any, Callable, Tuple
  4
  5from bokeh.core.serialization import (
  6    Buffer,
  7    Deserializer,
  8    Serialized,
  9    Serializer,
 10)
 11from bokeh.models import Model
 12from bokeh.plotting import figure as Figure
 13
 14from turbo_broccoli.context import Context
 15from turbo_broccoli.exceptions import DeserializationError, TypeNotSupported
 16
 17
 18def _buffer_to_json(obj: Buffer, ctx: Context) -> dict:
 19    return {
 20        "__type__": "bokeh.buffer",
 21        "__version__": 2,
 22        "id": obj.id,
 23        "data": obj.to_bytes(),
 24    }
 25
 26
 27def _generic_to_json(obj: Figure, ctx: Context) -> dict:
 28    s = Serializer().serialize(obj)
 29    return {
 30        "__type__": "bokeh.generic",
 31        "__version__": 2,
 32        "content": s.content,
 33        "buffers": s.buffers,
 34    }
 35
 36
 37def _json_to_buffer(dct: dict, ctx: Context) -> Buffer:
 38    decoders = {
 39        2: _json_to_buffer_v2,
 40    }
 41    return decoders[dct["__version__"]](dct, ctx)
 42
 43
 44def _json_to_buffer_v2(dct: dict, ctx: Context) -> Buffer:
 45    return Buffer(id=dct["id"], data=dct["data"])
 46
 47
 48def _json_to_generic(dct: dict, ctx: Context) -> Any:
 49    decoders = {
 50        2: _json_to_generic_v2,
 51    }
 52    return decoders[dct["__version__"]](dct, ctx)
 53
 54
 55def _json_to_generic_v2(dct: dict, ctx: Context) -> Any:
 56    c, b = dct["content"], dct["buffers"]
 57    return Deserializer().deserialize(Serialized(content=c, buffers=b))
 58
 59
 60# pylint: disable=missing-function-docstring
 61def from_json(dct: dict, ctx: Context) -> Any:
 62    ctx.raise_if_nodecode("bytes")
 63    decoders = {
 64        "bokeh.buffer": _json_to_buffer,
 65        "bokeh.generic": _json_to_generic,
 66    }
 67    try:
 68        type_name = dct["__type__"]
 69        return decoders[type_name](dct, ctx)
 70    except KeyError as exc:
 71        raise DeserializationError() from exc
 72
 73
 74def to_json(obj: Any, ctx: Context) -> dict:
 75    """
 76    Serializes a bokeh object. The return dict has the following structure:
 77
 78    - `bokeh.plotting._figure.figure` or `bokeh.models.Model`:
 79
 80        ```py
 81        {
 82            "__type__": "bokeh.generic",
 83            "__version__": 2,
 84            "content": {...},
 85            "buffers": [...],
 86        }
 87        ```
 88
 89    - `bokeh.core.serialization.Buffer`: (for internal use)
 90
 91        ```py
 92        {
 93            "__type__": "bokeh.buffer",
 94            "__version__": 2,
 95            "id": <str>,
 96            "data": <bytes>,
 97        }
 98        ```
 99
100    """
101    encoders: list[Tuple[type, Callable[[Any, Context], dict]]] = [
102        (Buffer, _buffer_to_json),
103        (Figure, _generic_to_json),
104        (Model, _generic_to_json),
105    ]
106    for t, f in encoders:
107        if isinstance(obj, t):
108            return f(obj, ctx)
109    raise TypeNotSupported()
def from_json(dct: dict, ctx: turbo_broccoli.context.Context) -> Any: 
62def from_json(dct: dict, ctx: Context) -> Any:
63    ctx.raise_if_nodecode("bytes")
64    decoders = {
65        "bokeh.buffer": _json_to_buffer,
66        "bokeh.generic": _json_to_generic,
67    }
68    try:
69        type_name = dct["__type__"]
70        return decoders[type_name](dct, ctx)
71    except KeyError as exc:
72        raise DeserializationError() from exc
def to_json(obj: Any, ctx: turbo_broccoli.context.Context) -> dict: 
 75def to_json(obj: Any, ctx: Context) -> dict:
 76    """
 77    Serializes a bokeh object. The return dict has the following structure:
 78
 79    - `bokeh.plotting._figure.figure` or `bokeh.models.Model`:
 80
 81        ```py
 82        {
 83            "__type__": "bokeh.generic",
 84            "__version__": 2,
 85            "content": {...},
 86            "buffers": [...],
 87        }
 88        ```
 89
 90    - `bokeh.core.serialization.Buffer`: (for internal use)
 91
 92        ```py
 93        {
 94            "__type__": "bokeh.buffer",
 95            "__version__": 2,
 96            "id": <str>,
 97            "data": <bytes>,
 98        }
 99        ```
100
101    """
102    encoders: list[Tuple[type, Callable[[Any, Context], dict]]] = [
103        (Buffer, _buffer_to_json),
104        (Figure, _generic_to_json),
105        (Model, _generic_to_json),
106    ]
107    for t, f in encoders:
108        if isinstance(obj, t):
109            return f(obj, ctx)
110    raise TypeNotSupported()

Serializes a bokeh object. The return dict has the following structure:

  • bokeh.plotting._figure.figure or bokeh.models.Model:

    {
    "__type__": "bokeh.generic",
    "__version__": 2,
    "content": {...},
    "buffers": [...],
}

  • bokeh.core.serialization.Buffer: (for internal use)

    {
    "__type__": "bokeh.buffer",
    "__version__": 2,
    "id": <str>,
    "data": <bytes>,
}