diff options
Diffstat (limited to 'python.d/python_modules/third_party')
| -rw-r--r-- | python.d/python_modules/third_party/__init__.py | 0 | ||||
| -rw-r--r-- | python.d/python_modules/third_party/lm_sensors.py | 257 | ||||
| -rw-r--r-- | python.d/python_modules/third_party/ordereddict.py | 128 |
3 files changed, 385 insertions, 0 deletions
diff --git a/python.d/python_modules/third_party/__init__.py b/python.d/python_modules/third_party/__init__.py new file mode 100644 index 000000000..e69de29bb --- /dev/null +++ b/python.d/python_modules/third_party/__init__.py diff --git a/python.d/python_modules/third_party/lm_sensors.py b/python.d/python_modules/third_party/lm_sensors.py new file mode 100644 index 000000000..1d868f0e2 --- /dev/null +++ b/python.d/python_modules/third_party/lm_sensors.py @@ -0,0 +1,257 @@ +""" +@package sensors.py +Python Bindings for libsensors3 + +use the documentation of libsensors for the low level API. +see example.py for high level API usage. + +@author: Pavel Rojtberg (http://www.rojtberg.net) +@see: https://github.com/paroj/sensors.py +@copyright: LGPLv2 (same as libsensors) <http://opensource.org/licenses/LGPL-2.1> +""" + +from ctypes import * +import ctypes.util + +_libc = cdll.LoadLibrary(ctypes.util.find_library("c")) +# see https://github.com/paroj/sensors.py/issues/1 +_libc.free.argtypes = [c_void_p] +_hdl = cdll.LoadLibrary(ctypes.util.find_library("sensors")) + +version = c_char_p.in_dll(_hdl, "libsensors_version").value.decode("ascii") + + +class bus_id(Structure): + _fields_ = [("type", c_short), + ("nr", c_short)] + + +class chip_name(Structure): + _fields_ = [("prefix", c_char_p), + ("bus", bus_id), + ("addr", c_int), + ("path", c_char_p)] + + +class feature(Structure): + _fields_ = [("name", c_char_p), + ("number", c_int), + ("type", c_int)] + + # sensors_feature_type + IN = 0x00 + FAN = 0x01 + TEMP = 0x02 + POWER = 0x03 + ENERGY = 0x04 + CURR = 0x05 + HUMIDITY = 0x06 + MAX_MAIN = 0x7 + VID = 0x10 + INTRUSION = 0x11 + MAX_OTHER = 0x12 + BEEP_ENABLE = 0x18 + + +class subfeature(Structure): + _fields_ = [("name", c_char_p), + ("number", c_int), + ("type", c_int), + ("mapping", c_int), + ("flags", c_uint)] + + +_hdl.sensors_get_detected_chips.restype = POINTER(chip_name) +_hdl.sensors_get_features.restype = POINTER(feature) +_hdl.sensors_get_all_subfeatures.restype = POINTER(subfeature) +_hdl.sensors_get_label.restype = c_void_p # return pointer instead of str so we can free it +_hdl.sensors_get_adapter_name.restype = c_char_p # docs do not say whether to free this or not +_hdl.sensors_strerror.restype = c_char_p + +### RAW API ### +MODE_R = 1 +MODE_W = 2 +COMPUTE_MAPPING = 4 + + +def init(cfg_file=None): + file = _libc.fopen(cfg_file.encode("utf-8"), "r") if cfg_file is not None else None + + if _hdl.sensors_init(file) != 0: + raise Exception("sensors_init failed") + + if file is not None: + _libc.fclose(file) + + +def cleanup(): + _hdl.sensors_cleanup() + + +def parse_chip_name(orig_name): + ret = chip_name() + err = _hdl.sensors_parse_chip_name(orig_name.encode("utf-8"), byref(ret)) + + if err < 0: + raise Exception(strerror(err)) + + return ret + + +def strerror(errnum): + return _hdl.sensors_strerror(errnum).decode("utf-8") + + +def free_chip_name(chip): + _hdl.sensors_free_chip_name(byref(chip)) + + +def get_detected_chips(match, nr): + """ + @return: (chip, next nr to query) + """ + _nr = c_int(nr) + + if match is not None: + match = byref(match) + + chip = _hdl.sensors_get_detected_chips(match, byref(_nr)) + chip = chip.contents if bool(chip) else None + return chip, _nr.value + + +def chip_snprintf_name(chip, buffer_size=200): + """ + @param buffer_size defaults to the size used in the sensors utility + """ + ret = create_string_buffer(buffer_size) + err = _hdl.sensors_snprintf_chip_name(ret, buffer_size, byref(chip)) + + if err < 0: + raise Exception(strerror(err)) + + return ret.value.decode("utf-8") + + +def do_chip_sets(chip): + """ + @attention this function was not tested + """ + err = _hdl.sensors_do_chip_sets(byref(chip)) + if err < 0: + raise Exception(strerror(err)) + + +def get_adapter_name(bus): + return _hdl.sensors_get_adapter_name(byref(bus)).decode("utf-8") + + +def get_features(chip, nr): + """ + @return: (feature, next nr to query) + """ + _nr = c_int(nr) + feature = _hdl.sensors_get_features(byref(chip), byref(_nr)) + feature = feature.contents if bool(feature) else None + return feature, _nr.value + + +def get_label(chip, feature): + ptr = _hdl.sensors_get_label(byref(chip), byref(feature)) + val = cast(ptr, c_char_p).value.decode("utf-8") + _libc.free(ptr) + return val + + +def get_all_subfeatures(chip, feature, nr): + """ + @return: (subfeature, next nr to query) + """ + _nr = c_int(nr) + subfeature = _hdl.sensors_get_all_subfeatures(byref(chip), byref(feature), byref(_nr)) + subfeature = subfeature.contents if bool(subfeature) else None + return subfeature, _nr.value + + +def get_value(chip, subfeature_nr): + val = c_double() + err = _hdl.sensors_get_value(byref(chip), subfeature_nr, byref(val)) + if err < 0: + raise Exception(strerror(err)) + return val.value + + +def set_value(chip, subfeature_nr, value): + """ + @attention this function was not tested + """ + val = c_double(value) + err = _hdl.sensors_set_value(byref(chip), subfeature_nr, byref(val)) + if err < 0: + raise Exception(strerror(err)) + + +### Convenience API ### +class ChipIterator: + def __init__(self, match=None): + self.match = parse_chip_name(match) if match is not None else None + self.nr = 0 + + def __iter__(self): + return self + + def __next__(self): + chip, self.nr = get_detected_chips(self.match, self.nr) + + if chip is None: + raise StopIteration + + return chip + + def __del__(self): + if self.match is not None: + free_chip_name(self.match) + + def next(self): # python2 compability + return self.__next__() + + +class FeatureIterator: + def __init__(self, chip): + self.chip = chip + self.nr = 0 + + def __iter__(self): + return self + + def __next__(self): + feature, self.nr = get_features(self.chip, self.nr) + + if feature is None: + raise StopIteration + + return feature + + def next(self): # python2 compability + return self.__next__() + + +class SubFeatureIterator: + def __init__(self, chip, feature): + self.chip = chip + self.feature = feature + self.nr = 0 + + def __iter__(self): + return self + + def __next__(self): + subfeature, self.nr = get_all_subfeatures(self.chip, self.feature, self.nr) + + if subfeature is None: + raise StopIteration + + return subfeature + + def next(self): # python2 compability + return self.__next__()
\ No newline at end of file diff --git a/python.d/python_modules/third_party/ordereddict.py b/python.d/python_modules/third_party/ordereddict.py new file mode 100644 index 000000000..d0b97d47c --- /dev/null +++ b/python.d/python_modules/third_party/ordereddict.py @@ -0,0 +1,128 @@ +# Copyright (c) 2009 Raymond Hettinger +# +# Permission is hereby granted, free of charge, to any person +# obtaining a copy of this software and associated documentation files +# (the "Software"), to deal in the Software without restriction, +# including without limitation the rights to use, copy, modify, merge, +# publish, distribute, sublicense, and/or sell copies of the Software, +# and to permit persons to whom the Software is furnished to do so, +# subject to the following conditions: +# +# The above copyright notice and this permission notice shall be +# included in all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES +# OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND +# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT +# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, +# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING +# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR +# OTHER DEALINGS IN THE SOFTWARE. + +from UserDict import DictMixin + + +class OrderedDict(dict, DictMixin): + + def __init__(self, *args, **kwds): + if len(args) > 1: + raise TypeError('expected at most 1 arguments, got %d' % len(args)) + try: + self.__end + except AttributeError: + self.clear() + self.update(*args, **kwds) + + def clear(self): + self.__end = end = [] + end += [None, end, end] # sentinel node for doubly linked list + self.__map = {} # key --> [key, prev, next] + dict.clear(self) + + def __setitem__(self, key, value): + if key not in self: + end = self.__end + curr = end[1] + curr[2] = end[1] = self.__map[key] = [key, curr, end] + dict.__setitem__(self, key, value) + + def __delitem__(self, key): + dict.__delitem__(self, key) + key, prev, next = self.__map.pop(key) + prev[2] = next + next[1] = prev + + def __iter__(self): + end = self.__end + curr = end[2] + while curr is not end: + yield curr[0] + curr = curr[2] + + def __reversed__(self): + end = self.__end + curr = end[1] + while curr is not end: + yield curr[0] + curr = curr[1] + + def popitem(self, last=True): + if not self: + raise KeyError('dictionary is empty') + if last: + key = reversed(self).next() + else: + key = iter(self).next() + value = self.pop(key) + return key, value + + def __reduce__(self): + items = [[k, self[k]] for k in self] + tmp = self.__map, self.__end + del self.__map, self.__end + inst_dict = vars(self).copy() + self.__map, self.__end = tmp + if inst_dict: + return self.__class__, (items,), inst_dict + return self.__class__, (items,) + + def keys(self): + return list(self) + + setdefault = DictMixin.setdefault + update = DictMixin.update + pop = DictMixin.pop + values = DictMixin.values + items = DictMixin.items + iterkeys = DictMixin.iterkeys + itervalues = DictMixin.itervalues + iteritems = DictMixin.iteritems + + def __repr__(self): + if not self: + return '%s()' % (self.__class__.__name__,) + return '%s(%r)' % (self.__class__.__name__, self.items()) + + def copy(self): + return self.__class__(self) + + @classmethod + def fromkeys(cls, iterable, value=None): + d = cls() + for key in iterable: + d[key] = value + return d + + def __eq__(self, other): + if isinstance(other, OrderedDict): + if len(self) != len(other): + return False + for p, q in zip(self.items(), other.items()): + if p != q: + return False + return True + return dict.__eq__(self, other) + + def __ne__(self, other): + return not self == other |