forked from guillaumeevin/pynonstationarygev
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathroot_utils.py
131 lines (93 loc) · 3.37 KB
/
root_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
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
import datetime
import math
import os.path as op
import subprocess
from itertools import chain
from multiprocessing import Pool
from cached_property import cached_property
VERSION = datetime.datetime.now()
VERSION_TIME = str(VERSION).split('.')[0]
for c in [' ', ':', '-']:
VERSION_TIME = VERSION_TIME.replace(c, '_')
SHORT_VERSION_TIME = VERSION_TIME[8:]
NB_CORES = 7
def batch_nb_cores(iterable, nb_cores=NB_CORES):
batchsize = math.ceil(len(iterable) / nb_cores)
return batch(iterable, batchsize)
def batch(iterable, batchsize=1):
l = len(iterable)
for ndx in range(0, l, batchsize):
yield iterable[ndx:min(ndx + batchsize, l)]
def multiprocessing_batch(function, argument_list, batchsize=None, nb_cores=NB_CORES):
nb_argument = len(argument_list)
if batchsize is None:
batchsize = math.ceil(nb_argument / nb_cores)
with Pool(nb_cores) as p:
result_list = p.map(function, batch(argument_list, batchsize=batchsize))
if None in result_list:
return None
else:
return list(chain.from_iterable(result_list))
def terminal_command(command_str):
return subprocess.check_output(command_str.split()).decode("utf-8").split('\n')
def get_root_path() -> str:
return op.dirname(op.abspath(__file__))
def get_full_path(relative_path: str) -> str:
return op.join(get_root_path(), relative_path)
def get_display_name_from_object_type(object_type):
# assert isinstance(object_type, type), object_type
return str(object_type).split('.')[-1].split("'")[0].split(' ')[0]
def first(s):
"""Return the first element from an ordered collection
or an arbitrary element from an unordered collection.
Raise StopIteration if the collection is empty.
"""
return next(iter(s))
def float_to_str_with_only_some_significant_digits(f, nb_digits) -> str:
assert isinstance(nb_digits, int)
assert nb_digits > 0
return '%s' % float('%.{}g'.format(nb_digits) % f)
def memoize(function):
memo = {}
def wrapper(*args):
if args in memo:
return memo[args]
else:
rv = function(*args)
memo[args] = rv
return rv
return wrapper
class Example(object):
@cached_property
def big_attribute(self):
print('Long loading only once...')
return 2
# https://stackoverflow.com/questions/5189699/how-to-make-a-class-property
# does not enable setter, but follow the link to activate setter option
class ClassPropertyDescriptor(object):
def __init__(self, fget, fset=None):
self.fget = fget
self.fset = fset
def __get__(self, obj, klass=None):
if klass is None:
klass = type(obj)
return self.fget.__get__(obj, klass)()
def __set__(self, obj, value):
if not self.fset:
raise AttributeError("can't set attribute")
type_ = type(obj)
return self.fset.__get__(obj, type_)(value)
def setter(self, func):
if not isinstance(func, (classmethod, staticmethod)):
func = classmethod(func)
self.fset = func
return self
def classproperty(func):
if not isinstance(func, (classmethod, staticmethod)):
func = classmethod(func)
return ClassPropertyDescriptor(func)
if __name__ == '__main__':
e = Example()
print(e.big_attribute)
print(e.big_attribute)
print(VERSION_TIME)