# -*- coding: utf-8 -*- # Description: # Author: Ilya Mashchenko (l2isbad) import os PATH = os.getenv('PATH', '/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin').split(':') CHART_BEGIN = 'BEGIN {0} {1}\n' CHART_CREATE = "CHART {0} '{1}' '{2}' '{3}' '{4}' '{5}' {6} {7} {8}\n" DIMENSION_CREATE = "DIMENSION '{0}' '{1}' {2} {3} {4} '{5}'\n" DIMENSION_SET = "SET '{0}' = {1}\n" def setdefault_values(config, base_dict): for key, value in base_dict.items(): config.setdefault(key, value) return config def run_and_exit(func): def wrapper(*args, **kwargs): func(*args, **kwargs) exit(1) return wrapper def on_try_except_finally(on_except=(None, ), on_finally=(None, )): except_func = on_except[0] finally_func = on_finally[0] def decorator(func): def wrapper(*args, **kwargs): try: func(*args, **kwargs) except Exception: if except_func: except_func(*on_except[1:]) finally: if finally_func: finally_func(*on_finally[1:]) return wrapper return decorator def static_vars(**kwargs): def decorate(func): for k in kwargs: setattr(func, k, kwargs[k]) return func return decorate @on_try_except_finally(on_except=(exit, 1)) def safe_print(*msg): """ :param msg: :return: """ print(''.join(msg)) def find_binary(binary): """ :param binary: :return: """ for directory in PATH: binary_name = '/'.join([directory, binary]) if os.path.isfile(binary_name) and os.access(binary_name, os.X_OK): return binary_name return None def read_last_line(f): with open(f, 'rb') as opened: opened.seek(-2, 2) while opened.read(1) != b'\n': opened.seek(-2, 1) if opened.tell() == 0: break result = opened.readline() return result.decode() class OldVersionCompatibility: def __init__(self): self._data_stream = str() def begin(self, type_id, microseconds=0): """ :param type_id: :param microseconds: or : must be a digit :return: """ self._data_stream += CHART_BEGIN.format(type_id, microseconds) def set(self, dim_id, value): """ :param dim_id: :param value: or : must be a digit :return: """ self._data_stream += DIMENSION_SET.format(dim_id, value) def end(self): self._data_stream += 'END\n' def chart(self, type_id, name='', title='', units='', family='', category='', chart_type='line', priority='', update_every=''): """ :param type_id: :param name: :param title: :param units: :param family: :param category: :param chart_type: :param priority: or :param update_every: or :return: """ self._data_stream += CHART_CREATE.format(type_id, name, title, units, family, category, chart_type, priority, update_every) def dimension(self, dim_id, name=None, algorithm="absolute", multiplier=1, divisor=1, hidden=False): """ :param dim_id: :param name: or None :param algorithm: :param multiplier: or : must be a digit :param divisor: or : must be a digit :param hidden: : literally "hidden" or "" :return: """ self._data_stream += DIMENSION_CREATE.format(dim_id, name or dim_id, algorithm, multiplier, divisor, hidden or str()) @on_try_except_finally(on_except=(exit, 1)) def commit(self): print(self._data_stream) self._data_stream = str()