airpurifiermiot_stats.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# Plugin installation:
# wget "https://pasternok.org/sources/airpurifiermiot_stats.py?raw" -O /etc/munin/plugins/airpurifiermiot_stats
# chmod +x /etc/munin/plugins/airpurifiermiot_stats
# systemctl restart munin-node

### config
DEVICE = {
    "name": "Mi Air Purifier",
    "ip": "10.1.1.203",
    "token": "00112233445566778899aabbccddeeff",
}

PARAMS = [
    # [0] - label
    # [1] - unit
    # [2] - AirPurifierMiotStatus property name
    # [3] - additional unit conversions (optional)
    # [4] - warning threshold (for filter lifespan)
    # [5] - critical threshold (for filter lifespan)
    ("AQI", "ug/m^3", "aqi"),
    ("average AQI", "ug/m^3", "average_aqi"),
    ("humidity", "%", "humidity"),
    ("temperature", "C", "temperature"),
    ("filter lifespan", "%", "filter_life_remaining", None, 5, 2),
    ("filter use time", "h", "filter_hours_used"),
    ("total use time", "h", "use_time", " / 3600"),
    ("purify volume", "m^3", "purify_volume"),
    ("motor speed", "rpm", "motor_speed"),
]


### code - do not change anything below here
__author__ = "Damian Pasternok <my_forename at pasternok.org>"
__version__ = "0.1.1"
__date__ = "2023/02/27"
__copyright__ = "Copyright (C) 2022 - 2023 Damian Pasternok"
__license__ = "GPL"

import argparse
import socket
from miio import AirPurifierMiot
from miio.exceptions import DeviceException
from time import sleep

# set default timeout for sockets
socket.setdefaulttimeout(10)


class AirPurifier:
    def __init__(self):
        tries = 10
        for i in range(tries):
            try:
                self.data = AirPurifierMiot(
                    DEVICE.get("ip"), DEVICE.get("token")
                ).status()
            except DeviceException as e:
                if i < tries - 1:
                    sleep(2)
                    continue
                else:
                    raise
            else:
                break


class Munin(AirPurifier):
    def __init__(self):
        pass

    def config(self):
        super(Munin, self).__init__()

        out = f"""graph_title {DEVICE["name"]} stats
graph_args --logarithmic --units=si
graph_scale no
graph_category smarthome
graph_vlabel device parameters
"""
        for param in PARAMS:
            out += f"""{param[2]}.label {param[0]} [{param[1]}]\n"""
            try:
                if param[4]:
                    out += f"""{param[2]}.draw LINE2\n"""
                out += f"""{param[2]}.warning {param[4]}:\n"""
                out += f"""{param[2]}.critical {param[5]}:\n"""
            except IndexError:
                pass

        return out.rstrip()

    def data(self):
        super(Munin, self).__init__()

        out = ""

        for param in PARAMS:
            value = eval(f"self.data.{param[2]}")
            try:
                if param[3]:
                    value = int(eval(f"{value}{param[3]}"))
            except IndexError:
                pass
            out += f"""{param[2]}.value {value}\n"""

        return out.rstrip()


def main():
    munin = Munin()

    parser = argparse.ArgumentParser(description=f"""{DEVICE["name"]} Munin Plugin.""")
    parser.add_argument("config", help="print graph configuration", nargs="?")
    args = parser.parse_args()

    if args.config == "config":
        print(munin.config())
    elif not args.config:
        print(munin.data())
    else:
        parser.print_help()


if __name__ == "__main__":
    main()