Add Tplink T315 temperature and humidity exporter

tapo_exporter.py

@@ -0,0 +1,162 @@
+#!/usr/bin/python3
+
+import argparse
+import asyncio
+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
+            print(f"[TplinkT315::connect] 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
+            print(f"[TplinkT315::connect] 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.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:
+        print(f"File not found: {fileanme}")
+        return None
+
+
+async def main(configs):
+    secrect = getSecrets(configs.get("secret_filepath", DEFAULT_POLLING_INTERVAL))
+    if secrect is None:
+        return
+
+    hubIp = secrect["ip"]
+    username = secrect["username"]
+    password = secrect["password"]
+    devName = "鐵皮屋-溫濕度感測器"
+
+    t315 = TplinkT315(hubIp, username, password, devName)
+    if await t315.connect() is False:
+        print(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"Set {gaugeName}: {value}")
+
+            time.sleep(pollingInterval)
+        except KeyboardInterrupt:
+            print("\nUser stopped process.")
+            break
+
+    await t315.disconnect()
+
+
+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))