Source code for timeflux.core.message

"""timeflux.core.message: serialize and unserialize dataframes"""

import pickle
import pandas as pd

# import pyarrow as pa


[docs]def pickle_serialize(message): topic = message[0] data = message[1] meta = message[2] return [ topic, pickle.dumps(data, pickle.HIGHEST_PROTOCOL), pickle.dumps(meta, pickle.HIGHEST_PROTOCOL), ]
[docs]def pickle_deserialize(message): topic = message[0].decode("utf-8") data = message[1] meta = message[2] return [topic, pickle.loads(data), pickle.loads(meta)]
[docs]def msgpack_serialize(message): # TODO: handle meta and cases where data is None topic = message[0] data = message[1] return [topic, data.to_msgpack()]
[docs]def msgpack_deserialize(message): # TODO: handle meta and cases where data is None topic = message[0].decode("utf-8") data = message[1] return [topic, pd.read_msgpack(data)]
# def arrow_serialize(message): # topic = message[0].decode('utf-8') # df = message[1] # return [topic, pa.serialize(df).to_buffer()] # def arrow_deserialize(message): # topic = message[0] # data = message[1] # return [topic, pa.deserialize(data)]