-
Notifications
You must be signed in to change notification settings - Fork 0
/
communication_utils.py
55 lines (43 loc) · 1.4 KB
/
communication_utils.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
import requests
import sys
import time
import os
import json
import types
from bson import Binary
import base64
import pickle
import cloudpickle
import zlib
from exception_mixin import generic_exception_handler
from flask_socketio import SocketIO
def is_jsonizable(dat):
try:
_ = json.dumps(dat)
return True
except:
return False
def make_jsonizable_and_compress(dat):
return zlib.compress(make_python_object_jsonizable(dat, output_string=False))
def make_python_object_jsonizable(dat, output_string=True):
if isinstance(dat, types.FunctionType): # handle functions specially
dat.__module__ = "__main__" # without this, cloudpickle only generates a reference to the function
jdat = base64.b64encode(cloudpickle.dumps(dat))
else:
try:
jdat = base64.b64encode(pickle.dumps(dat, protocol=2))
except:
jdat = base64.b64encode(cloudpickle.dumps(dat))
if output_string and not isinstance(jdat, str):
jdat = jdat.decode("utf-8")
return jdat
def debinarize_python_object(bdat):
if isinstance(bdat, Binary):
dat = bdat.decode()
else:
dat = base64.b64decode(bdat)
return pickle.loads(dat)
def read_project_dict(fs, file_id):
binarized_python_object = zlib.decompress(fs.get(file_id).read())
project_dict = debinarize_python_object(binarized_python_object)
return project_dict