import base64
import copy
import json
import marshal
import os
import types
from abc import ABC, abstractmethod
from typing import Any, Callable, ClassVar
import jsonpickle
import jsonpickle.ext.numpy as jsonpickle_numpy
from ..logs import logger
from ..utils.inspect import dynamic_import
jsonpickle.set_encoder_options("json", indent=4)
jsonpickle_numpy.register_handlers()
[docs]
class FileSerializable(ABC):
"""
A class that can be serialized to a file and reconstructed from a file."""
[docs]
@abstractmethod
def toFile(self, filename: str) -> str:
"""Serialize object to a metafile. This metafile should contain all
data necessary to reconstruct the object.
Args:
filename (str): filename to save object to
Returns:
filename (str): absolute path to the saved metafile of the object
"""
[docs]
@classmethod
@abstractmethod
def fromFile(cls, filename: str) -> object:
"""Reconstruct object from a metafile.
Args:
filename (str): filename of the metafile to load object from
Returns:
obj (object): reconstructed object
"""
[docs]
class JSONSerializable(FileSerializable):
"""A class that can be serialized to JSON and reconstructed from JSON.
Attributes:
_notJSON (list):
list of attributes that should not be serialized to JSON explicitly
"""
_notJSON: ClassVar = ["_json_main"]
def __getstate__(self) -> dict:
"""Get state of object for JSON serialization. Whatever
is returned should be serializable to JSON.
Returns:
o_dict (dict): dictionary of object attributes serializable to JSON
"""
o_dict = {}
for key in self.__dict__:
if key in self._notJSON:
continue
try:
o_dict[key] = copy.deepcopy(self.__dict__[key])
except Exception as exp:
logger.error(f"Could not deepcopy '{key}' because of: {exp}")
raise exp
return o_dict
def __setstate__(self, state: dict):
"""Set state of object from a JSON serialization.
Args:
state (dict): dictionary of object attributes serializable to JSON
"""
self.__dict__.update(state)
[docs]
def toFile(self, filename: str) -> str:
"""Serialize object to a JSON file. This JSON file should
contain all data necessary to reconstruct the object.
Args:
filename (str): filename to save object to
Returns:
filename (str): absolute path to the saved JSON file of the object
"""
json = self.toJSON()
with open(filename, "w") as f:
f.write(json)
return os.path.abspath(filename)
[docs]
@classmethod
def fromFile(cls, filename: str) -> Any:
"""Initialize a new instance from a JSON file.
Args:
filename (str): path to the JSON file
Returns:
instance (object): new instance of the class
"""
with open(filename, "r") as f:
json_str = f.read()
# inject the path to the JSON file itself as a hidden attribute
new_dict = json.loads(json_str)
if "py/state" not in new_dict:
new_dict["py/state"] = {}
new_dict["py/state"]["_json_main"] = os.path.abspath(filename)
json_str = json.dumps(new_dict)
return cls.fromJSON(json_str)
[docs]
def toJSON(self) -> str:
"""Serialize object to a JSON string. This JSON string should
contain all data necessary to reconstruct the object.
Returns:
json (str): JSON string of the object
"""
return jsonpickle.encode(self, unpicklable=True)
[docs]
@classmethod
def fromJSON(cls, json: str) -> Any:
"""Reconstruct object from a JSON string.
Args:
json (str): JSON string of the object
Returns:
obj (object): reconstructed object
"""
return jsonpickle.decode(json)
[docs]
def function_as_string(func: Callable) -> str:
"""Convert a function to a string.
Args:
func (Callable): function to convert
"""
json_form = json.loads(jsonpickle.encode(func, unpicklable=True))
if not json_form or "__main__" in json_form:
return base64.b64encode(marshal.dumps(func.__code__)).decode("ascii")
elif "py/function" in json_form:
return json_form["py/function"]
elif "py/object" in json_form:
return json_form["py/object"]
[docs]
def function_from_string(func_str: str) -> Callable:
"""Convert a function from a string. Conversion from encoded bytecode is
attempted first. If that fails, the string is assumed to be a fully
qualified name of the function.
Args:
func_str (str): string representation of the function
Returns:
processor (Callable): function
"""
try:
return types.FunctionType(marshal.loads(base64.b64decode(func_str)), globals())
except Exception:
return dynamic_import(func_str)