Source code for cosapp.utils.json

import base64
import io
import json
import pickle
from typing import Any, Dict

import numpy


[docs] class JSONEncoder(json.JSONEncoder): """Encode CoSApp object for JSON serialization."""
[docs] def default(self, obj): if isinstance(obj, bytes): return "data:text/plain;base64," + base64.b64encode(obj).decode("utf-8") elif isinstance(obj, numpy.ndarray): buf = io.BytesIO() numpy.save(buf, obj) return "data:application/vnd.numpy.ndarray;base64," + base64.b64encode(buf.getvalue()).decode('utf-8') elif hasattr(obj, '__json__'): return getattr(obj, '__json__')() else: try: pickled_data = pickle.dumps(obj, protocol=4) except (pickle.PicklingError, RecursionError): return super(JSONEncoder, self).default(obj) else: return "data:application/vnd.python3.pickle;base64," + base64.b64encode(pickled_data).decode('utf-8')
[docs] def decode_cosapp_dict(document: Dict[str, Any]) -> Dict[str, Any]: """Convert JSON serialization back.""" for key, value in document.items(): if isinstance(value, dict): document[key] = decode_cosapp_dict(value) elif isinstance(value, str): if value.startswith("data:"): meta, data = value.split(",", maxsplit=1) meta = meta[5:].lower() if meta.endswith(";base64"): data = base64.b64decode(bytes(data, encoding="utf-8")) if meta.startswith("text/plain;"): document[key] = data elif meta.startswith("application/vnd.numpy.ndarray;"): buf = io.BytesIO(data) document[key] = numpy.load(buf) elif meta.startswith("application/vnd.python3.pickle;"): document[key] = pickle.loads(data) else: raise ValueError(f"Unknown data type {meta}.") return document