Lab Power Monitoring with Kasa and Grafana

I'll add some context soon, but here's the script a friend of mine wrote to poll Kasa smart plugs for energy usage.

from influxdb import InfluxDBClient
import schedule
import time
import json
import random
from datetime import datetime
import asyncio
from kasa import SmartPlug
from kasa import SmartStrip
import sys

plug_ips = [ "10.10.60.25", "10.10.60.27", "10.10.60.28" ]

def sample_plugs():

    result = []
    for ip in plug_ips:

        plug = SmartPlug(ip)
        asyncio.run(plug.update())
        energy_usage_realtime = asyncio.run(plug.current_consumption())
        energy_usage_this_month = plug.emeter_this_month
        energy_usage_today = plug.emeter_today

        print(plug.alias)
        print(energy_usage_realtime)
        print(energy_usage_this_month)
        print(energy_usage_today)

        result.append({ "id":plug.alias, "energy_realtime_watt":energy_usage_realtime, "energy_month_kwh":energy_usage_this_month, "energy_today_kwh":energy_usage_today, "is_on":plug.is_on })
        
    return result

def job():
    sample_time = datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ')

    print("Sampling plugs ...")
    samples = sample_plugs()
    if (len(samples) == 0):
        return
    
    print("Inserting samples into influxdb ...")
    client = InfluxDBClient(host='10.10.10.51', port=8086, username='influxdbusername', password='influxdbpassword', database='databasename', ssl=True, verify_ssl=False)

    # Insert energy usage
    insert_data = []
    for sample in samples:
        insert_data.append({ 
            "measurement":"plug_status", 
            "tags":{ 
                "location":sample["id"] 
            },
            "time": sample_time,
            "fields": {
                "energy_realtime_watt": sample["energy_realtime_watt"],
                "energy_month_kwh": sample["energy_month_kwh"],
                "energy_today_kwh": sample["energy_today_kwh"],
                "is_on": sample["is_on"]
            }
        })

    client.write_points(insert_data, time_precision='ms')



print("Setting up and running continual polling.")
while 1:
    try:
        job()
    except:
        print("Unexpected error:", sys.exc_info()[0])
    time.sleep(5)

Plug_ips is the array of all your smart plugs. You'll also want to change to your influx db credentials on line 43. If you're not using ssl, you'll want to set that to False.

Here's the dockerfile to turn it into a docker container.

FROM python:3-alpine
WORKDIR /usr/src/app
COPY ./poll_energy.py .
RUN python -m pip install --upgrade pip
RUN pip3 install python-kasa schedule influxdb
CMD ["poll_energy.py"]
ENTRYPOINT ["python3"]