emporia-reader/main.py

# This script will find all devices in your installation and
# output their per-minute usage for the previous DAYS_BACK days.
# Per-minute data is stored for a maximum of 7 days.
#
# To change the timing, update the DAYS_BACK variable below.
#
# Reads credentials from `keys.json`
#
# To ensure data is not read for a minute that has not yet ended,
# this will read up to 2 minutes ago.

import pyemvue
import json
import time
import sys

from pyemvue.enums import Scale, Unit
from pyemvue import PyEmVue
from pprint import pprint
from datetime import datetime, timezone, timedelta

NOW = datetime.now(timezone.utc) - timedelta(minutes=2)
DAYS_BACK = 14 # Number of days in the past to get per-minute data for.
MAX_HOURS = 12 # The maxinum number of hours that can be retrieved at once.
MAX_ATTEMPTS = 5

def eprint(*args, **kwargs):
    print(*args, file=sys.stderr, **kwargs)

# Get & display minutely data for all devices.
def allUsageOverTime(vue: PyEmVue):
	devices = vue.get_devices()
	print(f'time\tdevice_gid\tchannel_num\twH\twatts\tmultiplier\tname')

	for device in devices:
		# print('=== Device ===')
		# pprint(vars(device))
		for channel in device.channels:
			# pprint(vars(channel))
			if not channel.name is None:
				channelUsageOverTime(vue, channel)

def channelUsageOverTime(vue: PyEmVue, c):
	start_time = NOW - timedelta(days=DAYS_BACK)
	end_time = start_time + timedelta(hours=MAX_HOURS)
	while start_time < NOW:
		usage_over_time, start = getChartUsage(
				vue, c,
				start_time, end_time
			)
		time = start
		for kwh in usage_over_time:
			if kwh is None:
				continue
			wattHours = kwh * 1000
			watts = wattHours * 60
			print(f'{time}\t{c.device_gid}\t{c.channel_num}\t{wattHours}\t{watts}\t{c.channel_multiplier}\t{c.name}')
			time += timedelta(minutes=1)
		
		# Jump to the next start time.
		start_time = end_time
		end_time += timedelta(minutes=(MAX_HOURS*60))

def getChartUsage(vue: PyEmVue, channel, start_time, end_time):
	"""
	Get per-minute usage for the channel, with up to MAX_ATTEMPTS
	attempts to account for errors.
	"""
	attempt = 0
	while True:
		attempt += 1
		try:
			return vue.get_chart_usage(
					channel,
					start_time, end_time,
					scale=Scale.MINUTE.value,
					unit=Unit.KWH.value
				)
		except Exception as e:
			if attempt < MAX_ATTEMPTS:
				eprint(
					f'Attempt {attempt} for channel {channel.name}',
					f'from {start_time} to {end_time}',
					f'failed:', e)
				time.sleep(5)
			else:
				eprint(f'Failed after {attempt} attempts')
				raise

if __name__ == '__main__':
	with open('keys.json') as f:
		keys = json.load(f)
	vue = pyemvue.PyEmVue()

	if 'id_token' in keys:
		# Login using access tokens.
		vue.login(
				id_token=keys['id_token'],
				access_token=keys['access_token'],
				refresh_token=keys['refresh_token'],
				token_storage_file='keys.json'
			)
	else:
		# Login with username & password, and update keys.json.
		vue.login(
				username=keys['username'],
				password=keys['password'],
				token_storage_file='keys.json'
			)
	
	allUsageOverTime(vue)