Resetting utils subtree.
This commit is contained in:
@@ -1,265 +0,0 @@
|
||||
"""
|
||||
|
||||
AUTHOR:
|
||||
|
||||
Khushal P Soonderji
|
||||
|
||||
DATE:
|
||||
|
||||
Sunday 1st Sept. 2024.
|
||||
|
||||
OBJECTIVE:
|
||||
|
||||
To provide a way to convert any input data to serialized bytes, and back.
|
||||
|
||||
REFERENCES:
|
||||
|
||||
N/A
|
||||
|
||||
DOWNLOADS:
|
||||
|
||||
N/A
|
||||
|
||||
"""
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** IMPORT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# To make sibling directories accessible for imports:
|
||||
import sys
|
||||
sys.path.append(".")
|
||||
sys.path.append("..")
|
||||
|
||||
# To work with files:
|
||||
from utils_v2.string import json
|
||||
|
||||
# To work with tabulated data:
|
||||
import pandas as pd
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MACROS / ONE-TIME INIT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** VARIABLES ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** FUNCTIONS ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** CLASSES ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
class UniversalSerializer:
|
||||
|
||||
def __init__(self):
|
||||
|
||||
"""
|
||||
Use this when you are working with varied datatypes. You can add custom data-converters also using the
|
||||
'add_converters' method. Otherwise, most default pythonic datatypes are supported out of the box. Note that this
|
||||
is NOT recommended because of how large the serialized messages become. Try using 'JSONSerializer' when you know
|
||||
you will be working specifically with JSOn-compatible inputs.
|
||||
"""
|
||||
|
||||
# These are the converters to use when serializing data:
|
||||
self.__forward_converters = {
|
||||
"set": lambda x: list(x),
|
||||
"tuple": lambda x: list(x),
|
||||
"complex": lambda x: {"r": x.real, "i": x.imag},
|
||||
"DataFrame": lambda x: x.to_dict()
|
||||
}
|
||||
|
||||
# These are the converters to use when deserializing data:
|
||||
self.__reverse_converters = {
|
||||
"set": lambda x: set(x),
|
||||
"tuple": lambda x: tuple(x),
|
||||
"complex": lambda x: complex(x["r"], x["i"]),
|
||||
"DataFrame": lambda x: pd.DataFrame.from_dict(x)
|
||||
}
|
||||
|
||||
def add_converters(
|
||||
self,
|
||||
type_name,
|
||||
forward_converter_func,
|
||||
reverse_converter_func
|
||||
):
|
||||
|
||||
"""
|
||||
Add custom datatype converters.
|
||||
RULES:
|
||||
01. Each of the converter functions must take in exactly on argument and return one output of native python
|
||||
type. This is very important.
|
||||
02. Each forward and reverse converters must give symmetric results.
|
||||
:param type_name: The name of the datatype. HINT: type(obj).__name__
|
||||
:param forward_converter_func: The function to handle conversion to bytes. Use when serializing.
|
||||
:param reverse_converter_func: The function to handle conversion from bytes. Used when deserializing.
|
||||
:return: None.
|
||||
"""
|
||||
|
||||
self.__forward_converters[type_name] = lambda x: forward_converter_func(x)
|
||||
self.__reverse_converters[type_name] = lambda x: reverse_converter_func(x)
|
||||
|
||||
def __describe(self, data):
|
||||
|
||||
"""
|
||||
Notes down the input datatypes of everything.
|
||||
Does everything upto conversion to byes.
|
||||
:param data: The data to process.
|
||||
:return: The description of the datatypes and values of what was given.
|
||||
"""
|
||||
|
||||
# Note down the type of data that was sent as the input:
|
||||
data_type = type(data).__name__
|
||||
|
||||
# Handle iterables:
|
||||
if isinstance(data, list): data = [self.__describe(item) for item in data]
|
||||
elif isinstance(data, set): data = [self.__describe(item) for item in data]
|
||||
elif isinstance(data, tuple): data = [self.__describe(item) for item in data]
|
||||
elif isinstance(data, dict): data = [
|
||||
{
|
||||
"k": self.__describe(k),
|
||||
"v": self.__describe(v)
|
||||
} for k, v in data.items()
|
||||
]
|
||||
|
||||
# Convert here, and return:
|
||||
conv = self.__forward_converters.get(data_type)
|
||||
if conv is not None: data = conv(data)
|
||||
return {"d": data, "t": data_type}
|
||||
|
||||
def serialize(self, data, encoding = "utf-8"):
|
||||
|
||||
"""
|
||||
Serializes the data that is given to it.
|
||||
:param data: The data to serialize.
|
||||
:param encoding: The encoding to use.
|
||||
:return: The bytes representing the data.
|
||||
"""
|
||||
|
||||
data = self.__describe(data)
|
||||
data = json.to_string(data, no_space = True)
|
||||
return data.encode(encoding)
|
||||
|
||||
def __interpret(self, data):
|
||||
|
||||
"""
|
||||
Interprets the types of data that were serialized originally.
|
||||
:param data: The data in the serialized form.
|
||||
:return: Data where the appropriate datatypes have been applied.
|
||||
"""
|
||||
|
||||
# Handle iterables:
|
||||
if data["t"] == "list": data = [self.__interpret(item) for item in data["d"]]
|
||||
elif data["t"] == "set": data = set([self.__interpret(item) for item in data["d"]])
|
||||
elif data["t"] == "tuple": data = tuple([self.__interpret(item) for item in data["d"]])
|
||||
elif data["t"] == "dict": data = {
|
||||
self.__interpret(item["k"]): self.__interpret(item["v"])
|
||||
for item in data["d"]
|
||||
}
|
||||
|
||||
# Handle custom types:
|
||||
else:
|
||||
conv = self.__reverse_converters.get(data["t"])
|
||||
data = data["d"]
|
||||
if conv is not None: data = conv(data)
|
||||
|
||||
# Done here
|
||||
return data
|
||||
|
||||
def deserialize(self, data, encoding = "utf-8"):
|
||||
|
||||
"""
|
||||
Deserializes the bytes that are given to it.
|
||||
:param data: The bytes to deserialize.
|
||||
:param encoding: The encoding to use.
|
||||
:return: The data from the bytes that described it.
|
||||
"""
|
||||
|
||||
data = data.decode(encoding)
|
||||
data = json.from_string(data)
|
||||
return self.__interpret(data)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
class JSONSerializer:
|
||||
|
||||
def __init__(self):
|
||||
|
||||
"""
|
||||
Use this serializer when dealing with JSON-compatible data like direct JSON-strings, python dicts, and
|
||||
python-lists. Beware that non-compatible data will cause either direct exceptions or unexpected behaviour.
|
||||
"""
|
||||
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def serialize(data, encoding = "utf-8"):
|
||||
|
||||
"""
|
||||
Serializes the data that is given to it.
|
||||
The input has to be JSON-compatible.
|
||||
:param data: The data to serialize.
|
||||
:param encoding: The encoding to use.
|
||||
:return: The bytes representing the data.
|
||||
"""
|
||||
|
||||
# If the data is not already a JSON string, parse it. Then return it as bytes:
|
||||
data = data if isinstance(data, str) else json.to_string(data, no_space = True)
|
||||
return data.encode(encoding)
|
||||
|
||||
@staticmethod
|
||||
def deserialize(data, encoding = "utf-8"):
|
||||
|
||||
"""
|
||||
Deserializes the bytes that are given to it.
|
||||
The input has to be JSON-compatible.
|
||||
:param data: The bytes to deserialize.
|
||||
:param encoding: The encoding to use.
|
||||
:return: The data from the bytes that described it.
|
||||
"""
|
||||
|
||||
data = data.decode(encoding)
|
||||
return json.from_string(data)
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MAIN PROGRAM ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
pass
|
||||
Reference in New Issue
Block a user