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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
|
from copy import deepcopy
import json
from os import getenv, getpid, path
import platform
from sys import version_info, argv
class Empty(Exception):
pass
class DummyQueue(object):
"""
Dummy queue thread that does nothing. Should only be used if imports fail.
"""
def __init__(self, maxsize=0):
pass
def qsize(self):
return 0
def empty(self):
return True
def full(self):
return False
def put(self, obj, *args, **kwargs):
pass
def put_nowait(self, obj):
pass
def get(self, *args, **kwargs):
raise Empty
def get_nowait(self):
raise Empty
def task_done(self):
pass
def join(self):
pass
def is_python3():
return version_info[0] == 3
def is_windows():
return platform.system() == 'Windows'
def is_linux():
return platform.system() == 'Linux'
def is_mac_osx():
# this may not be accurate, just going off of what I find off the internet
return platform.system() == 'Darwin'
def get_temp_path():
if is_windows():
return None
for val in ('XDG_RUNTIME_DIR', 'TMPDIR', 'TMP', 'TEMP'):
tmp = getenv(val)
if tmp is not None:
return tmp
return '/tmp'
def get_process_id():
return getpid()
def is_callable(obj):
try:
# for Python 2.x or Python 3.2+
return callable(obj)
except Exception:
# for Python version: 3 - 3.2
return hasattr(obj, '__call__')
# python 2 + 3 compatibility
if is_python3():
unicode = str
bytes = bytes
else:
bytes = str
unicode = unicode
def to_bytes(obj):
if isinstance(obj, type(b'')):
return obj
if hasattr(obj, 'encode') and is_callable(obj.encode):
return obj.encode('ascii', 'replace')
raise TypeError('Could not convert object type "{}" to bytes!'.format(type(obj)))
def to_unicode(obj):
if isinstance(obj, type(u'')):
return obj
if hasattr(obj, 'decode') and is_callable(obj.decode):
return obj.decode(encoding='utf-8')
raise TypeError('Could not convert object type "{}" to unicode!'.format(type(obj)))
def iter_keys(obj):
if not isinstance(obj, dict):
raise TypeError('Object must be of type dict!')
if is_python3():
return obj.keys()
return obj.iterkeys()
def iter_items(obj):
if not isinstance(obj, dict):
raise TypeError('Object must be of type dict!')
if is_python3():
return obj.items()
return obj.iteritems()
def iter_values(obj):
if not isinstance(obj, dict):
raise TypeError('Object must be of type dict!')
if is_python3():
return obj.values()
return obj.itervalues()
def _py_dict(obj):
if not isinstance(obj, dict):
raise TypeError('Object must be of type dict!')
new_dict = dict()
for name, val in iter_items(obj):
if isinstance(name, type(b'')) and is_python3():
name = to_unicode(name)
elif isinstance(name, type(u'')) and not is_python3():
name = to_bytes(name)
if isinstance(val, dict):
val = _py_dict(val)
elif isinstance(val, type(b'')) and is_python3():
val = to_unicode(val)
elif isinstance(val, type(u'')) and not is_python3():
val = to_bytes(val)
new_dict[name] = val
return deepcopy(new_dict)
def json2dict(obj):
if isinstance(obj, dict):
return deepcopy(_py_dict(obj))
if obj is None:
return dict()
if hasattr(obj, 'strip'):
if obj.strip() == '':
return dict()
else:
return deepcopy(_py_dict(deepcopy(json.loads(obj))))
raise TypeError('Object must be of string type!')
if not is_python3():
range = xrange
else:
range = range
def get_executable_directory():
return path.abspath(path.dirname(argv[0]))
def get_executable_path():
return path.join(get_executable_directory(), argv[0])
|