mqtt-sensord/mqtt_sensord/sensorlib/base.py

try:
    import json
except ImportError:
    import ujson as json


def sensor(cls):
    Sensor.register(cls)

    return cls


class Sensor(object):
    hostname = None
    platform = None
    _sensor_classes = {}
    _mqtt_callbacks = {}
    HAS_SENSOR_DATA = True

    def __init__(self, id, name, type, pin, description, **sensor_conf):
        self.id = id
        self.name = name
        self.type = type
        self.pin = pin
        self.description = description

    @classmethod
    def configure(cls, hostname, platform):
        cls.hostname = hostname
        cls.platform = platform

    @classmethod
    def register(cls, new_cls):
        cls._sensor_classes[new_cls.sensor_class] = new_cls

    @classmethod
    def make_sensor(cls, **kwargs):
        sensor_cls = cls._sensor_classes[kwargs['type']]
        return sensor_cls(**kwargs)

    def get_data(self):
        raise NotImplementedError("You missed a spot!")

    def gen_datapoint(self, legacy=True):
        return {
            'type': 'measurement',
            'tags': {
                'name': self.name,
                'sub-id': self.id,
                'sensor': self.type,
                'hostname': self.hostname,
                'platform': self.platform,
            },
            'fields': self.get_data()
        }

    def register_callbacks(self):
        pass

    def get_mqtt_topics(self):
        return []

    @classmethod
    def register_callback(cls, mqtt_topic, callback):
        cls._mqtt_callbacks[mqtt_topic] = callback

    @classmethod
    def run_mqtt_callback(cls, mqtt_topic, data):
        try:
            data = json.loads(data)
        except ValueError:
            print("Could not parse data {}".format(data))

        if mqtt_topic in cls._mqtt_callbacks:
            cls._mqtt_callbacks[mqtt_topic](data)
        else:
            print("Unknown callback for", mqtt_topic)

    def get_topics(self):
        return []