pyovo.py

"""
Python 3.6 library to interact with OVO Energy's REST API. This isn't an officially supported API
so this library may or may not continue to work. YMMV, don't blame me if it breaks etc etc.

Tom Oinn, 22nd May 2018
"""

import json
from datetime import datetime, timedelta
from http import HTTPStatus

from requests import Session


class NoSessionError(Exception):
    pass


GAS_VOLUME_CORRECTION_FACTOR = 1.02264
GAS_CALORIFIC_VALUE = 39.3


def gas_m3_to_kwh(gas_volume):
    """
    Uses OVO's volume correction and calorific value values to convert a number of M3 of
    gas into a kWh figure. Not currently used as OVO's consumption figures are already costed

    :param gas_volume:
        volume of gas in M3
    :return:
        energy in kWh
    """
    return gas_volume * GAS_CALORIFIC_VALUE * GAS_VOLUME_CORRECTION_FACTOR / 3.60


class OVOSession():
    """
    Login session to OVO's smart meter API
    """

    HEADERS = {
        'content-type': 'application/json;charset=UTF-8',
        'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/62.0.3202.94 Safari/537.36'
    }

    DATE_FORMAT = '%Y-%m-%d'

    TIME_FORMAT = '%Y-%m-%dT%H:%M:%SZ'

    GRANULARITY_DAY = 'DAY'
    GRANULARITY_HALF_HOUR = 'HH'
    GRANULARITY_MONTH = 'MONTH'

    def __init__(self, gas_volume_correction_factor=1.02264, gas_calorific_value=40.0):
        self.session = None
        self.accounts_data = None

    def __check_auth(self):
        if not self.authenticated:
            raise NoSessionError()

    def login(self, user, password):
        """
        Authenticate with OVO

        :param user:
            the user name
        :param password:
            unencrypted password
        :return:
            the response from the authentication call
        """
        login_url = 'https://my.ovoenergy.com/api/auth/login'
        login_postdata = {'username': user, 'password': password, 'rememberMe': 'true'}
        self.session = Session()
        login_response = self.session.post(login_url, data=json.dumps(login_postdata), headers=OVOSession.HEADERS)
        accounts_response = self.session.get(url='https://paym.ovoenergy.com/api/paym/accounts',
                                             headers=OVOSession.HEADERS)
        self.accounts_data = json.loads(accounts_response.content)[0]
        return login_response

    @property
    def authenticated(self):
        return self.session is not None

    @property
    def utilities(self):
        self.__check_auth()
        contracts = {c['consumerId']: {'contract_id': c['id'],
                                       'kwh_rate_gbp': c['rates']['amount']['amount'],
                                       'standing_charge_daily_gbp': c['standingCharge']['amount']['amount'],
                                       'start': datetime.strptime(c['startDate'], OVOSession.DATE_FORMAT),
                                       'end': datetime.strptime(c['expiryDate'], OVOSession.DATE_FORMAT),
                                       'plan_name': c['plan']['name']} for c in
                     self.accounts_data['contracts']}
        consumers = {c['id']: {'utility': c['utilityType'],
                               'mpan': c['mpan'],
                               'meter_serial': c['meters'][0]['meterSerialNumber'],
                               'meter_unit': c['meters'][0]['unitOfMeasure']} for c in
                     self.accounts_data['consumers']}
        return list({**contracts[c], **consumers[c], **{'consumer_id': c}} for c in contracts)

    @property
    def live_load(self):
        """
        Attempts to get live load readings for each utility associated with this account.

        :return:
            List of all utility objects that have live readings, augmented with those readings.
        """
        self.__check_auth()

        def live():
            for u in self.utilities:
                url = 'https://live.ovoenergy.com/api/live/meters/{}/consumptions/instant'.format(u['mpan'])
                response = self.session.get(url, headers=OVOSession.HEADERS)
                if response.status_code == HTTPStatus.OK:
                    data = json.loads(response.content)
                    yield {**u, **{'demand': data['consumption']['demand'],
                                   'demand_unit': data['measurementUnit'],
                                   'demand_cost_per_hour': data['consumption']['consumptionPrice']['amount']}}

        return list(live())

    @property
    def usage_yesterday(self):
        """
        Alias to the no-argument form of usage() which defaults to showing all utilities on a half-hour granularity for
        the previous day.
        """
        return self.usage()

    def usage(self, start_date=None, end_date=None, granularity=GRANULARITY_HALF_HOUR):
        """
        Get historical usage for each utility.

        :param datetime.datetime start_date:
            The start date from which to read values. If not specified, this defaults to being 23:00 two days ago, this
            results in the previous day's readings being retrieved. This value is exclusive, only readings which start
            strictly after this time point will be returned.
        :param datetime.datetime end_date:
            The end date up to which to read values. If not specified this will be exactly 24 hours after the start
            date, so if you don't specify either value you'll end up with the previous day's readinngs. This value is
            inclusive, any readings which do not start strictly after this point will be included.
        :param granularity:
            One of 'HH', 'DAY', or 'MONTH' to determine the bin size of the returned values
        :return:
            Augmented utility structs, containing a 'history' key, which in turn contains a list of dicts with
            'consumption' and 'start' keys, where 'consumption' indicates an amount of energy consumed in kWh (even for
            gas) and 'start' is a datetime indicating the start of the measured period.
        """
        if start_date is None:
            start_date = datetime.today().replace(hour=23, minute=0, second=0, microsecond=0) - timedelta(days=2)
        if end_date is None:
            end_date = start_date + timedelta(days=1)

        def historical_usage():
            for u in self.utilities:
                url = 'https://live.ovoenergy.com/api/live/meters/' \
                      '{}/consumptions/aggregated?from={}&to={}&granularity={}' \
                    .format(u['mpan'], datetime.strftime(start_date, '%Y-%m-%dT%H:%M:%SZ'),
                            datetime.strftime(end_date, '%Y-%m-%dT%H:%M:%SZ'), granularity)
                response = self.session.get(url, headers=OVOSession.HEADERS)
                if response.status_code == HTTPStatus.OK:
                    data = json.loads(response.content)
                    history = list(
                        {'consumption': c['consumption'], 'start': datetime.fromtimestamp(int(c['startTime']) / 1000)}
                        for c in
                        data['consumptions'] if 'dataError' not in c)
                    yield {**u, **{'history': history}}

        return list(historical_usage())