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