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
|
import time
import threading
from loggers import debuglog
def log_method(fun):
def wrapper(self, *args, **kwargs):
s_args = s_kwargs = ''
if args: s_args = ','.join(str(a) for a in args)
if kwargs: s_kwargs = ','.join('{}={}'.format(k, v)
for k, v in kwargs.items())
debuglog.debug('CALL: {}.{}({},{})'.format(self.__class__.__name__,
fun.__name__,
s_args, s_kwargs))
result = fun(self, *args, **kwargs)
return result
return wrapper
def log_function(fun):
def wrapper(*args, **kwargs):
s_args = s_kwargs = ''
if args: s_args = ','.join(str(a) for a in args)
if kwargs: s_kwargs = ','.join('{}={}'.format(k, v)
for k, v in kwargs.items())
debuglog.debug('CALL: {}({},{})'.format(fun.__name__,
s_args, s_kwargs))
result = fun(*args, **kwargs)
return result
return wrapper
class Schedule:
"""Class to schedule repeatable events"""
def __init__(self, start, delay, func, *args, **kwargs):
self._active = True
self._next_event = time.time() + start
self._delay = delay
self._func = func
self._thread = threading.Thread(target=self._threadfunc,
args=args, kwargs=kwargs)
self._thread.start()
def _threadfunc(self, *args, **kwargs):
while self._active:
now = time.time()
if now >= self._next_event:
self._func(*args, **kwargs)
self._next_event += self._delay
if self._delay <= 0:
return
time.sleep(0.1)
def cancel(self):
if self._active:
self._active = False
if self._thread is not threading.current_thread():
self._thread.join()
# The following group of functions is to provide a way to extend
# a module's functions by decorating them with @extendable
_extensions = {}
def register_extension(name, func):
if name in _extensions:
if func not in _extensions[name]:
_extensions[name].append(func)
else:
_extensions[name] = [func]
def extendable(func):
"""
Decorator function. If a function is decorated with
@extendable, each call of it will be followed by calls of
_extentions[fun.__name__](*args, **kwargs) with same args
"""
def wrapper(*args, **kwargs):
name = func.__name__
func(*args, **kwargs)
if name in _extensions:
for f in _extensions[name]:
f(*args, **kwargs)
wrapper.__name__ = func.__name__
return wrapper
def extends(func_name):
"""
Return a decorator that registers the wrapped function
as an extention to func_name call.
See @extendable.
"""
def decorator(func):
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
register_extension(func_name, func)
return wrapper
return decorator
def preprocess_argument(pp_fun, arg=0):
"""
Make a decorator that modifies 1 argument of a given function
before calling it.
pp_fun accepts this argument and returns the modified.
arg can be either int, then args[arg] is modified, or it
can be str, then kwargs[arg] is modified
"""
def decorator(fun):
def wrapper(*args, **kwargs):
if isinstance(arg, int):
if arg < len(args):
args = list(args)
args[arg] = pp_fun(args[arg])
elif isinstance(arg, str):
if arg in kwargs:
kwargs[arg] = pp_fun(kwargs[arg])
return fun(*args, **kwargs)
return wrapper
return decorator
# Encode string - used with 4144 shop compatibility.
def encode_str(value, size):
output = ''
base = 94
start = 33
while value:
output += chr(value % base + start)
value /= base
while len(output) < size:
output += chr(start)
return output
|
