Simple Mi Air Purifier 2 prometheus exporter

main.py

import os

from prometheus_client.core import GaugeMetricFamily, REGISTRY
from miio.airpurifier import AirPurifier
from prometheus_client.metrics_core import CounterMetricFamily
from prometheus_client.twisted import MetricsResource
from twisted.web.server import Site
from twisted.web.resource import Resource
from twisted.internet import reactor


class PurifierCollector(object):
    def __init__(self, ip, token):
        self.device = AirPurifier(ip=ip, token=token)

    def _poll_stats(self):
        return self.device.status()

    def collect(self):
        status = self._poll_stats()
        yield GaugeMetricFamily('aqi', 'Air Quality Index', value=status.aqi)
        yield GaugeMetricFamily('temperature', 'Temperature', value=status.temperature)
        yield GaugeMetricFamily('humidity', 'Humidity', value=status.humidity)
        yield GaugeMetricFamily('motor_speed', 'Motor 1 speed', value=status.motor_speed)
        yield GaugeMetricFamily('motor2_speed', 'Motor 2 speed', value=status.motor2_speed)
        yield CounterMetricFamily('filter_hours_used', 'Filter time running',
                                  value=status.filter_hours_used)
        is_running = (1 if status.power == "on" else 0)
        up_status = GaugeMetricFamily('power', 'Is alive', labels=['power'])
        up_status.add_metric(["on"], is_running)
        up_status.add_metric(["off"], 1 ^ is_running)
        yield up_status


if __name__ == '__main__':
    # Start up the server to expose the metrics.
    device_ip = os.environ.get('DEVICE_IP')
    device_token = os.environ.get('DEVICE_TOKEN')
    pc = PurifierCollector(device_ip, device_token)
    REGISTRY.register(pc)
    root = Resource()
    root.putChild(b'metrics', MetricsResource())

    factory = Site(root)
    reactor.listenTCP(8000, factory)
    reactor.run()