How to Measure Daily Electrical Usage with OpenCV

Using Hexing kWh Prepaid Meter PLN

Hexing PLN Prepaid Meter while the red LED lamp is blinking

Hexing Prepaid Meter is owned by PLN. This tool to replace the analogue meter which must be noted by the operator manually. With a prepaid meter now, I can see a more accurate measurement from KWH, ampere, voltage, and my credit.

This article will be an example to read the measurement from a prepaid meter with OpenCV library. Currently, I use the raspberry pi 4 to run my code and use the camera sized 4 x 3.1 x 2.5cm. If you want to extend the USB cable, I recommend setup a maximum USB cable length of 3 meters. More than the length of 3 meters will lose the USB signal and makes the camera cannot be connected.

Face the camera to prepaid meter impulse LED light

After you set up the camera like the above picture, you ready to measure it.

Let’s code !

kwh_pln_sensor.py

I use this code for real-time monitoring the impulse LED lamp

First, import required libraries

from sqlalchemy import create_engine
import cv2
import numpy as np
import pandas as pd
import datetime
import time

• sqlalchemy is a helper to connect database (Object Relational Mapper), in this case, PostgreSQL
• cv2 (Open Source Computer Vision Library or called OpenCV) the python library to help pixel colour filters from the camera, which filters the red colour
• pandas , make it easier to create a matrix column and row to save the data into the database
• datetime and time , libraries to manage date and time format

If you found the error ModuleNotFoundError, you should install the library first

pip install opencv-python
pip install sqlalchemy
pip install pandas

Create the engine to connect the database

conn = 
create_engine("postgresql+psycopg2://root:12345678@localhost:5432/plnstats").connect()

I create the database named plnstats with username root and password 12345678 on localhost server port 5432

Initialize camera and capture it.

cap = cv2.VideoCapture(0)
(ret, frame) = cap.read()

cv2.VideoCapture(0), you can change the index from 0 to 1, 2, …, etc depending on your computer video devices index.

cap.read() to get the status frame read correctly and image frame from the camera

while True:
(ret, frame) = cap.read()

size = frame.size

image = cv2.cvtColor(frame, cv2.COLOR_BGR2HSV)

Convert original frame to RGB colour, image = cv2.cvtColor(frame, cv2.COLOR_BGR2HSV)

Filter lower and upper range on colour wheel

To use the mask filter, set the lower and upper range RGB, I try and error to determine constant RGB and apply with code

lower = np.array([168, 25, 0])
upper = np.array([179, 255, 255])
mask = cv2.inRange(image, lower, upper)

Calculate the percentage red colour divide with non-red colour

no_red = cv2.countNonZero(mask)
frac_red = np.divide(float(no_red), size)
percent_red = np.multiply((float(frac_red)), 100)

Set the threshold, depending on the camera, I use 10.0 threshold for this

if (percent_red >= 10.0):

Save to pandas dataframe

data_capture = {
'color_percentage': percent_red,
'created_on': datetime.datetime.now()
}
df = pd.DataFrame(columns=['color_percentage','created_on'],data=data_capture,index=[0])

And save into the database

df.to_sql('home_pln_kwh_sensor', schema='public', con=conn, if_exists='append',index=False)

This is full code for kwh_pln_sensor.py

you can remove the code on line 31. The statement will help you check the camera is successfully captured the LED blink

when print(percent_red) executed

And then how to summarise the sensor capture?

I want to extract information about hour usage and credit usage per day. After that, I can estimate my usage at home and can be noticed before my credit is empty (and dark like stone age)

kwh_pln_calculate.py

Like before, import the required libraries

import pandas as pd
from datetime import datetime,timedelta
from sqlalchemy import create_engine

I used timedelta to get yesterday day (and you can be used to calculate a different time too)

create the constants. IMPULSE_KWH = 1000 means for every 1000 LED blink means 1 KWH, and RUPIAH_PER_KWH = 1444.70 means price for electricity home type R-1/TR 1.301–2.200 VA is 1444.70 rupiah

IMPULSE_KWH = 1000 
RUPIAH_PER_KWH = 1444.70

This day value and what the yesterday value, format it with year-month-day..

now_date = datetime.now().strftime('%Y-%m-%d')
yesterday_date = datetime.now() - timedelta(days=1)
yesterday_date = yesterday_date.strftime('%Y-%m-%d')

Connect to PostgreSQL database

conn = create_engine("postgresql+psycopg2://root:12345678@localhost:5432/plnstats").connect()

I want to summarise the yesterday kWh usage

df = pd.read_sql(sql='SELECT color_percentage, created_on from public.home_pln_kwh_sensor where created_on >= \'{yesterday} 00:00:00\' and created_on < \'{currdate} 00:00:00\''.format(yesterday=yesterday_date,currdate=now_date),con=conn)

Convert created_on column into pandas datetime

df['created_on'] = pd.to_datetime(df.created_on)
df['created_on'] = df['created_on'].dt.strftime('%Y-%m-%d %H:%M:%S')

normalize the data, which one second only one LED blink

df = df.drop_duplicates(subset=['created_on'])

Make another column hour and day for more explanations

df['created_on'] = pd.to_datetime(df.created_on)
df['hour'] = df['created_on'].dt.strftime('%H')
df['day'] =  df['created_on'].dt.strftime('%Y-%m-%d')

Create a list of day to examine the credit usage per day

list_day = list(df['day'].unique())

Make a new dataframe to summarise

df_summary = pd.DataFrame(columns=['day','hour','sum_impulse','charges_amount'])

Calculate it

for dday in list_day:
list_hour = list(df[df['day'] == str(dday)]['hour'].unique())
for y in list_hour:
wSumKWH = df[(df['day'] == str(dday)) & (df['hour'] == str(y))]
df_summary = df_summary.append({'hour':y,'day':dday,'sum_impulse':wSumKWH.groupby('hour').count()['day'][0],'charges_amount':(wSumKWH.groupby('hour').count()['day'][0])*RUPIAH_PER_KWH/IMPULSE_KWH},ignore_index=True)
df_summary.to_sql('home_pln_kwh_hour', schema='public', con=conn, if_exists='append', index=False)
credit_summary = df_summary[df_summary['day'] == yesterday_date].groupby('day').sum()
credit_summary.to_sql('home_pln_kwh_summary', schema='public', con=conn, if_exists='append', index=True)

And you can query the database for the summary or you can extract from dataframe like this

Left image is groupby per hour from sensor, Right image is summary after calculate with kWh * price per kWh

Then, you can see the highest price per hour.

This is full code for kwh_pln_calculate.py

Thank you for reading, hopefully, you can use it to measure the credit electricity per month :)