turbo_broccoli.custom

  1"""Custom type encoder and decoders, all grouped in a dedicated submodule"""
  2
  3from typing import Any, Callable
  4
  5from ..context import Context
  6from . import bytes as _bytes
  7from . import collections as _collections
  8from . import dataclass as _dataclass
  9from . import datetime as _datetime
 10from . import dct as _dict
 11from . import embedded as _embedded
 12from . import external as _external
 13from . import generic as _generic
 14from . import networkx as _networkx
 15from . import pathlib as _pathlib
 16from . import uuid as _uuid
 17
 18try:
 19    from turbo_broccoli.custom import keras as _keras
 20
 21    HAS_KERAS = True
 22except:
 23    HAS_KERAS = False
 24
 25try:
 26    from turbo_broccoli.custom import numpy as _numpy
 27
 28    HAS_NUMPY = True
 29except:
 30    HAS_NUMPY = False
 31
 32try:
 33    from turbo_broccoli.custom import pandas as _pandas
 34
 35    HAS_PANDAS = True
 36except:
 37    HAS_PANDAS = False
 38
 39
 40try:
 41    from turbo_broccoli.custom import secret as _secret
 42
 43    HAS_SECRET = True
 44except:
 45    HAS_SECRET = False
 46
 47try:
 48    from turbo_broccoli.custom import tensorflow as _tensorflow
 49
 50    HAS_TENSORFLOW = True
 51except:
 52    HAS_TENSORFLOW = False
 53
 54try:
 55    from turbo_broccoli.custom import pytorch as _pytorch
 56
 57    HAS_PYTORCH = True
 58except:
 59    HAS_PYTORCH = False
 60
 61try:
 62    from turbo_broccoli.custom import scipy as _scipy
 63
 64    HAS_SCIPY = True
 65except:
 66    HAS_SCIPY = False
 67
 68try:
 69    from turbo_broccoli.custom import sklearn as _sklearn
 70
 71    HAS_SKLEARN = True
 72except:
 73    HAS_SKLEARN = False
 74
 75try:
 76    from turbo_broccoli.custom import bokeh as _bokeh
 77
 78    HAS_BOKEH = True
 79except:
 80    HAS_BOKEH = False
 81
 82
 83def get_decoders() -> dict[str, Callable[[dict, Context], Any]]:
 84    """
 85    Returns the dict of all available decoders, which looks like this:
 86
 87    ```py
 88    {
 89        "mytype": mytype_decoder,
 90        ...
 91    }
 92    ```
 93
 94    `mytype_decoder` is a function that takes an vanilla JSON dict that
 95    looks like this (excluding comments):
 96
 97    ```py
 98    {
 99        "__type__": "mytype.mysubtype",  # or simply "mytype"
100        "__version__": <int>,
101        ...
102    }
103    ```
104    """
105    decoders: dict[str, Callable[[dict, Context], Any]] = {
106        "bytes": _bytes.from_json,
107        "datetime": _datetime.from_json,
108        "dict": _dict.from_json,
109        "external": _external.from_json,
110        "networkx": _networkx.from_json,
111        "pathlib": _pathlib.from_json,
112        "uuid": _uuid.from_json,
113    }
114    if HAS_KERAS:
115        decoders["keras"] = _keras.from_json
116    if HAS_NUMPY:
117        decoders["numpy"] = _numpy.from_json
118    if HAS_PANDAS:
119        decoders["pandas"] = _pandas.from_json
120    if HAS_PYTORCH:
121        decoders["pytorch"] = _pytorch.from_json
122    if HAS_SECRET:
123        decoders["secret"] = _secret.from_json
124    if HAS_TENSORFLOW:
125        decoders["tensorflow"] = _tensorflow.from_json
126    if HAS_SCIPY:
127        decoders["scipy"] = _scipy.from_json
128    if HAS_SKLEARN:
129        decoders["sklearn"] = _sklearn.from_json
130    if HAS_BOKEH:
131        decoders["bokeh"] = _bokeh.from_json
132    # Intentionally put last
133    decoders["collections"] = _collections.from_json
134    decoders["dataclass"] = _dataclass.from_json
135    decoders["embedded"] = _embedded.from_json
136    return decoders
137
138
139def get_encoders() -> list[Callable[[Any, Context], dict]]:
140    """
141    Returns the dict of all available encoder. An encoder is a function that
142    takes an object and returns a readily vanilla JSON-serializable dict. This
143    this should be of the form
144
145    ```py
146    {
147        "__type__": "mytype.mysubtype",  # or simply "mytype"
148        "__version__": <int>,
149        ...
150    }
151    ```
152
153    The encoder should raise a `turbo_broccoli.utils.TypeNotSupported` if it
154    doesn't handle the kind of object it was given.
155    """
156    encoders: list[Callable[[Any, Context], dict]] = [
157        _bytes.to_json,
158        _datetime.to_json,
159        _dict.to_json,
160        _external.to_json,
161        _networkx.to_json,
162        _pathlib.to_json,
163        _uuid.to_json,
164    ]
165    if HAS_KERAS:
166        encoders.append(_keras.to_json)
167    if HAS_NUMPY:
168        encoders.append(_numpy.to_json)
169    if HAS_PANDAS:
170        encoders.append(_pandas.to_json)
171    if HAS_PYTORCH:
172        encoders.append(_pytorch.to_json)
173    if HAS_SECRET:
174        encoders.append(_secret.to_json)
175    if HAS_TENSORFLOW:
176        encoders.append(_tensorflow.to_json)
177    if HAS_SCIPY:
178        encoders.append(_scipy.to_json)
179    if HAS_SKLEARN:
180        encoders.append(_sklearn.to_json)
181    if HAS_BOKEH:
182        encoders.append(_bokeh.to_json)
183    # Intentionally put last
184    encoders += [
185        _collections.to_json,
186        _dataclass.to_json,
187        _generic.to_json,
188        _embedded.to_json,
189    ]
190    return encoders