linux_script/tapo_exporter.py

#!/usr/bin/python3

import argparse
import asyncio
import datetime
import json
import time

from kasa import Discover
from kasa.exceptions import KasaException
from prometheus_client import Gauge, start_http_server
from rich import print as pp

ENABLE_EXPORTER = True
DEFAULT_PORT = 8089
DEFAULT_POLLING_INTERVAL = 60
DEFAULT_SECRET_FILEPATH = "~/.secret"
DEFAULT_CONFIGS = {
    "port": DEFAULT_PORT,
    "polling_interval": DEFAULT_POLLING_INTERVAL,
    "secret_filepath": DEFAULT_SECRET_FILEPATH
}


class TplinkT315:
    def __init__(self, ip, username, password, devName):
        self.ip = ip
        self.username = username
        self.password = password
        self.devName = devName
        self.hubDev = None
        self.t315Dev = None

    async def connect(self):
        try:
            self.hubDev = await Discover.discover_single(self.ip, username=self.username, password=self.password)
            await self.hubDev.update()
        except KasaException as e:
            self.hubDev = None
            pp(f"[TplinkT315::connect] Hub Error: {e}")
            return False

        try:
            self.t315Dev = self.hubDev.get_plug_by_name(self.devName)
            await self.t315Dev.update()
        except KasaException as e:
            self.t315Dev = None
            pp(f"[TplinkT315::connect] Device Error: {e}")
            return False

        return True

    async def getTemperature(self):
        temperatureValue = await self.getProperty("temperature")
        if temperatureValue is None:
            return None

        temperatureValue = float(temperatureValue)
        return temperatureValue

    async def getHumidity(self):
        humidityValue = await self.getProperty("humidity")
        if humidityValue is None:
            return None

        humidityValue = float(humidityValue)
        return humidityValue

    async def getProperty(self, propName):
        if self.t315Dev is None:
            return None

        await self.hubDev.update()
        await self.t315Dev.update()

        featuresOfTempDev = self.t315Dev.features
        if propName not in featuresOfTempDev:
            return None

        return featuresOfTempDev[propName].value

    async def disconnect(self):
        if self.hubDev is None:
            return

        await self.hubDev.disconnect()


def getSecrets(fileanme):
    try:
        with open(fileanme, "r") as f:
            secrets = json.load(f)

        return secrets
    except FileNotFoundError:
        pp(f"File not found: {fileanme}")
        return None


def getDatetime():
    return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")

async def main(configs):
    pp(f"Start: {getDatetime()}")
    secrect = getSecrets(configs.get("secret_filepath", DEFAULT_POLLING_INTERVAL))
    if secrect is None:
        pp(f'Cannot read secret file({configs.get("secret_filepath", DEFAULT_POLLING_INTERVAL)})')
        return

    hubIp = secrect["ip"]
    username = secrect["username"]
    password = secrect["password"]
    devName = "鐵皮屋-溫濕度感測器"

    t315 = TplinkT315(hubIp, username, password, devName)
    if await t315.connect() is False:
        pp(f"Failed to connect to {hubIp}")
        return

    time.sleep(10)
    gaugeDict = {}
    pollingInterval = configs.get("polling_interval", DEFAULT_POLLING_INTERVAL)
    start_http_server(configs.get("port", DEFAULT_PORT)) if ENABLE_EXPORTER else None

    while True:
        try:
            getters = [
                {
                    "name": "temperature",
                    "func": t315.getTemperature,
                },
                {
                    "name": "humidity",
                    "func": t315.getHumidity,
                }
            ]

            for getter in getters:
                name = getter["name"]
                value = await getter["func"]()
                gaugeName = f"tplink_t315_{name}"
                gaugeDesc = f"The {name} of Tplink T315"

                if gaugeName not in gaugeDict:
                    gaugeDict[gaugeName] = Gauge(gaugeName, gaugeDesc)

                if ENABLE_EXPORTER:
                    gaugeDict[gaugeName].set(value)

                pp(f"[{getDatetime()}] Set {gaugeName}: {value}")

            time.sleep(pollingInterval)
        except KeyboardInterrupt:
            pp("\nUser stopped process.")
            break

    await t315.disconnect()
    pp(f"End: {getDatetime()}")


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("-p", "--port", type=int, default=DEFAULT_PORT, help="The port of exportor")
    parser.add_argument("-i", "--polling_interval", type=int, default=DEFAULT_POLLING_INTERVAL, help="Polling interval for collect HDD temperature")
    parser.add_argument("-f", "--secrect_file", default=".secret", help="The file that contains the secrets")
    args = parser.parse_args()
    DEFAULT_CONFIGS["port"] = args.port
    DEFAULT_CONFIGS["polling_interval"] = args.polling_interval
    DEFAULT_CONFIGS["secret_filepath"] = args.secrect_file
    asyncio.run(main(DEFAULT_CONFIGS))