Rheem Water Heater Energy Usage Calculator

rheem_usage.py

import asyncio
import argparse
from collections import defaultdict
from datetime import date, timedelta
import logging
from typing import Dict
from pprint import pprint

from pyeconet import EcoNetApiInterface
from pyeconet.equipment import EquipmentType
from pyeconet.equipment.water_heater import UsageFormat


_client_id = 'c511070a-1ae2-4d0a-b378-1eebd786d824.apps.rheemapi.com'
_client_secret = 'MjQ3Mjc5MjEtYTNjMC00NDJiLTg2YjUtM2NiZWRhZDU4NjMx'

# TODO: get this from PGE
# prices are NEM + generation (nem is refundable with solar)
# EV2-A
_PGE_PEAK = 0.43 + 0.11107  # 0.671
_PGE_PARTIAL_PEAK = 0.42901 + 0.15180  # 0.601
_PGE_OFF_PEAK = 0.24699 + 0.19606  # 0.443


def _get_pge_rate(hour: int):
    if 4 <= hour < 9:
        return _PGE_PEAK

    if 3 <= hour < 4 or 9 <= hour < 24:
        return _PGE_PARTIAL_PEAK

    return _PGE_OFF_PEAK


async def main():
    """
    To use you must install pip3 install https://pypi.org/project/pyeconet/ and have the patches from my PR:

    """
    logging.basicConfig()
    logging.getLogger().setLevel(logging.INFO)

    parser = argparse.ArgumentParser()
    parser.add_argument('-email', required=True, help="Rheem account email address")
    parser.add_argument('-password', required=True, help="Rheem account password")
    # TODO: better support for 1 month delta
    parser.add_argument('-start', type=date.fromisoformat, default=date.today() - timedelta(days=30), help='ISO Start date')
    parser.add_argument('-end', type=date.fromisoformat, default=date.today(), help='ISO Start date')
    app_args = parser.parse_args()

    api = await EcoNetApiInterface.login(app_args.email, password=app_args.password)
    all_equipment = await api.get_equipment_by_type([EquipmentType.WATER_HEATER, EquipmentType.THERMOSTAT])

    kwh_by_hour: Dict[int, float] = defaultdict(float)
    kwh_by_day: Dict[date, float] = defaultdict(float)

    for equip_list in all_equipment.values():
        for equipment in equip_list:
            print(f"Name: {equipment.device_name}")
            day: date = app_args.start
            while day <= app_args.end:
                print(f"Getting usage for {day} - {day + timedelta(days=1)}")
                await equipment.get_energy_usage(UsageFormat.DAILY, year=day.year, month=day.month, period=day.day)

                for hour, amount in equipment.historical_energy_usage.items():
                    kwh_by_hour[hour] += amount
                    kwh_by_day[day - timedelta(days=1)] += amount

                for hour, amount in equipment.energy_usage.items():
                    kwh_by_hour[hour] += amount
                    kwh_by_day[day] += amount

                day += timedelta(days=2)

            print("kwh/$ by hour:")
            pprint({hour: f'{amount} KWH (${amount * _get_pge_rate(hour)})' for hour, amount in kwh_by_hour.items()}, width=200)

            print("kwh by day:")
            pprint(kwh_by_day, width=200)

            total_cost = 0
            for hour, amount in kwh_by_hour.items():
                total_cost += _get_pge_rate(hour) * amount

            print(f"total cost: ${total_cost}")

if __name__ == "__main__":
    asyncio.run(main())